public inbox for [email protected]
help / color / mirror / Atom feedRefactor replication origin state reset helpers
9+ messages / 3 participants
[nested] [flat]
* Refactor replication origin state reset helpers
@ 2025-12-24 01:28 Chao Li <[email protected]>
2025-12-24 10:57 ` Re: Refactor replication origin state reset helpers Ashutosh Bapat <[email protected]>
0 siblings, 1 reply; 9+ messages in thread
From: Chao Li @ 2025-12-24 01:28 UTC (permalink / raw)
To: Postgres hackers <[email protected]>
Hi Hacker,
While reviewing patches [1] and [2], I noticed some duplicate code of
clearing replication origin states, I am proposing a small patch that
removes the duplicate code blocks by introducing a couple helper
functions. No functional change at all.
[1]
https://postgr.es/m/TY4PR01MB169078771FB31B395AB496A6B94B4A@TY4PR01MB16907.jpnprd01.prod.outlook.com
[2] https://postgr.es/m/[email protected]
Best regards,
--
Chao Li (Evan)
HighGo Software Co., Ltd.
https://www.highgo.com/
Attachments:
[application/octet-stream] v1-0001-Refactor-replication-origin-state-reset-helpers.patch (4.3K, ../../CAEoWx2=pYvfRthXHTzSrOsf5_FfyY4zJyK4zV2v4W=yjUij1cA@mail.gmail.com/3-v1-0001-Refactor-replication-origin-state-reset-helpers.patch)
download | inline diff:
From 49596baf76cfa393da071de672b193bf6ba83eca Mon Sep 17 00:00:00 2001
From: "Chao Li (Evan)" <[email protected]>
Date: Wed, 24 Dec 2025 09:17:27 +0800
Subject: [PATCH v1] Refactor replication origin state reset helpers
Factor out common logic for clearing per-transaction and per-session
replication origin state into dedicated helper functions.
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]>
---
src/backend/replication/logical/origin.c | 28 +++++++++++++++++----
src/backend/replication/logical/tablesync.c | 4 +--
src/backend/replication/logical/worker.c | 4 +--
src/include/replication/origin.h | 1 +
4 files changed, 26 insertions(+), 11 deletions(-)
diff --git a/src/backend/replication/logical/origin.c b/src/backend/replication/logical/origin.c
index 2380f369578..aec9e30b4d6 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -200,6 +200,17 @@ replorigin_check_prerequisites(bool check_origins, bool recoveryOK)
errmsg("cannot manipulate replication origins during recovery")));
}
+/*
+ * replorigin_xact_clear_state
+ * Clear per transaction state variables.
+ */
+static void
+replorigin_xact_clear_state(void)
+{
+ replorigin_session_origin_lsn = InvalidXLogRecPtr;
+ replorigin_session_origin_timestamp = 0;
+}
+
/*
* IsReservedOriginName
@@ -1244,6 +1255,16 @@ replorigin_session_reset(void)
ConditionVariableBroadcast(cv);
}
+/*
+ * Clear per-session replication origin state.
+ */
+void
+replorigin_session_clear_state(void)
+{
+ replorigin_xact_clear_state();
+ replorigin_session_origin = InvalidRepOriginId;
+}
+
/*
* Do the same work replorigin_advance() does, just on the session's
* configured origin.
@@ -1412,9 +1433,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();
PG_RETURN_VOID();
}
@@ -1482,8 +1501,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_xact_clear_state();
PG_RETURN_VOID();
}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 2522e372036..d5b6782bc1d 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();
/*
* 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 d1ee0261c64..c67be657737 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -5857,9 +5857,7 @@ InitializeLogRepWorker(void)
static void
replorigin_reset(int code, Datum arg)
{
- replorigin_session_origin = InvalidRepOriginId;
- replorigin_session_origin_lsn = InvalidXLogRecPtr;
- replorigin_session_origin_timestamp = 0;
+ replorigin_session_clear_state();
}
/*
diff --git a/src/include/replication/origin.h b/src/include/replication/origin.h
index 2a73f6aa492..325f12bfd57 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(void);
extern XLogRecPtr replorigin_session_get_progress(bool flush);
/* Checkpoint/Startup integration */
--
2.39.5 (Apple Git-154)
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Refactor replication origin state reset helpers
2025-12-24 01:28 Refactor replication origin state reset helpers Chao Li <[email protected]>
@ 2025-12-24 10:57 ` Ashutosh Bapat <[email protected]>
2025-12-25 05:05 ` Re: Refactor replication origin state reset helpers Chao Li <[email protected]>
2025-12-29 14:43 ` Re: Refactor replication origin state reset helpers Álvaro Herrera <[email protected]>
0 siblings, 2 replies; 9+ messages in thread
From: Ashutosh Bapat @ 2025-12-24 10:57 UTC (permalink / raw)
To: Chao Li <[email protected]>; +Cc: Postgres hackers <[email protected]>
On Wed, Dec 24, 2025 at 6:58 AM Chao Li <[email protected]> wrote:
>
> Hi Hacker,
>
> While reviewing patches [1] and [2], I noticed some duplicate code of clearing replication origin states, I am proposing a small patch that removes the duplicate code blocks by introducing a couple helper functions. No functional change at all.
>
The new functions bring together the global variables that need to be
reset under certain conditions. The functions will help not to miss
resetting some variable. However, this can be a mild backpatching
pain. So, I am +.5 on this.
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.
Further, does it make sense to put together all the state variables
into a single structure?
It's also quite easy to confuse between these functions and
replorigin_session_reset(). It's not clear where the boundaries of the
latter end and where those of the new ones start. I think the latter
deals with the shared memory structures while the new ones deal with
the backend local state. And then there's replorigin_reset() which
adds to the confusion. That function doesn't call
replorigin_session_reset() which the other two callers of
replorigin_session_clear_state() call. Why? I think there is more to
clean here than what's in the patch. That doesn't mean that we cannot
accept this patch without larger cleanup, but it should not add to the
existing confusion.
--
Best Wishes,
Ashutosh Bapat
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Refactor replication origin state reset helpers
2025-12-24 01:28 Refactor replication origin state reset helpers Chao Li <[email protected]>
2025-12-24 10:57 ` Re: Refactor replication origin state reset helpers Ashutosh Bapat <[email protected]>
@ 2025-12-25 05:05 ` Chao Li <[email protected]>
2025-12-25 05:06 ` Re: Refactor replication origin state reset helpers Chao Li <[email protected]>
1 sibling, 1 reply; 9+ messages in thread
From: Chao Li @ 2025-12-25 05:05 UTC (permalink / raw)
To: Ashutosh Bapat <[email protected]>; +Cc: Postgres hackers <[email protected]>
Hi Ashutosh,
Thanks for your review.
On Wed, Dec 24, 2025 at 6:57 PM Ashutosh Bapat <[email protected]>
wrote:
> On Wed, Dec 24, 2025 at 6:58 AM Chao Li <[email protected]> wrote:
> >
> > Hi Hacker,
> >
> > While reviewing patches [1] and [2], I noticed some duplicate code of
> clearing replication origin states, I am proposing a small patch that
> removes the duplicate code blocks by introducing a couple helper functions.
> No functional change at all.
> >
>
> The new functions bring together the global variables that need to be
> reset under certain conditions. The functions will help not to miss
> resetting some variable. However, this can be a mild backpatching
> pain. So, I am +.5 on this.
>
> 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.
>
I like the idea of making the helpers static inline and moving them to
origin.h to stay close to the 3 global variables, which improves
readability as well. See attached v2 for the change.
> Further, does it make sense to put together all the state variables
> into a single structure?
>
I hesitate to go that far, because the 3 global variables are all
exported. So, if we really want to do that, that should belong to a
dedicated thread.
> It's also quite easy to confuse between these functions and
> replorigin_session_reset(). It's not clear where the boundaries of the
> latter end and where those of the new ones start. I think the latter
> deals with the shared memory structures while the new ones deal with
> the backend local state. And then there's replorigin_reset() which
> adds to the confusion. That function doesn't call
> replorigin_session_reset() which the other two callers of
> replorigin_session_clear_state() call. Why? I think there is more to
> clean here than what's in the patch. That doesn't mean that we cannot
> accept this patch without larger cleanup, but it should not add to the
> existing confusion.
>
Good point. I am trying to resolve the confusions in v2.
For replorigin_reset(), it's easy to address. As this function is local
static and the only purpose is to satisfy the pg_on_exit_callback
contract, I just renamed it to on_exit_clear_state() in v2.
For replorigin_session_reset(), there are only 2 callers and both callers
call replorigin_session_clear_state() immediately after
replorigin_session_reset(). So in v2, I made replorigin_session_reset() to
call replorigin_session_clear_state(), and added some comments in
replorigin_session_reset(). Accordingly, replorigin_session_reset() is used
to reset states set by replorigin_session_setup(), and every call of
replorigin_session_setup() is immediately followed by
"replorigin_session_origin = originid;", so I moved this assignment into
replorigin_session_setup().
With these broader changes, this patch is no longer trivial, so I just
added it to CF:
Best regards,
--
Chao Li (Evan)
HighGo Software Co., Ltd.
https://www.highgo.com/
Attachments:
[application/octet-stream] v2-0001-Refactor-replication-origin-state-reset-helpers.patch (7.3K, ../../CAEoWx2n53BjnE8daMA6yVKEN3e=ZPsMhH8ivC7TgeYsetet=Dg@mail.gmail.com/3-v2-0001-Refactor-replication-origin-state-reset-helpers.patch)
download | inline diff:
From 496cc60401705a4512915db7ebf3358f7004014e Mon Sep 17 00:00:00 2001
From: "Chao Li (Evan)" <[email protected]>
Date: Wed, 24 Dec 2025 09:17:27 +0800
Subject: [PATCH v2] Refactor replication origin state reset helpers
Factor out common logic for clearing per-transaction and per-session
replication origin state into dedicated helper functions.
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]>
---
.../replication/logical/applyparallelworker.c | 1 -
src/backend/replication/logical/origin.c | 20 +++++++++++-------
src/backend/replication/logical/tablesync.c | 5 -----
src/backend/replication/logical/worker.c | 15 ++++++-------
src/include/replication/origin.h | 21 +++++++++++++++++++
5 files changed, 39 insertions(+), 23 deletions(-)
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index a4aafcf5b6e..b05279e0809 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -962,7 +962,6 @@ ParallelApplyWorkerMain(Datum main_arg)
* origin which was already acquired by its leader process.
*/
replorigin_session_setup(originid, MyLogicalRepWorker->leader_pid);
- replorigin_session_origin = originid;
CommitTransactionCommand();
/*
diff --git a/src/backend/replication/logical/origin.c b/src/backend/replication/logical/origin.c
index 2380f369578..45d7bc5abc8 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -1213,6 +1213,9 @@ replorigin_session_setup(RepOriginId node, int acquired_by)
/* probably this one is pointless */
ConditionVariableBroadcast(&session_replication_state->origin_cv);
+
+ /* set local state too */
+ replorigin_session_origin = node;
}
/*
@@ -1233,6 +1236,9 @@ replorigin_session_reset(void)
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("no replication origin is configured")));
+ /*
+ * Clear sessioin state in shared memory
+ */
LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
session_replication_state->acquired_by = 0;
@@ -1242,6 +1248,11 @@ replorigin_session_reset(void)
LWLockRelease(ReplicationOriginLock);
ConditionVariableBroadcast(cv);
+
+ /*
+ * Clear local session state
+ */
+ replorigin_session_clear_state();
}
/*
@@ -1395,8 +1406,6 @@ pg_replication_origin_session_setup(PG_FUNCTION_ARGS)
pid = PG_GETARG_INT32(1);
replorigin_session_setup(origin, pid);
- replorigin_session_origin = origin;
-
pfree(name);
PG_RETURN_VOID();
@@ -1412,10 +1421,6 @@ 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;
-
PG_RETURN_VOID();
}
@@ -1482,8 +1487,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_xact_clear_state();
PG_RETURN_VOID();
}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 2522e372036..6ac467a9e19 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -323,9 +323,6 @@ 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;
/*
* Drop the tablesync's origin tracking if exists.
@@ -1320,7 +1317,6 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
*/
originid = replorigin_by_name(originname, false);
replorigin_session_setup(originid, 0);
- replorigin_session_origin = originid;
*origin_startpos = replorigin_session_get_progress(false);
CommitTransactionCommand();
@@ -1407,7 +1403,6 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
replorigin_session_setup(originid, 0);
- replorigin_session_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 718408bb599..651045debee 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();
if (MySubscription->disableonerr)
DisableSubscriptionAndExit();
@@ -5652,7 +5652,6 @@ run_apply_worker(void)
if (!OidIsValid(originid))
originid = replorigin_create(originname);
replorigin_session_setup(originid, 0);
- replorigin_session_origin = originid;
origin_startpos = replorigin_session_get_progress(false);
CommitTransactionCommand();
@@ -5865,18 +5864,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();
}
/*
diff --git a/src/include/replication/origin.h b/src/include/replication/origin.h
index 2a73f6aa492..288f9ff658f 100644
--- a/src/include/replication/origin.h
+++ b/src/include/replication/origin.h
@@ -44,6 +44,27 @@ extern PGDLLIMPORT RepOriginId replorigin_session_origin;
extern PGDLLIMPORT XLogRecPtr replorigin_session_origin_lsn;
extern PGDLLIMPORT TimestampTz replorigin_session_origin_timestamp;
+/*
+ * Clear per-transaction replication origin state.
+ */
+static inline void
+replorigin_xact_clear_state(void)
+{
+ replorigin_session_origin_lsn = InvalidXLogRecPtr;
+ replorigin_session_origin_timestamp = 0;
+}
+
+/*
+ * Clear per-session replication origin state.
+ */
+static inline void
+replorigin_session_clear_state(void)
+{
+ replorigin_xact_clear_state();
+ replorigin_session_origin = InvalidRepOriginId;
+}
+
+
/* GUCs */
extern PGDLLIMPORT int max_active_replication_origins;
--
2.39.5 (Apple Git-154)
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Refactor replication origin state reset helpers
2025-12-24 01:28 Refactor replication origin state reset helpers Chao Li <[email protected]>
2025-12-24 10:57 ` Re: Refactor replication origin state reset helpers Ashutosh Bapat <[email protected]>
2025-12-25 05:05 ` Re: Refactor replication origin state reset helpers Chao Li <[email protected]>
@ 2025-12-25 05:06 ` Chao Li <[email protected]>
0 siblings, 0 replies; 9+ messages in thread
From: Chao Li @ 2025-12-25 05:06 UTC (permalink / raw)
To: Ashutosh Bapat <[email protected]>; +Cc: Postgres hackers <[email protected]>
> On Dec 25, 2025, at 13:05, Chao Li <[email protected]> wrote:
>
> With these broader changes, this patch is no longer trivial, so I just added it to CF:
https://commitfest.postgresql.org/patch/6345/
Best regards,
--
Chao Li (Evan)
HighGo Software Co., Ltd.
https://www.highgo.com/
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Refactor replication origin state reset helpers
2025-12-24 01:28 Refactor replication origin state reset helpers Chao Li <[email protected]>
2025-12-24 10:57 ` Re: Refactor replication origin state reset helpers Ashutosh Bapat <[email protected]>
@ 2025-12-29 14:43 ` Álvaro Herrera <[email protected]>
2025-12-29 15:00 ` Re: Refactor replication origin state reset helpers Álvaro Herrera <[email protected]>
2025-12-30 04:47 ` Re: Refactor replication origin state reset helpers Ashutosh Bapat <[email protected]>
1 sibling, 2 replies; 9+ messages in thread
From: Álvaro Herrera @ 2025-12-29 14:43 UTC (permalink / raw)
To: Ashutosh Bapat <[email protected]>; +Cc: Chao Li <[email protected]>; Postgres hackers <[email protected]>
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?
> 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.
> It's also quite easy to confuse between these functions and
> replorigin_session_reset(). It's not clear where the boundaries of the
> latter end and where those of the new ones start. I think the latter
> deals with the shared memory structures while the new ones deal with
> the backend local state. And then there's replorigin_reset() which
> adds to the confusion. That function doesn't call
> replorigin_session_reset() which the other two callers of
> replorigin_session_clear_state() call. Why? I think there is more to
> clean here than what's in the patch. That doesn't mean that we cannot
> accept this patch without larger cleanup, but it should not add to the
> existing confusion.
Good points. A decrease in the total quantity of cruft would be a good
outcome of a patch in this area.
--
Álvaro Herrera 48°01'N 7°57'E — https://www.EnterpriseDB.com/
"Update: super-fast reaction on the Postgres bugs mailing list. The report
was acknowledged [...], and a fix is under discussion.
The wonders of open-source !"
https://twitter.com/gunnarmorling/status/1596080409259003906
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Refactor replication origin state reset helpers
2025-12-24 01:28 Refactor replication origin state reset helpers Chao Li <[email protected]>
2025-12-24 10:57 ` Re: Refactor replication origin state reset helpers Ashutosh Bapat <[email protected]>
2025-12-29 14:43 ` Re: Refactor replication origin state reset helpers Álvaro Herrera <[email protected]>
@ 2025-12-29 15:00 ` Álvaro Herrera <[email protected]>
1 sibling, 0 replies; 9+ messages in thread
From: Álvaro Herrera @ 2025-12-29 15:00 UTC (permalink / raw)
To: Ashutosh Bapat <[email protected]>; +Cc: Chao Li <[email protected]>; Postgres hackers <[email protected]>
On 2025-Dec-29, Álvaro Herrera wrote:
> On 2025-Dec-24, Ashutosh Bapat wrote:
> > It's also quite easy to confuse between these functions and
> > replorigin_session_reset(). It's not clear where the boundaries of the
> > latter end and where those of the new ones start. I think the latter
> > deals with the shared memory structures while the new ones deal with
> > the backend local state. And then there's replorigin_reset() which
> > adds to the confusion. That function doesn't call
> > replorigin_session_reset() which the other two callers of
> > replorigin_session_clear_state() call. Why? I think there is more to
> > clean here than what's in the patch. That doesn't mean that we cannot
> > accept this patch without larger cleanup, but it should not add to the
> > existing confusion.
I think we should just rename the worker.c function to have a less
generic-sounding name (so that it becomes more obvious that it's a
worker.c-specific function); have both replorigin_session_clear_state()
and replorigin_xact_clear_state() be a single function, with a boolean
flag to indicate whether to reset replorigin_session_origin or not; and
have replorigin_session_reset() call replorigin_session_clear_state()
internally rather than require every single of its callers do it
separately, which IMO makes no sense. With these three changes I think
the code would become quite clear.
--
Álvaro Herrera PostgreSQL Developer — https://www.EnterpriseDB.com/
Thou shalt check the array bounds of all strings (indeed, all arrays), for
surely where thou typest "foo" someone someday shall type
"supercalifragilisticexpialidocious" (5th Commandment for C programmers)
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Refactor replication origin state reset helpers
2025-12-24 01:28 Refactor replication origin state reset helpers Chao Li <[email protected]>
2025-12-24 10:57 ` Re: Refactor replication origin state reset helpers Ashutosh Bapat <[email protected]>
2025-12-29 14:43 ` Re: Refactor replication origin state reset helpers Álvaro Herrera <[email protected]>
@ 2025-12-30 04:47 ` Ashutosh Bapat <[email protected]>
2025-12-30 05:07 ` Re: Refactor replication origin state reset helpers Chao Li <[email protected]>
1 sibling, 1 reply; 9+ messages in thread
From: Ashutosh Bapat @ 2025-12-30 04:47 UTC (permalink / raw)
To: Álvaro Herrera <[email protected]>; +Cc: Chao Li <[email protected]>; Postgres hackers <[email protected]>
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.
>
Agreed.
--
Best Wishes,
Ashutosh Bapat
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Refactor replication origin state reset helpers
2025-12-24 01:28 Refactor replication origin state reset helpers Chao Li <[email protected]>
2025-12-24 10:57 ` Re: Refactor replication origin state reset helpers Ashutosh Bapat <[email protected]>
2025-12-29 14:43 ` Re: Refactor replication origin state reset helpers Álvaro Herrera <[email protected]>
2025-12-30 04:47 ` Re: Refactor replication origin state reset helpers Ashutosh Bapat <[email protected]>
@ 2025-12-30 05:07 ` Chao Li <[email protected]>
2025-12-30 07:17 ` Re: Refactor replication origin state reset helpers Chao Li <[email protected]>
0 siblings, 1 reply; 9+ messages in thread
From: Chao Li @ 2025-12-30 05:07 UTC (permalink / raw)
To: Ashutosh Bapat <[email protected]>; +Cc: Álvaro Herrera <[email protected]>; Postgres hackers <[email protected]>
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.
Best regards,
--
Chao Li (Evan)
HighGo Software Co., Ltd.
https://www.highgo.com/
Attachments:
[application/octet-stream] v3-0001-Refactor-replication-origin-state-reset-helpers.patch (5.5K, ../../CAEoWx2mT+gu8H-KhaeNodk4T7csX-_wwqRew=-t3zo6b6jXfaQ@mail.gmail.com/3-v3-0001-Refactor-replication-origin-state-reset-helpers.patch)
download | inline diff:
From ef62c41cec580e9fd6e25e4e7ed10ce24391c65e Mon Sep 17 00:00:00 2001
From: "Chao Li (Evan)" <[email protected]>
Date: Wed, 24 Dec 2025 09:17:27 +0800
Subject: [PATCH v3 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 | 22 +++++++++++++--------
src/backend/replication/logical/tablesync.c | 4 +---
src/backend/replication/logical/worker.c | 14 ++++++-------
src/include/replication/origin.h | 1 +
4 files changed, 22 insertions(+), 19 deletions(-)
diff --git a/src/backend/replication/logical/origin.c b/src/backend/replication/logical/origin.c
index 2380f369578..2635bf3d54f 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;
+}
/* ---------------------------------------------------------------------------
@@ -1395,8 +1408,6 @@ pg_replication_origin_session_setup(PG_FUNCTION_ARGS)
pid = PG_GETARG_INT32(1);
replorigin_session_setup(origin, pid);
- replorigin_session_origin = origin;
-
pfree(name);
PG_RETURN_VOID();
@@ -1412,10 +1423,6 @@ 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;
-
PG_RETURN_VOID();
}
@@ -1482,8 +1489,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)
[application/octet-stream] v3-0002-Consolidate-replication-origin-session-globals-in.patch (18.8K, ../../CAEoWx2mT+gu8H-KhaeNodk4T7csX-_wwqRew=-t3zo6b6jXfaQ@mail.gmail.com/4-v3-0002-Consolidate-replication-origin-session-globals-in.patch)
download | inline diff:
From afd6e9fc22bed5e135d3cc5a5b970466f771c22f Mon Sep 17 00:00:00 2001
From: "Chao Li (Evan)" <[email protected]>
Date: Tue, 30 Dec 2025 12:55:31 +0800
Subject: [PATCH v3 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 | 30 ++++++++---------
src/backend/access/transam/xact.c | 32 +++++++++----------
src/backend/access/transam/xloginsert.c | 6 ++--
.../replication/logical/applyparallelworker.c | 6 ++--
src/backend/replication/logical/origin.c | 18 +++++------
src/backend/replication/logical/tablesync.c | 4 +--
src/backend/replication/logical/worker.c | 28 ++++++++--------
src/include/replication/origin.h | 11 +++++--
8 files changed, 70 insertions(+), 65 deletions(-)
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 3bc85986829..fd69f687bf2 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,7 +2376,7 @@ 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);
/*
@@ -2387,12 +2387,12 @@ RecordTransactionCommitPrepared(TransactionId xid,
* 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..0c32f633c6d 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,7 +1462,7 @@ 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);
/*
@@ -1475,12 +1475,12 @@ RecordTransactionCommit(void)
* 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 2635bf3d54f..96c9c7e5f7e 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
@@ -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;
}
@@ -1434,7 +1434,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);
}
@@ -1478,8 +1478,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;
--
2.39.5 (Apple Git-154)
^ permalink raw reply [nested|flat] 9+ messages in thread
* Re: Refactor replication origin state reset helpers
2025-12-24 01:28 Refactor replication origin state reset helpers Chao Li <[email protected]>
2025-12-24 10:57 ` Re: Refactor replication origin state reset helpers Ashutosh Bapat <[email protected]>
2025-12-29 14:43 ` Re: Refactor replication origin state reset helpers Álvaro Herrera <[email protected]>
2025-12-30 04:47 ` Re: Refactor replication origin state reset helpers Ashutosh Bapat <[email protected]>
2025-12-30 05:07 ` Re: Refactor replication origin state reset helpers Chao Li <[email protected]>
@ 2025-12-30 07:17 ` Chao Li <[email protected]>
0 siblings, 0 replies; 9+ messages in thread
From: Chao Li @ 2025-12-30 07:17 UTC (permalink / raw)
To: Ashutosh Bapat <[email protected]>; +Cc: Álvaro Herrera <[email protected]>; Postgres hackers <[email protected]>
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)
^ permalink raw reply [nested|flat] 9+ messages in thread
end of thread, other threads:[~2025-12-30 07:17 UTC | newest]
Thread overview: 9+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2025-12-24 01:28 Refactor replication origin state reset helpers Chao Li <[email protected]>
2025-12-24 10:57 ` Ashutosh Bapat <[email protected]>
2025-12-25 05:05 ` Chao Li <[email protected]>
2025-12-25 05:06 ` Chao Li <[email protected]>
2025-12-29 14:43 ` Álvaro Herrera <[email protected]>
2025-12-29 15:00 ` Álvaro Herrera <[email protected]>
2025-12-30 04:47 ` Ashutosh Bapat <[email protected]>
2025-12-30 05:07 ` Chao Li <[email protected]>
2025-12-30 07:17 ` Chao Li <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox