public inbox for [email protected]
help / color / mirror / Atom feedFrom: Chao Li <[email protected]>
To: Ashutosh Bapat <[email protected]>
Cc: Álvaro Herrera <[email protected]>
Cc: Postgres hackers <[email protected]>
Subject: Re: Refactor replication origin state reset helpers
Date: Tue, 30 Dec 2025 15:17:26 +0800
Message-ID: <CAEoWx2=7xcxQPJgnPfWsKj2SzCNiBaxOXa1u9fSKME1Yd=r=fQ@mail.gmail.com> (raw)
In-Reply-To: <CAEoWx2mT+gu8H-KhaeNodk4T7csX-_wwqRew=-t3zo6b6jXfaQ@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>
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.
Best regards,
--
Chao Li (Evan)
HighGo Software Co., Ltd.
https://www.highgo.com/
Attachments:
[application/octet-stream] v4-0002-Consolidate-replication-origin-session-globals-in.patch (20.6K, ../CAEoWx2=7xcxQPJgnPfWsKj2SzCNiBaxOXa1u9fSKME1Yd=r=fQ@mail.gmail.com/3-v4-0002-Consolidate-replication-origin-session-globals-in.patch)
download | inline diff:
From 39cdd382fe483c5e5576fc02f7a5ed7928293fa4 Mon Sep 17 00:00:00 2001
From: "Chao Li (Evan)" <[email protected]>
Date: Tue, 30 Dec 2025 12:55:31 +0800
Subject: [PATCH v4 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
RepOriginSessionState 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_session_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]>
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 | 22 ++++++------
src/backend/replication/logical/tablesync.c | 4 +--
src/backend/replication/logical/worker.c | 28 +++++++--------
src/include/replication/origin.h | 11 ++++--
src/tools/pgindent/typedefs.list | 1 +
9 files changed, 76 insertions(+), 70 deletions(-)
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 3bc85986829..7d34692f05f 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_session_state.origin != InvalidRepOriginId &&
+ replorigin_session_state.origin != DoNotReplicateId);
if (replorigin)
{
- hdr->origin_lsn = replorigin_session_origin_lsn;
- hdr->origin_timestamp = replorigin_session_origin_timestamp;
+ hdr->origin_lsn = replorigin_session_state.origin_lsn;
+ hdr->origin_timestamp = replorigin_session_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_session_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_session_state.origin != InvalidRepOriginId &&
+ replorigin_session_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_session_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_session_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_session_state.origin_timestamp == 0)
+ replorigin_session_state.origin_timestamp = committs;
TransactionTreeSetCommitTsData(xid, nchildren, children,
- replorigin_session_origin_timestamp,
- replorigin_session_origin);
+ replorigin_session_state.origin_timestamp,
+ replorigin_session_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_session_state.origin != InvalidRepOriginId &&
+ replorigin_session_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_session_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 1b5c1f6b763..5e7122b23f3 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_session_state.origin != InvalidRepOriginId &&
+ replorigin_session_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_session_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_session_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_session_state.origin_timestamp == 0)
+ replorigin_session_state.origin_timestamp = GetCurrentTransactionStopTimestamp();
TransactionTreeSetCommitTsData(xid, nchildren, children,
- replorigin_session_origin_timestamp,
- replorigin_session_origin);
+ replorigin_session_state.origin_timestamp,
+ replorigin_session_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_session_state.origin != InvalidRepOriginId &&
+ replorigin_session_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_session_state.origin_lsn,
XactLastRecEnd);
/*
@@ -5927,12 +5927,12 @@ XactLogCommitRecord(TimestampTz commit_time,
}
/* dump transaction origin information */
- if (replorigin_session_origin != InvalidRepOriginId)
+ if (replorigin_session_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_session_state.origin_lsn;
+ xl_origin.origin_timestamp = replorigin_session_state.origin_timestamp;
}
if (xl_xinfo.xinfo != 0)
@@ -6080,12 +6080,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_session_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_session_state.origin_lsn;
+ xl_origin.origin_timestamp = replorigin_session_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 a56d5a55282..ed576ae578b 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_session_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_session_state.origin, sizeof(replorigin_session_state.origin));
+ scratch += sizeof(replorigin_session_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 a4aafcf5b6e..15c600faa64 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_session_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_session_state.origin_lsn = abort_data->abort_lsn;
+ replorigin_session_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 fc9e14ecbc5..5324da2719f 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -160,9 +160,9 @@ 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;
+RepOriginSessionState replorigin_session_state = {
+ InvalidRepOriginId, InvalidXLogRecPtr, 0
+};
/*
* Base address into a shared memory array of replication states of size
@@ -896,7 +896,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_session_state.origin_lsn and replorigin_session_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).
*
@@ -1295,10 +1295,10 @@ replorigin_session_get_progress(bool flush)
void
replorigin_session_clear_state(bool xact_only)
{
- replorigin_session_origin_lsn = InvalidXLogRecPtr;
- replorigin_session_origin_timestamp = 0;
+ replorigin_session_state.origin_lsn = InvalidXLogRecPtr;
+ replorigin_session_state.origin_timestamp = 0;
if (!xact_only)
- replorigin_session_origin = InvalidRepOriginId;
+ replorigin_session_state.origin = InvalidRepOriginId;
}
@@ -1408,7 +1408,7 @@ pg_replication_origin_session_setup(PG_FUNCTION_ARGS)
pid = PG_GETARG_INT32(1);
replorigin_session_setup(origin, pid);
- replorigin_session_origin = origin;
+ replorigin_session_state.origin = origin;
pfree(name);
@@ -1438,7 +1438,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_session_state.origin != InvalidRepOriginId);
}
@@ -1482,8 +1482,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_session_state.origin_lsn = location;
+ replorigin_session_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 47104ed676c..61952e7d995 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_session_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_session_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 4df177664b7..135c65e56bd 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_session_state.origin_lsn = prepare_data->end_lsn;
+ replorigin_session_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_session_state.origin_lsn = prepare_data.end_lsn;
+ replorigin_session_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_session_state.origin_lsn = rollback_data.rollback_end_lsn;
+ replorigin_session_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_session_state.origin_lsn = commit_data->end_lsn;
+ replorigin_session_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_session_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_session_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_session_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_session_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_session_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_session_state.origin = originid;
origin_startpos = replorigin_session_get_progress(false);
CommitTransactionCommand();
diff --git a/src/include/replication/origin.h b/src/include/replication/origin.h
index ab34ef97c46..8eace3a16db 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 RepOriginSessionState
+{
+ RepOriginId origin;
+ XLogRecPtr origin_lsn;
+ TimestampTz origin_timestamp;
+} RepOriginSessionState;
+
+extern PGDLLIMPORT RepOriginSessionState replorigin_session_state;
/* GUCs */
extern PGDLLIMPORT int max_active_replication_origins;
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index ceb3fc5d980..92a6567d4f1 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2569,6 +2569,7 @@ ReorderBufferTupleCidKey
ReorderBufferUpdateProgressTxnCB
ReorderTuple
RepOriginId
+RepOriginSessionState
ReparameterizeForeignPathByChild_function
ReplaceVarsFromTargetList_context
ReplaceVarsNoMatchOption
--
2.39.5 (Apple Git-154)
[application/octet-stream] v4-0001-Refactor-replication-origin-state-reset-helpers.patch (5.3K, ../CAEoWx2=7xcxQPJgnPfWsKj2SzCNiBaxOXa1u9fSKME1Yd=r=fQ@mail.gmail.com/4-v4-0001-Refactor-replication-origin-state-reset-helpers.patch)
download | inline diff:
From 51b64d83b1c6cd8798693c5a2593fa7bce29565c Mon Sep 17 00:00:00 2001
From: "Chao Li (Evan)" <[email protected]>
Date: Wed, 24 Dec 2025 09:17:27 +0800
Subject: [PATCH v4 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: 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 2380f369578..fc9e14ecbc5 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.
+ *
+ * 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;
+}
/* ---------------------------------------------------------------------------
@@ -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_state(false);
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_state(true);
PG_RETURN_VOID();
}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 2522e372036..47104ed676c 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_state(false);
/*
* 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 718408bb599..4df177664b7 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_state(false);
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_state(false);
}
/*
diff --git a/src/include/replication/origin.h b/src/include/replication/origin.h
index 2a73f6aa492..ab34ef97c46 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_state(bool xact_only);
extern XLogRecPtr replorigin_session_get_progress(bool flush);
/* Checkpoint/Startup integration */
--
2.39.5 (Apple Git-154)
view thread (9+ 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]
Subject: Re: Refactor replication origin state reset helpers
In-Reply-To: <CAEoWx2=7xcxQPJgnPfWsKj2SzCNiBaxOXa1u9fSKME1Yd=r=fQ@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