database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-10-21 02:52:47 +03:00
Code Activity

Refactor logical worker synchronization code into a separate file.

Browse Source 
To support the upcoming addition of a sequence synchronization worker,
this patch extracts common synchronization logic shared by table sync
workers and the new sequence sync worker into a dedicated file. This
modularization improves code reuse, maintainability, and clarity in the
logical workers framework.

Author: vignesh C <vignesh21@gmail.com>
Author: Hou Zhijie <houzj.fnst@fujitsu.com>
Reviewed-by: shveta malik <shveta.malik@gmail.com>
Reviewed-by: Dilip Kumar <dilipbalaut@gmail.com>
Reviewed-by: Peter Smith <smithpb2250@gmail.com>
Reviewed-by: Hayato Kuroda <kuroda.hayato@fujitsu.com>
Reviewed-by: Chao Li <li.evan.chao@gmail.com>
Reviewed-by: Amit Kapila <amit.kapila16@gmail.com>
Discussion: https://postgr.es/m/CAA4eK1LC+KJiAkSrpE_NwvNdidw9F2os7GERUeSxSKv71gXysQ@mail.gmail.com
This commit is contained in:
Amit Kapila
2025-10-16 05:10:50 +00:00
parent 905e932f09
commit 41c674d2e3
13 changed files with 243 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -117,58 +117,15 @@
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rls.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
#include "utils/usercontext.h"
typedef enum
{
SYNC_TABLE_STATE_NEEDS_REBUILD,
SYNC_TABLE_STATE_REBUILD_STARTED,
SYNC_TABLE_STATE_VALID,
} SyncingTablesState;
static SyncingTablesState table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD;
static List *table_states_not_ready = NIL;
static bool FetchTableStates(bool *started_tx);
List *table_states_not_ready = NIL;
static StringInfo copybuf = NULL;
/*
* Exit routine for synchronization worker.
*/
pg_noreturn static void
finish_sync_worker(void)
{
/*
* Commit any outstanding transaction. This is the usual case, unless
* there was nothing to do for the table.
*/
if (IsTransactionState())
{
CommitTransactionCommand();
pgstat_report_stat(true);
}
/* And flush all writes. */
XLogFlush(GetXLogWriteRecPtr());
StartTransactionCommand();
ereport(LOG,
(errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has finished",
MySubscription->name,
get_rel_name(MyLogicalRepWorker->relid))));
CommitTransactionCommand();
/* Find the leader apply worker and signal it. */
logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid);
/* Stop gracefully */
proc_exit(0);
}
/*
* Wait until the relation sync state is set in the catalog to the expected
* one; return true when it happens.
@@ -180,7 +137,7 @@ finish_sync_worker(void)
* CATCHUP state to SYNCDONE.
*/
static bool
wait_for_relation_state_change(Oid relid, char expected_state)
wait_for_table_state_change(Oid relid, char expected_state)
{
char state;
@@ -273,15 +230,6 @@ wait_for_worker_state_change(char expected_state)
return false;
}
/*
* Callback from syscache invalidation.
*/
void
invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue)
{
table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD;
}
/*
* Handle table synchronization cooperation from the synchronization
* worker.
@@ -290,8 +238,8 @@ invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue)
* predetermined synchronization point in the WAL stream, mark the table as
* SYNCDONE and finish.
*/
static void
process_syncing_tables_for_sync(XLogRecPtr current_lsn)
void
ProcessSyncingTablesForSync(XLogRecPtr current_lsn)
{
SpinLockAcquire(&MyLogicalRepWorker->relmutex);
@@ -349,9 +297,9 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
/*
* Start a new transaction to clean up the tablesync origin tracking.
* This transaction will be ended within the finish_sync_worker().
* Now, even, if we fail to remove this here, the apply worker will
* ensure to clean it up afterward.
* This transaction will be ended within the FinishSyncWorker(). Now,
* even, if we fail to remove this here, the apply worker will ensure
* to clean it up afterward.
*
* We need to do this after the table state is set to SYNCDONE.
* Otherwise, if an error occurs while performing the database
@@ -387,7 +335,7 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
*/
replorigin_drop_by_name(originname, true, false);
finish_sync_worker();
FinishSyncWorker();
}
else
SpinLockRelease(&MyLogicalRepWorker->relmutex);
@@ -414,8 +362,8 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
* If the synchronization position is reached (SYNCDONE), then the table can
* be marked as READY and is no longer tracked.
*/
static void
process_syncing_tables_for_apply(XLogRecPtr current_lsn)
void
ProcessSyncingTablesForApply(XLogRecPtr current_lsn)
{
struct tablesync_start_time_mapping
{
@@ -431,7 +379,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
Assert(!IsTransactionState());
/* We need up-to-date sync state info for subscription tables here. */
FetchTableStates(&started_tx);
FetchRelationStates(&started_tx);
/*
* Prepare a hash table for tracking last start times of workers, to avoid
@@ -586,8 +534,8 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
StartTransactionCommand();
started_tx = true;
wait_for_relation_state_change(rstate->relid,
SUBREL_STATE_SYNCDONE);
wait_for_table_state_change(rstate->relid,
SUBREL_STATE_SYNCDONE);
}
else
LWLockRelease(LogicalRepWorkerLock);
@@ -689,37 +637,6 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
}
}
/*
* Process possible state change(s) of tables that are being synchronized.
*/
void
process_syncing_tables(XLogRecPtr current_lsn)
{
switch (MyLogicalRepWorker->type)
{
case WORKERTYPE_PARALLEL_APPLY:
/*
* Skip for parallel apply workers because they only operate on
* tables that are in a READY state. See pa_can_start() and
* should_apply_changes_for_rel().
*/
break;
case WORKERTYPE_TABLESYNC:
process_syncing_tables_for_sync(current_lsn);
break;
case WORKERTYPE_APPLY:
process_syncing_tables_for_apply(current_lsn);
break;
case WORKERTYPE_UNKNOWN:
/* Should never happen. */
elog(ERROR, "Unknown worker type");
}
}
/*
* Create list of columns for COPY based on logical relation mapping.
*/
@@ -1356,7 +1273,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
case SUBREL_STATE_SYNCDONE:
case SUBREL_STATE_READY:
case SUBREL_STATE_UNKNOWN:
finish_sync_worker(); /* doesn't return */
FinishSyncWorker(); /* doesn't return */
}
/* Calculate the name of the tablesync slot. */
@@ -1599,77 +1516,6 @@ copy_table_done:
return slotname;
}
/*
* Common code to fetch the up-to-date sync state info into the static lists.
*
* Returns true if subscription has 1 or more tables, else false.
*
* Note: If this function started the transaction (indicated by the parameter)
* then it is the caller's responsibility to commit it.
*/
static bool
FetchTableStates(bool *started_tx)
{
static bool has_subrels = false;
*started_tx = false;
if (table_states_validity != SYNC_TABLE_STATE_VALID)
{
MemoryContext oldctx;
List *rstates;
ListCell *lc;
SubscriptionRelState *rstate;
table_states_validity = SYNC_TABLE_STATE_REBUILD_STARTED;
/* Clean the old lists. */
list_free_deep(table_states_not_ready);
table_states_not_ready = NIL;
if (!IsTransactionState())
{
StartTransactionCommand();
*started_tx = true;
}
/* Fetch all non-ready tables. */
rstates = GetSubscriptionRelations(MySubscription->oid, true);
/* Allocate the tracking info in a permanent memory context. */
oldctx = MemoryContextSwitchTo(CacheMemoryContext);
foreach(lc, rstates)
{
rstate = palloc(sizeof(SubscriptionRelState));
memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState));
table_states_not_ready = lappend(table_states_not_ready, rstate);
}
MemoryContextSwitchTo(oldctx);
/*
* Does the subscription have tables?
*
* If there were not-READY relations found then we know it does. But
* if table_states_not_ready was empty we still need to check again to
* see if there are 0 tables.
*/
has_subrels = (table_states_not_ready != NIL) ||
HasSubscriptionRelations(MySubscription->oid);
/*
* If the subscription relation cache has been invalidated since we
* entered this routine, we still use and return the relations we just
* finished constructing, to avoid infinite loops, but we leave the
* table states marked as stale so that we'll rebuild it again on next
* access. Otherwise, we mark the table states as valid.
*/
if (table_states_validity == SYNC_TABLE_STATE_REBUILD_STARTED)
table_states_validity = SYNC_TABLE_STATE_VALID;
}
return has_subrels;
}
/*
* Execute the initial sync with error handling. Disable the subscription,
* if it's required.
@@ -1755,7 +1601,7 @@ TablesyncWorkerMain(Datum main_arg)
run_tablesync_worker();
finish_sync_worker();
FinishSyncWorker();
}
/*
@@ -1773,7 +1619,7 @@ AllTablesyncsReady(void)
bool has_subrels = false;
/* We need up-to-date sync state info for subscription tables here. */
has_subrels = FetchTableStates(&started_tx);
has_subrels = FetchRelationStates(&started_tx);
if (started_tx)
{
@@ -1789,21 +1635,21 @@ AllTablesyncsReady(void)
}
/*
* Return whether the subscription currently has any relations.
* Return whether the subscription currently has any tables.
*
* Note: Unlike HasSubscriptionRelations(), this function relies on cached
* information for subscription relations. Additionally, it should not be
* Note: Unlike HasSubscriptionTables(), this function relies on cached
* information for subscription tables. Additionally, it should not be
* invoked outside of apply or tablesync workers, as MySubscription must be
* initialized first.
*/
bool
HasSubscriptionRelationsCached(void)
HasSubscriptionTablesCached(void)
{
bool started_tx;
bool has_subrels;
/* We need up-to-date subscription tables info here */
has_subrels = FetchTableStates(&started_tx);
has_subrels = FetchRelationStates(&started_tx);
if (started_tx)
{