hcq/postgresql
1
0
Fork 0
mirror of https://git.postgresql.org/git/postgresql.git synced 2026-02-24 07:17:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor replication origin state reset helpers.

Browse Source 
Factor out common logic for clearing replorigin_session_* variables
into a dedicated helper function, replorigin_xact_clear().

This removes duplicated assignments of these variables across multiple
call sites, and makes the intended scope of each reset explicit.

Author: Chao Li <lic@highgo.com>
Reviewed-by: Masahiko Sawada <sawada.mshk@gmail.com>
Reviewed-by: Ashutosh Bapat <ashutosh.bapat.oss@gmail.com>
Reviewed-by: Álvaro Herrera <alvherre@kurilemu.de>
Discussion: https://postgr.es/m/CAEoWx2=pYvfRthXHTzSrOsf5_FfyY4zJyK4zV2v4W=yjUij1cA@mail.gmail.com
This commit is contained in:
Masahiko Sawada
2026-01-28 11:45:26 -08:00
parent 1fdbca159e
commit 227eb4eea2
4 changed files with 26 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/backend/replication/logical/origin.c
View File

@ -1341,6 +1341,19 @@ replorigin_session_get_progress(bool flush)
return remote_lsn;
}
/*
* Clear the per-transaction replication origin state.
*
* replorigin_session_origin is also cleared if clear_origin is set.
*/
void
replorigin_xact_clear(bool clear_origin)
{
replorigin_session_origin_lsn = InvalidXLogRecPtr;
replorigin_session_origin_timestamp = 0;
if (clear_origin)
replorigin_session_origin = InvalidReplOriginId;
}
/* ---------------------------------------------------------------------------
@ -1466,9 +1479,7 @@ pg_replication_origin_session_reset(PG_FUNCTION_ARGS)
replorigin_session_reset();
replorigin_session_origin = InvalidReplOriginId;
replorigin_session_origin_lsn = InvalidXLogRecPtr;
replorigin_session_origin_timestamp = 0;
replorigin_xact_clear(true);
PG_RETURN_VOID();
}
@ -1536,8 +1547,8 @@ pg_replication_origin_xact_reset(PG_FUNCTION_ARGS)
{
replorigin_check_prerequisites(true, false);
replorigin_session_origin_lsn = InvalidXLogRecPtr;
replorigin_session_origin_timestamp = 0;
/* Do not clear the session origin */
replorigin_xact_clear(false);
PG_RETURN_VOID();
}

4
src/backend/replication/logical/tablesync.c
View File

@ -323,9 +323,7 @@ ProcessSyncingTablesForSync(XLogRecPtr current_lsn)
* This is needed to allow the origin to be dropped.
*/
replorigin_session_reset();
replorigin_session_origin = InvalidReplOriginId;
replorigin_session_origin_lsn = InvalidXLogRecPtr;
replorigin_session_origin_timestamp = 0;
replorigin_xact_clear(true);
/*
* Drop the tablesync's origin tracking if exists.

14
src/backend/replication/logical/worker.c
View File

@ -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_xact_state(int code, Datum arg);
/*
* Form the origin name for the subscription.
@ -5594,7 +5594,7 @@ start_apply(XLogRecPtr origin_startpos)
* transaction loss as that transaction won't be sent again by the
* server.
*/
replorigin_reset(0, (Datum) 0);
replorigin_xact_clear(true);
if (MySubscription->disableonerr)
DisableSubscriptionAndExit();
@ -5865,18 +5865,16 @@ InitializeLogRepWorker(void)
* replication workers that set up origins and apply remote transactions
* are protected.
*/
before_shmem_exit(replorigin_reset, (Datum) 0);
before_shmem_exit(on_exit_clear_xact_state, (Datum) 0);
}
/*
* Reset the origin state.
* Callback on exit to clear transaction-level replication origin state.
*/
static void
replorigin_reset(int code, Datum arg)
on_exit_clear_xact_state(int code, Datum arg)
{
replorigin_session_origin = InvalidReplOriginId;
replorigin_session_origin_lsn = InvalidXLogRecPtr;
replorigin_session_origin_timestamp = 0;
replorigin_xact_clear(true);
}
/*

3
src/include/replication/origin.h
View File

@ -67,6 +67,9 @@ extern void replorigin_session_setup(ReplOriginId node, int acquired_by);
extern void replorigin_session_reset(void);
extern XLogRecPtr replorigin_session_get_progress(bool flush);
/* Per-transaction replication origin state manipulation */
extern void replorigin_xact_clear(bool clear_origin);
/* Checkpoint/Startup integration */
extern void CheckPointReplicationOrigin(void);
extern void StartupReplicationOrigin(void);