}
/*
- * Does the subscription have any relations?
+ * Does the subscription have any tables?
*
* Use this function only to know true/false, and when you have no need for the
* List returned by GetSubscriptionRelations.
*/
bool
-HasSubscriptionRelations(Oid subid)
+HasSubscriptionTables(Oid subid)
{
Relation rel;
ScanKeyData skey[1];
reorderbuffer.o \
slotsync.o \
snapbuild.o \
+ syncutils.o \
tablesync.o \
worker.o
* the subscription relation state.
*/
CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
- invalidate_syncing_table_states,
+ InvalidateSyncingRelStates,
(Datum) 0);
set_apply_error_context_origin(originname);
'reorderbuffer.c',
'slotsync.c',
'snapbuild.c',
+ 'syncutils.c',
'tablesync.c',
'worker.c',
)
--- /dev/null
+/*-------------------------------------------------------------------------
+ * syncutils.c
+ * PostgreSQL logical replication: common synchronization code
+ *
+ * Copyright (c) 2025, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/backend/replication/logical/syncutils.c
+ *
+ * NOTES
+ * This file contains code common for synchronization workers.
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "catalog/pg_subscription_rel.h"
+#include "pgstat.h"
+#include "replication/worker_internal.h"
+#include "storage/ipc.h"
+#include "utils/lsyscache.h"
+#include "utils/memutils.h"
+
+/*
+ * Enum for phases of the subscription relations state.
+ *
+ * SYNC_RELATIONS_STATE_NEEDS_REBUILD indicates that the subscription relations
+ * state is no longer valid, and the subscription relations should be rebuilt.
+ *
+ * SYNC_RELATIONS_STATE_REBUILD_STARTED indicates that the subscription
+ * relations state is being rebuilt.
+ *
+ * SYNC_RELATIONS_STATE_VALID indicates that the subscription relation state is
+ * up-to-date and valid.
+ */
+typedef enum
+{
+ SYNC_RELATIONS_STATE_NEEDS_REBUILD,
+ SYNC_RELATIONS_STATE_REBUILD_STARTED,
+ SYNC_RELATIONS_STATE_VALID,
+} SyncingRelationsState;
+
+static SyncingRelationsState relation_states_validity = SYNC_RELATIONS_STATE_NEEDS_REBUILD;
+
+/*
+ * Exit routine for synchronization worker.
+ */
+pg_noreturn void
+FinishSyncWorker(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);
+}
+
+/*
+ * Callback from syscache invalidation.
+ */
+void
+InvalidateSyncingRelStates(Datum arg, int cacheid, uint32 hashvalue)
+{
+ relation_states_validity = SYNC_RELATIONS_STATE_NEEDS_REBUILD;
+}
+
+/*
+ * Process possible state change(s) of relations that are being synchronized.
+ */
+void
+ProcessSyncingRelations(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:
+ ProcessSyncingTablesForSync(current_lsn);
+ break;
+
+ case WORKERTYPE_APPLY:
+ ProcessSyncingTablesForApply(current_lsn);
+ break;
+
+ case WORKERTYPE_UNKNOWN:
+ /* Should never happen. */
+ elog(ERROR, "Unknown worker type");
+ }
+}
+
+/*
+ * 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.
+ */
+bool
+FetchRelationStates(bool *started_tx)
+{
+ static bool has_subtables = false;
+
+ *started_tx = false;
+
+ if (relation_states_validity != SYNC_RELATIONS_STATE_VALID)
+ {
+ MemoryContext oldctx;
+ List *rstates;
+ ListCell *lc;
+ SubscriptionRelState *rstate;
+
+ relation_states_validity = SYNC_RELATIONS_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 tables and sequences that are in non-ready state. */
+ 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 tables 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_subtables = (table_states_not_ready != NIL) ||
+ HasSubscriptionTables(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 (relation_states_validity == SYNC_RELATIONS_STATE_REBUILD_STARTED)
+ relation_states_validity = SYNC_RELATIONS_STATE_VALID;
+ }
+
+ return has_subtables;
+}
#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.
* 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;
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.
* 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);
/*
* 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
*/
replorigin_drop_by_name(originname, true, false);
- finish_sync_worker();
+ FinishSyncWorker();
}
else
SpinLockRelease(&MyLogicalRepWorker->relmutex);
* 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
{
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
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);
}
}
-/*
- * 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.
*/
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. */
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.
run_tablesync_worker();
- finish_sync_worker();
+ FinishSyncWorker();
}
/*
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)
{
}
/*
- * 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)
{
* behave as if two_phase = off. When the apply worker detects that all
* tablesyncs have become READY (while the tri-state was PENDING) it will
* restart the apply worker process. This happens in
- * process_syncing_tables_for_apply.
+ * ProcessSyncingTablesForApply.
*
* When the (re-started) apply worker finds that all tablesyncs are READY for a
* two_phase tri-state of PENDING it start streaming messages with the
apply_handle_commit_internal(&commit_data);
/* Process any tables that are being synchronized in parallel. */
- process_syncing_tables(commit_data.end_lsn);
+ ProcessSyncingRelations(commit_data.end_lsn);
pgstat_report_activity(STATE_IDLE, NULL);
reset_apply_error_context_info();
in_remote_transaction = false;
/* Process any tables that are being synchronized in parallel. */
- process_syncing_tables(prepare_data.end_lsn);
+ ProcessSyncingRelations(prepare_data.end_lsn);
/*
* Since we have already prepared the transaction, in a case where the
in_remote_transaction = false;
/* Process any tables that are being synchronized in parallel. */
- process_syncing_tables(prepare_data.end_lsn);
+ ProcessSyncingRelations(prepare_data.end_lsn);
clear_subscription_skip_lsn(prepare_data.end_lsn);
in_remote_transaction = false;
/* Process any tables that are being synchronized in parallel. */
- process_syncing_tables(rollback_data.rollback_end_lsn);
+ ProcessSyncingRelations(rollback_data.rollback_end_lsn);
pgstat_report_activity(STATE_IDLE, NULL);
reset_apply_error_context_info();
pgstat_report_stat(false);
/* Process any tables that are being synchronized in parallel. */
- process_syncing_tables(prepare_data.end_lsn);
+ ProcessSyncingRelations(prepare_data.end_lsn);
/*
* Similar to prepare case, the subskiplsn could be left in a case of
}
/* Process any tables that are being synchronized in parallel. */
- process_syncing_tables(commit_data.end_lsn);
+ ProcessSyncingRelations(commit_data.end_lsn);
pgstat_report_activity(STATE_IDLE, NULL);
maybe_reread_subscription();
/* Process any table synchronization changes. */
- process_syncing_tables(last_received);
+ ProcessSyncingRelations(last_received);
}
/* Cleanup the memory. */
* RDT_GET_CANDIDATE_XID phase in such cases, this is unsafe. If users
* concurrently add tables to the subscription, the apply worker may not
* process invalidations in time. Consequently,
- * HasSubscriptionRelationsCached() might miss the new tables, leading to
+ * HasSubscriptionTablesCached() might miss the new tables, leading to
* premature advancement of oldest_nonremovable_xid.
*
* Performing the check during RDT_WAIT_FOR_LOCAL_FLUSH is safe, as
* subscription tables at this stage to prevent unnecessary tuple
* retention.
*/
- if (HasSubscriptionRelationsCached() && !AllTablesyncsReady())
+ if (HasSubscriptionTablesCached() && !AllTablesyncsReady())
{
TimestampTz now;
* the subscription relation state.
*/
CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
- invalidate_syncing_table_states,
+ InvalidateSyncingRelStates,
(Datum) 0);
}
pg_log_info("reading subscriptions");
getSubscriptions(fout);
- pg_log_info("reading subscription membership of tables");
- getSubscriptionTables(fout);
+ pg_log_info("reading subscription membership of relations");
+ getSubscriptionRelations(fout);
free(inhinfo); /* not needed any longer */
}
/*
- * getSubscriptionTables
- * Get information about subscription membership for dumpable tables. This
+ * getSubscriptionRelations
+ * Get information about subscription membership for dumpable relations. This
* will be used only in binary-upgrade mode for PG17 or later versions.
*/
void
-getSubscriptionTables(Archive *fout)
+getSubscriptionRelations(Archive *fout)
{
DumpOptions *dopt = fout->dopt;
SubscriptionInfo *subinfo = NULL;
tblinfo = findTableByOid(relid);
if (tblinfo == NULL)
- pg_fatal("failed sanity check, table with OID %u not found",
+ pg_fatal("failed sanity check, relation with OID %u not found",
relid);
/* OK, make a DumpableObject for this relationship */
extern void getPublicationTables(Archive *fout, TableInfo tblinfo[],
int numTables);
extern void getSubscriptions(Archive *fout);
-extern void getSubscriptionTables(Archive *fout);
+extern void getSubscriptionRelations(Archive *fout);
#endif /* PG_DUMP_H */
extern char GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn);
extern void RemoveSubscriptionRel(Oid subid, Oid relid);
-extern bool HasSubscriptionRelations(Oid subid);
+extern bool HasSubscriptionTables(Oid subid);
extern List *GetSubscriptionRelations(Oid subid, bool not_ready);
extern void UpdateDeadTupleRetentionStatus(Oid subid, bool active);
extern PGDLLIMPORT bool InitializingApplyWorker;
+extern PGDLLIMPORT List *table_states_not_ready;
+
extern void logicalrep_worker_attach(int slot);
extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
bool only_running);
char *originname, Size szoriginname);
extern bool AllTablesyncsReady(void);
-extern bool HasSubscriptionRelationsCached(void);
+extern bool HasSubscriptionTablesCached(void);
extern void UpdateTwoPhaseState(Oid suboid, char new_state);
-extern void process_syncing_tables(XLogRecPtr current_lsn);
-extern void invalidate_syncing_table_states(Datum arg, int cacheid,
- uint32 hashvalue);
+extern void ProcessSyncingTablesForSync(XLogRecPtr current_lsn);
+extern void ProcessSyncingTablesForApply(XLogRecPtr current_lsn);
+
+pg_noreturn extern void FinishSyncWorker(void);
+extern void InvalidateSyncingRelStates(Datum arg, int cacheid, uint32 hashvalue);
+extern void ProcessSyncingRelations(XLogRecPtr current_lsn);
+extern bool FetchRelationStates(bool *started_tx);
extern void stream_start_internal(TransactionId xid, bool first_segment);
extern void stream_stop_internal(TransactionId xid);
SyncRequestHandler
SyncRequestType
SyncStandbySlotsConfigData
-SyncingTablesState
+SyncingRelationsState
SysFKRelationship
SysScanDesc
SyscacheCallbackFunction
projects / thirdparty / postgresql.git / commitdiff
