database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2026-10-23 13:29:09 +03:00
Code Activity

Fix issues in logical replication sequence synchronization.

Browse Source 
1. Stop a running sequence synchronization worker when
ALTER SUBSCRIPTION ... DISABLE is executed.  The worker did not reread its
subscription after starting a transaction, so it kept running with a stale
copy and missed the disable. It now calls maybe_reread_subscription()
after StartTransactionCommand(), matching the apply worker.

2. Restore the invariant that publisher-side synchronization slots are
dropped last during ALTER SUBSCRIPTION ... REFRESH PUBLICATION.  The
slot-drop loop now runs after the sequence-removal loop, so the
non-transactional slot drops happen only after all catalog changes that
could still be rolled back on error.

3. Restore psql tab completion for
ALTER SUBSCRIPTION ... REFRESH PUBLICATION WITH (.

4. Make pg_stat_subscription report NULL for the fields that do not apply
to a sequence synchronization worker, which does not stream from a
walsender, and update the documentation accordingly.

5. Update the pg_subscription_rel.srsublsn catalog documentation to
describe its semantics for sequence rows.

Reported-by: Noah Misch <noah@leadboat.com>
Author: vignesh C <vignesh21@gmail.com>
Reviewed-by: Hayato Kuroda <kuroda.hayato@fujitsu.com>
Reviewed-by: Amit Kapila <amit.kapila16@gmail.com>
Backpatch-through: 19, where it was introduced
Discussion: https://postgr.es/m/20260710045217.f0.noahmisch@microsoft.com
This commit is contained in:
Amit Kapila
2026-07-27 09:04:31 +05:30
parent aa50f4f029
commit d2768065a9
6 changed files with 61 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
doc/src/sgml/catalogs.sgml
View File

@@ -8893,7 +8893,11 @@ SCRAM-SHA-256$<replaceable>&lt;iteration count&gt;</replaceable>:<replaceable>&l
<para>
Remote LSN of the state change used for synchronization coordination
when in <literal>s</literal> or <literal>r</literal> states,
otherwise null
otherwise null. For sequences, this instead holds the publisher
sequence's page LSN as of the last synchronization, which does not
track replication progress the way it does for tables; see
<xref linkend="sequences-out-of-sync"/> for how it is used to detect
out-of-sync sequences.
</para></entry>
</row>
</tbody>

19
doc/src/sgml/monitoring.sgml
View File

@@ -2318,8 +2318,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage
</para>
<para>
Process ID of the leader apply worker if this process is a parallel
apply worker; NULL if this process is a leader apply worker or a table
synchronization worker
apply worker; NULL if this process is a leader apply worker, a table
synchronization worker or a sequence synchronization worker
</para></entry>
</row>
@@ -2329,7 +2329,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage
</para>
<para>
OID of the relation that the worker is synchronizing; NULL for the
leader apply worker and parallel apply workers
leader apply worker, parallel apply workers and the sequence
synchronization worker
</para></entry>
</row>
@@ -2339,7 +2340,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage
</para>
<para>
Last write-ahead log location received, the initial value of
this field being 0; NULL for parallel apply workers
this field being 0; NULL for parallel apply workers and the sequence
synchronization worker
</para></entry>
</row>
@@ -2349,7 +2351,7 @@ description | Waiting for a newly initialized WAL file to reach durable storage
</para>
<para>
Send time of last message received from origin WAL sender; NULL for
parallel apply workers
parallel apply workers and the sequence synchronization worker
</para></entry>
</row>
@@ -2359,7 +2361,7 @@ description | Waiting for a newly initialized WAL file to reach durable storage
</para>
<para>
Receipt time of last message received from origin WAL sender; NULL for
parallel apply workers
parallel apply workers and the sequence synchronization worker
</para></entry>
</row>
@@ -2369,7 +2371,7 @@ description | Waiting for a newly initialized WAL file to reach durable storage
</para>
<para>
Last write-ahead log location reported to origin WAL sender; NULL for
parallel apply workers
parallel apply workers and the sequence synchronization worker
</para></entry>
</row>
@@ -2379,7 +2381,8 @@ description | Waiting for a newly initialized WAL file to reach durable storage
</para>
<para>
Time of last write-ahead log location reported to origin WAL
sender; NULL for parallel apply workers
sender; NULL for parallel apply workers and the sequence synchronization
worker
</para></entry>
</row>
</tbody>

56
src/backend/commands/subscriptioncmds.c
View File

@@ -1229,34 +1229,6 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
}
}
/*
* Drop the tablesync slots associated with removed tables. This has
* to be at the end because otherwise if there is an error while doing
* the database operations we won't be able to rollback dropped slots.
*/
foreach_ptr(SubRemoveRels, sub_remove_rel, sub_remove_rels)
{
if (sub_remove_rel->state != SUBREL_STATE_READY &&
sub_remove_rel->state != SUBREL_STATE_SYNCDONE)
{
char syncslotname[NAMEDATALEN] = {0};
/*
* For READY/SYNCDONE states we know the tablesync slot has
* already been dropped by the tablesync worker.
*
* For other states, there is no certainty, maybe the slot
* does not exist yet. Also, if we fail after removing some of
* the slots, next time, it will again try to drop already
* dropped slots and fail. For these reasons, we allow
* missing_ok = true for the drop.
*/
ReplicationSlotNameForTablesync(sub->oid, sub_remove_rel->relid,
syncslotname, sizeof(syncslotname));
ReplicationSlotDropAtPubNode(wrconn, syncslotname, true);
}
}
/*
* Next remove state for sequences we should not care about anymore
* using the data we collected above
@@ -1284,6 +1256,34 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
sub->name));
}
}
/*
* Drop the tablesync slots associated with removed tables. This has
* to be at the end because otherwise if there is an error while doing
* the database operations we won't be able to rollback dropped slots.
*/
foreach_ptr(SubRemoveRels, sub_remove_rel, sub_remove_rels)
{
if (sub_remove_rel->state != SUBREL_STATE_READY &&
sub_remove_rel->state != SUBREL_STATE_SYNCDONE)
{
char syncslotname[NAMEDATALEN] = {0};
/*
* For READY/SYNCDONE states we know the tablesync slot has
* already been dropped by the tablesync worker.
*
* For other states, there is no certainty, maybe the slot
* does not exist yet. Also, if we fail after removing some of
* the slots, next time, it will again try to drop already
* dropped slots and fail. For these reasons, we allow
* missing_ok = true for the drop.
*/
ReplicationSlotNameForTablesync(sub->oid, sub_remove_rel->relid,
syncslotname, sizeof(syncslotname));
ReplicationSlotDropAtPubNode(wrconn, syncslotname, true);
}
}
}
PG_FINALLY();
{

2
src/backend/replication/logical/sequencesync.c
View File

@@ -495,6 +495,7 @@ copy_sequences(WalReceiverConn *conn)
TupleTableSlot *slot;
StartTransactionCommand();
maybe_reread_subscription();
for (int idx = cur_batch_base_index; idx < n_seqinfos; idx++)
{
@@ -724,6 +725,7 @@ LogicalRepSyncSequences(void)
StringInfoData app_name;
StartTransactionCommand();
maybe_reread_subscription();
rel = table_open(SubscriptionRelRelationId, AccessShareLock);

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

@@ -5977,8 +5977,18 @@ SetupApplyOrSyncWorker(int worker_slot)
*/
/* Initialise stats to a sanish value */
MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time =
MyLogicalRepWorker->reply_time = GetCurrentTimestamp();
if (am_sequencesync_worker())
{
MyLogicalRepWorker->last_send_time =
MyLogicalRepWorker->last_recv_time =
MyLogicalRepWorker->reply_time = 0;
}
else
{
MyLogicalRepWorker->last_send_time =
MyLogicalRepWorker->last_recv_time =
MyLogicalRepWorker->reply_time = GetCurrentTimestamp();
}
/* Load the libpq-specific functions */
load_file("libpqwalreceiver", false);

3
src/bin/psql/tab-complete.in.c
View File

@@ -2368,6 +2368,9 @@ match_previous_words(int pattern_id,
/* ALTER SUBSCRIPTION <name> REFRESH */
else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "REFRESH"))
COMPLETE_WITH("PUBLICATION", "SEQUENCES");
/* ALTER SUBSCRIPTION <name> REFRESH PUBLICATION */
else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "REFRESH", "PUBLICATION"))
COMPLETE_WITH("WITH (");
/* ALTER SUBSCRIPTION <name> REFRESH PUBLICATION WITH ( */
else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "REFRESH", "PUBLICATION", "WITH", "("))
COMPLETE_WITH("copy_data");