From 06ba9ee59f42f157df199b8f5510c07b841a31eb Mon Sep 17 00:00:00 2001 From: Vignesh C Date: Tue, 2 Sep 2025 16:45:14 +0530 Subject: [PATCH v20250902 5/7] Introduce "REFRESH PUBLICATION SEQUENCES" for subscriptions This patch adds support for a new SQL command: ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES This command update the sequence entries present in the pg_subscription_rel catalog table with the INIT state to trigger resynchronization. Additionally, the following subscription commands: ALTER SUBSCRIPTION ... REFRESH PUBLICATION ALTER SUBSCRIPTION ... ADD PUBLICATION ALTER SUBSCRIPTION ... DROP PUBLICATION ALTER SUBSCRIPTION ... SET PUBLICATION have been extended to also refresh sequence mappings. These commands will: Add newly published sequences that are not yet part of the subscription. Remove sequences that are no longer included in the publication. This ensures that sequence replication remains aligned with the current state of the publication on the publisher side, improving consistency and reducing manual maintenance. Author: Vignesh C, Tomas Vondra Reviewer: Amit Kapila, Shveta Malik, Dilip Kumar, Peter Smith, Nisha Moond Discussion: https://www.postgresql.org/message-id/CAA4eK1LC+KJiAkSrpE_NwvNdidw9F2os7GERUeSxSKv71gXysQ@mail.gmail.com --- src/backend/catalog/pg_publication.c | 65 +++- src/backend/catalog/pg_subscription.c | 60 +++- src/backend/catalog/system_views.sql | 10 + src/backend/commands/subscriptioncmds.c | 329 ++++++++++++++------ src/backend/executor/execReplication.c | 4 +- src/backend/parser/gram.y | 9 + src/backend/replication/logical/syncutils.c | 5 +- src/backend/replication/logical/tablesync.c | 2 +- src/backend/replication/pgoutput/pgoutput.c | 2 +- src/bin/psql/tab-complete.in.c | 2 +- src/include/catalog/pg_proc.dat | 5 + src/include/catalog/pg_publication.h | 2 +- src/include/catalog/pg_subscription_rel.h | 11 +- src/include/nodes/parsenodes.h | 1 + src/test/regress/expected/rules.out | 8 + src/tools/pgindent/typedefs.list | 1 + 16 files changed, 402 insertions(+), 114 deletions(-) diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c index b306455aaad..dc46d24c05d 100644 --- a/src/backend/catalog/pg_publication.c +++ b/src/backend/catalog/pg_publication.c @@ -777,8 +777,8 @@ GetRelationPublications(Oid relid) /* * Gets list of relation oids for a publication. * - * This should only be used FOR TABLE publications, the FOR ALL TABLES - * should use GetAllTablesPublicationRelations(). + * This should only be used FOR TABLE publications, the FOR ALL TABLES/SEQUENCES + * should use GetAllPublicationRelations(). */ List * GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt) @@ -858,14 +858,16 @@ GetAllTablesPublications(void) } /* - * Gets list of all relation published by FOR ALL TABLES publication(s). + * Gets list of all relations published by FOR ALL TABLES/SEQUENCES + * publication(s). * * If the publication publishes partition changes via their respective root * partitioned tables, we must exclude partitions in favor of including the - * root partitioned tables. + * root partitioned tables. This is not applicable for FOR ALL SEQEUNCES + * publication. */ List * -GetAllTablesPublicationRelations(bool pubviaroot) +GetAllPublicationRelations(char relkind, bool pubviaroot) { Relation classRel; ScanKeyData key[1]; @@ -873,12 +875,14 @@ GetAllTablesPublicationRelations(bool pubviaroot) HeapTuple tuple; List *result = NIL; + Assert(!(relkind == RELKIND_SEQUENCE && pubviaroot)); + classRel = table_open(RelationRelationId, AccessShareLock); ScanKeyInit(&key[0], Anum_pg_class_relkind, BTEqualStrategyNumber, F_CHAREQ, - CharGetDatum(RELKIND_RELATION)); + CharGetDatum(relkind)); scan = table_beginscan_catalog(classRel, 1, key); @@ -1165,7 +1169,8 @@ pg_get_publication_tables(PG_FUNCTION_ARGS) * those. Otherwise, get the partitioned table itself. */ if (pub_elem->alltables) - pub_elem_tables = GetAllTablesPublicationRelations(pub_elem->pubviaroot); + pub_elem_tables = GetAllPublicationRelations(RELKIND_RELATION, + pub_elem->pubviaroot); else { List *relids, @@ -1337,3 +1342,49 @@ pg_get_publication_tables(PG_FUNCTION_ARGS) SRF_RETURN_DONE(funcctx); } + +/* + * Returns Oids of sequences in a publication. + */ +Datum +pg_get_publication_sequences(PG_FUNCTION_ARGS) +{ + FuncCallContext *funcctx; + List *sequences = NIL; + + /* stuff done only on the first call of the function */ + if (SRF_IS_FIRSTCALL()) + { + char *pubname = text_to_cstring(PG_GETARG_TEXT_PP(0)); + Publication *publication; + MemoryContext oldcontext; + + /* create a function context for cross-call persistence */ + funcctx = SRF_FIRSTCALL_INIT(); + + /* switch to memory context appropriate for multiple function calls */ + oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx); + + publication = GetPublicationByName(pubname, false); + + if (publication->allsequences) + sequences = GetAllPublicationRelations(RELKIND_SEQUENCE, false); + + funcctx->user_fctx = (void *) sequences; + + MemoryContextSwitchTo(oldcontext); + } + + /* stuff done on every call of the function */ + funcctx = SRF_PERCALL_SETUP(); + sequences = (List *) funcctx->user_fctx; + + if (funcctx->call_cntr < list_length(sequences)) + { + Oid relid = list_nth_oid(sequences, funcctx->call_cntr); + + SRF_RETURN_NEXT(funcctx, ObjectIdGetDatum(relid)); + } + + SRF_RETURN_DONE(funcctx); +} diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c index e06587b0265..5a8275d49ba 100644 --- a/src/backend/catalog/pg_subscription.c +++ b/src/backend/catalog/pg_subscription.c @@ -480,7 +480,9 @@ RemoveSubscriptionRel(Oid subid, Oid relid) * leave tablesync slots or origins in the system when the * corresponding table is dropped. */ - if (!OidIsValid(subid) && subrel->srsubstate != SUBREL_STATE_READY) + if (!OidIsValid(subid) && + get_rel_relkind(subrel->srrelid) != RELKIND_SEQUENCE && + subrel->srsubstate != SUBREL_STATE_READY) { ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), @@ -517,7 +519,8 @@ HasSubscriptionTables(Oid subid) Relation rel; ScanKeyData skey[1]; SysScanDesc scan; - bool has_subrels; + HeapTuple tup; + bool has_subrels = false; rel = table_open(SubscriptionRelRelationId, AccessShareLock); @@ -529,8 +532,22 @@ HasSubscriptionTables(Oid subid) scan = systable_beginscan(rel, InvalidOid, false, NULL, 1, skey); - /* If even a single tuple exists then the subscription has tables. */ - has_subrels = HeapTupleIsValid(systable_getnext(scan)); + while (HeapTupleIsValid(tup = systable_getnext(scan))) + { + Form_pg_subscription_rel subrel; + + subrel = (Form_pg_subscription_rel) GETSTRUCT(tup); + + /* + * Skip sequence tuples. If even a single table tuple exists then the + * subscription has tables. + */ + if (get_rel_relkind(subrel->srrelid) != RELKIND_SEQUENCE) + { + has_subrels = true; + break; + } + } /* Cleanup */ systable_endscan(scan); @@ -542,12 +559,22 @@ HasSubscriptionTables(Oid subid) /* * Get the relations for the subscription. * - * If not_ready is true, return only the relations that are not in a ready - * state, otherwise return all the relations of the subscription. The - * returned list is palloc'ed in the current memory context. + * get_tables: get relations for tables of the subscription. + * + * get_sequences: get relations for sequences of the subscription. + * + * not_ready: + * If getting tables and not_ready is false get all tables, otherwise, + * only get tables that have not reached READY state. + * If getting sequences and not_ready is false get all sequences, + * otherwise, only get sequences that have not reached READY state (i.e. are + * still in INIT state). + * + * The returned list is palloc'ed in the current memory context. */ List * -GetSubscriptionRelations(Oid subid, bool not_ready) +GetSubscriptionRelations(Oid subid, bool get_tables, bool get_sequences, + bool not_ready) { List *res = NIL; Relation rel; @@ -556,6 +583,9 @@ GetSubscriptionRelations(Oid subid, bool not_ready) ScanKeyData skey[2]; SysScanDesc scan; + /* One or both of 'get_tables' and 'get_sequences' must be true. */ + Assert(get_tables || get_sequences); + rel = table_open(SubscriptionRelRelationId, AccessShareLock); ScanKeyInit(&skey[nkeys++], @@ -578,9 +608,23 @@ GetSubscriptionRelations(Oid subid, bool not_ready) SubscriptionRelState *relstate; Datum d; bool isnull; + char relkind; subrel = (Form_pg_subscription_rel) GETSTRUCT(tup); + /* Relation is either a sequence or a table */ + relkind = get_rel_relkind(subrel->srrelid); + Assert(relkind == RELKIND_SEQUENCE || relkind == RELKIND_RELATION || + relkind == RELKIND_PARTITIONED_TABLE); + + /* Skip sequences if they were not requested */ + if (!get_sequences && (relkind == RELKIND_SEQUENCE)) + continue; + + /* Skip tables if they were not requested */ + if (!get_tables && (relkind != RELKIND_SEQUENCE)) + continue; + relstate = (SubscriptionRelState *) palloc(sizeof(SubscriptionRelState)); relstate->relid = subrel->srrelid; relstate->state = subrel->srsubstate; diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index c77fa0234bb..01d300d3cf4 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -394,6 +394,16 @@ CREATE VIEW pg_publication_tables AS pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace) WHERE C.oid = GPT.relid; +CREATE VIEW pg_publication_sequences AS + SELECT + P.pubname AS pubname, + N.nspname AS schemaname, + C.relname AS sequencename + FROM pg_publication P, + LATERAL pg_get_publication_sequences(P.pubname) GPS, + pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace) + WHERE C.oid = GPS.relid; + CREATE VIEW pg_locks AS SELECT * FROM pg_lock_status() AS L; diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 5c757776afc..344dfa8e894 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -27,6 +27,7 @@ #include "catalog/objectaddress.h" #include "catalog/pg_authid_d.h" #include "catalog/pg_database_d.h" +#include "catalog/pg_sequence.h" #include "catalog/pg_subscription.h" #include "catalog/pg_subscription_rel.h" #include "catalog/pg_type.h" @@ -106,7 +107,7 @@ typedef struct SubOpts XLogRecPtr lsn; } SubOpts; -static List *fetch_table_list(WalReceiverConn *wrconn, List *publications); +static List *fetch_relation_list(WalReceiverConn *wrconn, List *publications); static void check_publications_origin(WalReceiverConn *wrconn, List *publications, bool copydata, bool retain_dead_tuples, char *origin, @@ -736,6 +737,12 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, recordDependencyOnOwner(SubscriptionRelationId, subid, owner); + /* + * XXX: Currently, a replication origin is created for all subscriptions, + * including those for sequence-only publications. However, this is + * unnecessary, as incremental synchronization of sequences is not + * supported. + */ ReplicationOriginNameForLogicalRep(subid, InvalidOid, originname, sizeof(originname)); replorigin_create(originname); @@ -747,9 +754,6 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, { char *err; WalReceiverConn *wrconn; - List *tables; - ListCell *lc; - char table_state; bool must_use_password; /* Try to connect to the publisher. */ @@ -764,6 +768,10 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, PG_TRY(); { + bool has_tables = false; + List *relations; + char relation_state; + check_publications(wrconn, publications); check_publications_origin(wrconn, publications, opts.copy_data, opts.retaindeadtuples, opts.origin, @@ -776,25 +784,46 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, * Set sync state based on if we were asked to do data copy or * not. */ - table_state = opts.copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY; + relation_state = opts.copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY; /* - * Get the table list from publisher and build local table status - * info. + * Build local relation status info. Relations are for both tables + * and sequences from the publisher. */ - tables = fetch_table_list(wrconn, publications); - foreach(lc, tables) + relations = fetch_relation_list(wrconn, publications); + + foreach_ptr(SubscriptionRelKind, relinfo, relations) { - RangeVar *rv = (RangeVar *) lfirst(lc); Oid relid; + char relkind; + bool pubisseq; + bool subisseq; + RangeVar *rv = relinfo->rv; relid = RangeVarGetRelid(rv, AccessShareLock, false); + relkind = get_rel_relkind(relid); /* Check for supported relkind. */ - CheckSubscriptionRelkind(get_rel_relkind(relid), - rv->schemaname, rv->relname); + CheckSubscriptionRelkind(relkind, rv->schemaname, rv->relname); + has_tables |= (relkind != RELKIND_SEQUENCE); + + pubisseq = (relinfo->relkind == RELKIND_SEQUENCE); + subisseq = (relkind == RELKIND_SEQUENCE); + + /* + * Allow RELKIND_RELATION and RELKIND_PARTITIONED_TABLE to be + * treated interchangeably, but ensure that sequences + * (RELKIND_SEQUENCE) match exactly on both publisher and + * subscriber. + */ + if (pubisseq != subisseq) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("relation \"%s.%s\" has relkind \"%c\" on the publisher but relkind \"%c\" on the subscriber", + rv->schemaname, rv->relname, relinfo->relkind, relkind)); - AddSubscriptionRelState(subid, relid, table_state, + + AddSubscriptionRelState(subid, relid, relation_state, InvalidXLogRecPtr, true); } @@ -802,6 +831,11 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, * If requested, create permanent slot for the subscription. We * won't use the initial snapshot for anything, so no need to * export it. + * + * XXX: Currently, a replication slot is created for all + * subscriptions, including those for sequence-only publications. + * However, this is unnecessary, as incremental synchronization of + * sequences is not supported. */ if (opts.create_slot) { @@ -825,7 +859,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, * PENDING, to allow ALTER SUBSCRIPTION ... REFRESH * PUBLICATION to work. */ - if (opts.twophase && !opts.copy_data && tables != NIL) + if (opts.twophase && !opts.copy_data && has_tables) twophase_enabled = true; walrcv_create_slot(wrconn, opts.slot_name, false, twophase_enabled, @@ -869,13 +903,12 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, List *validate_publications) { char *err; - List *pubrel_names; + List *pubrels = NIL; List *subrel_states; Oid *subrel_local_oids; Oid *pubrel_local_oids; ListCell *lc; int off; - int remove_rel_len; int subrel_count; Relation rel = NULL; typedef struct SubRemoveRels @@ -883,7 +916,8 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, Oid relid; char state; } SubRemoveRels; - SubRemoveRels *sub_remove_rels; + + List *sub_remove_rels = NIL; WalReceiverConn *wrconn; bool must_use_password; @@ -905,17 +939,17 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, if (validate_publications) check_publications(wrconn, validate_publications); - /* Get the table list from publisher. */ - pubrel_names = fetch_table_list(wrconn, sub->publications); + /* Get the relation list from publisher. */ + pubrels = fetch_relation_list(wrconn, sub->publications); - /* Get local table list. */ - subrel_states = GetSubscriptionRelations(sub->oid, false); + /* Get local relation list. */ + subrel_states = GetSubscriptionRelations(sub->oid, true, true, false); subrel_count = list_length(subrel_states); /* - * Build qsorted array of local table oids for faster lookup. This can - * potentially contain all tables in the database so speed of lookup - * is important. + * Build qsorted array of local relation oids for faster lookup. This + * can potentially contain all relation in the database so speed of + * lookup is important. */ subrel_local_oids = palloc(subrel_count * sizeof(Oid)); off = 0; @@ -933,34 +967,47 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, subrel_local_oids, subrel_count, sub->name); /* - * Rels that we want to remove from subscription and drop any slots - * and origins corresponding to them. - */ - sub_remove_rels = palloc(subrel_count * sizeof(SubRemoveRels)); - - /* - * Walk over the remote tables and try to match them to locally known - * tables. If the table is not known locally create a new state for - * it. + * Walk over the remote relations and try to match them to locally + * known tables. If the table is not known locally create a new state + * for it. * - * Also builds array of local oids of remote tables for the next step. + * Also builds array of local oids of remote relations for the next + * step. */ off = 0; - pubrel_local_oids = palloc(list_length(pubrel_names) * sizeof(Oid)); + pubrel_local_oids = palloc(list_length(pubrels) * sizeof(Oid)); - foreach(lc, pubrel_names) + foreach_ptr(SubscriptionRelKind, relinfo, pubrels) { - RangeVar *rv = (RangeVar *) lfirst(lc); + RangeVar *rv = relinfo->rv; Oid relid; + char relkind; + bool pubisseq; + bool subisseq; relid = RangeVarGetRelid(rv, AccessShareLock, false); /* Check for supported relkind. */ - CheckSubscriptionRelkind(get_rel_relkind(relid), - rv->schemaname, rv->relname); + relkind = get_rel_relkind(relid); + CheckSubscriptionRelkind(relkind, rv->schemaname, rv->relname); pubrel_local_oids[off++] = relid; + pubisseq = (relinfo->relkind == RELKIND_SEQUENCE); + subisseq = (relkind == RELKIND_SEQUENCE); + + /* + * Allow RELKIND_RELATION and RELKIND_PARTITIONED_TABLE to be + * treated interchangeably, but ensure that sequences + * (RELKIND_SEQUENCE) match exactly on both publisher and + * subscriber. + */ + if (pubisseq != subisseq) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("relation \"%s.%s\" has relkind \"%c\" on the publisher but relkind \"%c\" on the subscriber", + rv->schemaname, rv->relname, relinfo->relkind, relkind)); + if (!bsearch(&relid, subrel_local_oids, subrel_count, sizeof(Oid), oid_cmp)) { @@ -968,28 +1015,29 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY, InvalidXLogRecPtr, true); ereport(DEBUG1, - (errmsg_internal("table \"%s.%s\" added to subscription \"%s\"", - rv->schemaname, rv->relname, sub->name))); + errmsg_internal("%s \"%s.%s\" added to subscription \"%s\"", + relkind == RELKIND_SEQUENCE ? "sequence" : "table", + rv->schemaname, rv->relname, sub->name)); } } /* - * Next remove state for tables we should not care about anymore using - * the data we collected above + * Next remove state for relations we should not care about anymore + * using the data we collected above */ - qsort(pubrel_local_oids, list_length(pubrel_names), + qsort(pubrel_local_oids, list_length(pubrels), sizeof(Oid), oid_cmp); - remove_rel_len = 0; for (off = 0; off < subrel_count; off++) { Oid relid = subrel_local_oids[off]; if (!bsearch(&relid, pubrel_local_oids, - list_length(pubrel_names), sizeof(Oid), oid_cmp)) + list_length(pubrels), sizeof(Oid), oid_cmp)) { char state; XLogRecPtr statelsn; + char relkind = get_rel_relkind(relid); /* * Lock pg_subscription_rel with AccessExclusiveLock to @@ -1011,41 +1059,55 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, /* Last known rel state. */ state = GetSubscriptionRelState(sub->oid, relid, &statelsn); - sub_remove_rels[remove_rel_len].relid = relid; - sub_remove_rels[remove_rel_len++].state = state; - RemoveSubscriptionRel(sub->oid, relid); - logicalrep_worker_stop(sub->oid, relid); - /* - * For READY state, we would have already dropped the - * tablesync origin. + * A single sequencesync worker synchronizes all sequences, so + * only stop workers when relation kind is not sequence. */ - if (state != SUBREL_STATE_READY) + if (relkind != RELKIND_SEQUENCE) { - char originname[NAMEDATALEN]; + SubRemoveRels *rel = palloc(sizeof(SubRemoveRels)); + + rel->relid = relid; + rel->state = state; + + sub_remove_rels = lappend(sub_remove_rels, rel); + + logicalrep_worker_stop(sub->oid, relid); /* - * Drop the tablesync's origin tracking if exists. - * - * It is possible that the origin is not yet created for - * tablesync worker, this can happen for the states before - * SUBREL_STATE_FINISHEDCOPY. The tablesync worker or - * apply worker can also concurrently try to drop the - * origin and by this time the origin might be already - * removed. For these reasons, passing missing_ok = true. + * For READY state, we would have already dropped the + * tablesync origin. */ - ReplicationOriginNameForLogicalRep(sub->oid, relid, originname, - sizeof(originname)); - replorigin_drop_by_name(originname, true, false); + if (state != SUBREL_STATE_READY) + { + char originname[NAMEDATALEN]; + + /* + * Drop the tablesync's origin tracking if exists. + * + * It is possible that the origin is not yet created + * for tablesync worker, this can happen for the + * states before SUBREL_STATE_FINISHEDCOPY. The + * tablesync worker or apply worker can also + * concurrently try to drop the origin and by this + * time the origin might be already removed. For these + * reasons, passing missing_ok = true. + */ + ReplicationOriginNameForLogicalRep(sub->oid, relid, + originname, + sizeof(originname)); + replorigin_drop_by_name(originname, true, false); + } } ereport(DEBUG1, - (errmsg_internal("table \"%s.%s\" removed from subscription \"%s\"", - get_namespace_name(get_rel_namespace(relid)), - get_rel_name(relid), - sub->name))); + errmsg_internal("%s \"%s.%s\" removed from subscription \"%s\"", + relkind == RELKIND_SEQUENCE ? "sequence" : "table", + get_namespace_name(get_rel_namespace(relid)), + get_rel_name(relid), + sub->name)); } } @@ -1054,10 +1116,10 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, * 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. */ - for (off = 0; off < remove_rel_len; off++) + foreach_ptr(SubRemoveRels, rel, sub_remove_rels) { - if (sub_remove_rels[off].state != SUBREL_STATE_READY && - sub_remove_rels[off].state != SUBREL_STATE_SYNCDONE) + if (rel->state != SUBREL_STATE_READY && + rel->state != SUBREL_STATE_SYNCDONE) { char syncslotname[NAMEDATALEN] = {0}; @@ -1071,11 +1133,13 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, * dropped slots and fail. For these reasons, we allow * missing_ok = true for the drop. */ - ReplicationSlotNameForTablesync(sub->oid, sub_remove_rels[off].relid, + ReplicationSlotNameForTablesync(sub->oid, rel->relid, syncslotname, sizeof(syncslotname)); ReplicationSlotDropAtPubNode(wrconn, syncslotname, true); } } + + list_free_deep(sub_remove_rels); } PG_FINALLY(); { @@ -1087,6 +1151,30 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, table_close(rel, NoLock); } +/* + * Marks all sequences with DATASYNC state. + */ +static void +AlterSubscription_refresh_seq(Subscription *sub) +{ + List *subrel_states; + + /* Get local relation list. */ + subrel_states = GetSubscriptionRelations(sub->oid, false, true, false); + foreach_ptr(SubscriptionRelState, subrel, subrel_states) + { + Oid relid = subrel->relid; + + UpdateSubscriptionRelState(sub->oid, relid, SUBREL_STATE_DATASYNC, + InvalidXLogRecPtr, false); + ereport(DEBUG1, + errmsg_internal("sequence \"%s.%s\" of subscription \"%s\" set to DATASYNC state", + get_namespace_name(get_rel_namespace(relid)), + get_rel_name(relid), + sub->name)); + } +} + /* * Common checks for altering failover, two_phase, and retain_dead_tuples * options. @@ -1722,6 +1810,18 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, break; } + case ALTER_SUBSCRIPTION_REFRESH_PUBLICATION_SEQ: + { + if (!sub->enabled) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("ALTER SUBSCRIPTION ... REFRESH PUBLICATION SEQUENCES is not allowed for disabled subscriptions")); + + AlterSubscription_refresh_seq(sub); + + break; + } + case ALTER_SUBSCRIPTION_SKIP: { parse_subscription_options(pstate, stmt->options, SUBOPT_LSN, &opts); @@ -1997,7 +2097,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) * the apply and tablesync workers and they can't restart because of * exclusive lock on the subscription. */ - rstates = GetSubscriptionRelations(subid, true); + rstates = GetSubscriptionRelations(subid, true, false, true); foreach(lc, rstates) { SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); @@ -2393,11 +2493,15 @@ check_publications_origin(WalReceiverConn *wrconn, List *publications, for (i = 0; i < subrel_count; i++) { Oid relid = subrel_local_oids[i]; - char *schemaname = get_namespace_name(get_rel_namespace(relid)); - char *tablename = get_rel_name(relid); - appendStringInfo(&cmd, "AND NOT (N.nspname = '%s' AND C.relname = '%s')\n", - schemaname, tablename); + if (get_rel_relkind(relid) != RELKIND_SEQUENCE) + { + char *schemaname = get_namespace_name(get_rel_namespace(relid)); + char *tablename = get_rel_name(relid); + + appendStringInfo(&cmd, "AND NOT (N.nspname = '%s' AND C.relname = '%s')\n", + schemaname, tablename); + } } } @@ -2583,8 +2687,23 @@ CheckSubDeadTupleRetention(bool check_guc, bool sub_disabled, } /* - * Get the list of tables which belong to specified publications on the - * publisher connection. + * Return true iff 'rv' is a member of the list. + */ +static bool +list_member_rangevar(const List *list, RangeVar *rv) +{ + foreach_ptr(SubscriptionRelKind, relinfo, list) + { + if (equal(relinfo->rv, rv)) + return true; + } + + return false; +} + +/* + * Get the list of tables and sequences which belong to specified publications + * on the publisher connection. * * Note that we don't support the case where the column list is different for * the same table in different publications to avoid sending unwanted column @@ -2592,15 +2711,17 @@ CheckSubDeadTupleRetention(bool check_guc, bool sub_disabled, * list and row filter are specified for different publications. */ static List * -fetch_table_list(WalReceiverConn *wrconn, List *publications) +fetch_relation_list(WalReceiverConn *wrconn, List *publications) { WalRcvExecResult *res; StringInfoData cmd; TupleTableSlot *slot; - Oid tableRow[3] = {TEXTOID, TEXTOID, InvalidOid}; + Oid tableRow[4] = {TEXTOID, TEXTOID, InvalidOid, CHAROID}; List *tablelist = NIL; int server_version = walrcv_server_version(wrconn); bool check_columnlist = (server_version >= 150000); + bool check_relkind = (server_version >= 190000); + int column_count = check_columnlist ? (check_relkind ? 4 : 3) : 2; StringInfo pub_names = makeStringInfo(); initStringInfo(&cmd); @@ -2608,8 +2729,25 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications) /* Build the pub_names comma-separated string. */ GetPublicationsStr(publications, pub_names, true); - /* Get the list of tables from the publisher. */ - if (server_version >= 160000) + /* Get the list of tables and sequences from the publisher. */ + if (server_version >= 190000) + { + tableRow[2] = INT2VECTOROID; + + appendStringInfo(&cmd, "SELECT DISTINCT n.nspname, c.relname, gpt.attrs, c.relkind\n" + " FROM pg_class c\n" + " JOIN pg_namespace n ON n.oid = c.relnamespace\n" + " JOIN ( SELECT (pg_get_publication_tables(VARIADIC array_agg(pubname::text))).*\n" + " FROM pg_publication\n" + " WHERE pubname IN (%s)) AS gpt\n" + " ON gpt.relid = c.oid\n" + " UNION ALL\n" + " SELECT DISTINCT s.schemaname, s.sequencename, NULL::int2vector AS attrs, 'S'::\"char\" AS relkind\n" + " FROM pg_catalog.pg_publication_sequences s\n" + " WHERE s.pubname IN (%s)", + pub_names->data, pub_names->data); + } + else if (server_version >= 160000) { tableRow[2] = INT2VECTOROID; @@ -2638,7 +2776,7 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications) else { tableRow[2] = NAMEARRAYOID; - appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename \n"); + appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename\n"); /* Get column lists for each relation if the publisher supports it */ if (check_columnlist) @@ -2651,7 +2789,7 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications) destroyStringInfo(pub_names); - res = walrcv_exec(wrconn, cmd.data, check_columnlist ? 3 : 2, tableRow); + res = walrcv_exec(wrconn, cmd.data, column_count, tableRow); pfree(cmd.data); if (res->status != WALRCV_OK_TUPLES) @@ -2667,22 +2805,31 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications) char *nspname; char *relname; bool isnull; - RangeVar *rv; + char relkind = RELKIND_RELATION; + SubscriptionRelKind *relinfo = (SubscriptionRelKind *) palloc(sizeof(SubscriptionRelKind)); nspname = TextDatumGetCString(slot_getattr(slot, 1, &isnull)); Assert(!isnull); relname = TextDatumGetCString(slot_getattr(slot, 2, &isnull)); Assert(!isnull); - rv = makeRangeVar(nspname, relname, -1); + if (check_relkind) + { + relkind = DatumGetChar(slot_getattr(slot, 4, &isnull)); + Assert(!isnull); + } + + relinfo->rv = makeRangeVar(nspname, relname, -1); + relinfo->relkind = relkind; - if (check_columnlist && list_member(tablelist, rv)) + if (relkind != RELKIND_SEQUENCE && check_columnlist && + list_member_rangevar(tablelist, relinfo->rv)) ereport(ERROR, errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("cannot use different column lists for table \"%s.%s\" in different publications", nspname, relname)); else - tablelist = lappend(tablelist, rv); + tablelist = lappend(tablelist, relinfo); ExecClearTuple(slot); } diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c index b409d4ecbf5..4f0f8a38555 100644 --- a/src/backend/executor/execReplication.c +++ b/src/backend/executor/execReplication.c @@ -1120,7 +1120,9 @@ void CheckSubscriptionRelkind(char relkind, const char *nspname, const char *relname) { - if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE) + if (relkind != RELKIND_RELATION && + relkind != RELKIND_PARTITIONED_TABLE && + relkind != RELKIND_SEQUENCE) ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE), errmsg("cannot use relation \"%s.%s\" as logical replication target", diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index 48104c22c4b..a359f3e293e 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -10988,6 +10988,15 @@ AlterSubscriptionStmt: n->options = $6; $$ = (Node *) n; } + | ALTER SUBSCRIPTION name REFRESH PUBLICATION SEQUENCES + { + AlterSubscriptionStmt *n = + makeNode(AlterSubscriptionStmt); + + n->kind = ALTER_SUBSCRIPTION_REFRESH_PUBLICATION_SEQ; + n->subname = $3; + $$ = (Node *) n; + } | ALTER SUBSCRIPTION name ADD_P PUBLICATION name_list opt_definition { AlterSubscriptionStmt *n = diff --git a/src/backend/replication/logical/syncutils.c b/src/backend/replication/logical/syncutils.c index 5109b197805..45b6d429558 100644 --- a/src/backend/replication/logical/syncutils.c +++ b/src/backend/replication/logical/syncutils.c @@ -152,8 +152,9 @@ FetchRelationStates(bool *started_tx) *started_tx = true; } - /* Fetch tables that are in non-ready state. */ - rstates = GetSubscriptionRelations(MySubscription->oid, true); + /* Fetch tables and sequences that are in non-ready state. */ + rstates = GetSubscriptionRelations(MySubscription->oid, true, true, + true); /* Allocate the tracking info in a permanent memory context. */ oldctx = MemoryContextSwitchTo(CacheMemoryContext); diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 3c777363243..a2ba0cef007 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -840,7 +840,7 @@ fetch_remote_table_info(char *nspname, char *relname, LogicalRepRelation *lrel, /* * We don't support the case where the column list is different for * the same table when combining publications. See comments atop - * fetch_table_list. So there should be only one row returned. + * fetch_relation_list. So there should be only one row returned. * Although we already checked this when creating the subscription, we * still need to check here in case the column list was changed after * creating the subscription and before the sync worker is started. diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 80540c017bd..d708f3b0266 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -1114,7 +1114,7 @@ pgoutput_column_list_init(PGOutputData *data, List *publications, * * Note that we don't support the case where the column list is different * for the same table when combining publications. See comments atop - * fetch_table_list. But one can later change the publication so we still + * fetch_relation_list. But one can later change the publication so we still * need to check all the given publication-table mappings and report an * error if any publications have a different column list. */ diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c index ec559146640..4a638fbecc9 100644 --- a/src/bin/psql/tab-complete.in.c +++ b/src/bin/psql/tab-complete.in.c @@ -2312,7 +2312,7 @@ match_previous_words(int pattern_id, "ADD PUBLICATION", "DROP PUBLICATION"); /* ALTER SUBSCRIPTION REFRESH PUBLICATION */ else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "REFRESH", "PUBLICATION")) - COMPLETE_WITH("WITH ("); + COMPLETE_WITH("SEQUENCES", "WITH ("); /* ALTER SUBSCRIPTION REFRESH PUBLICATION WITH ( */ else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "REFRESH", "PUBLICATION", "WITH", "(")) COMPLETE_WITH("copy_data"); diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 62bcd9d921c..4660e42d775 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -12282,6 +12282,11 @@ proargmodes => '{v,o,o,o,o}', proargnames => '{pubname,pubid,relid,attrs,qual}', prosrc => 'pg_get_publication_tables' }, +{ oid => '8052', descr => 'get OIDs of sequences in a publication', + proname => 'pg_get_publication_sequences', prorows => '1000', proretset => 't', + provolatile => 's', prorettype => 'oid', proargtypes => 'text', + proallargtypes => '{text,oid}', proargmodes => '{i,o}', + proargnames => '{pubname,relid}', prosrc => 'pg_get_publication_sequences' }, { oid => '6121', descr => 'returns whether a relation can be part of a publication', proname => 'pg_relation_is_publishable', provolatile => 's', diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h index 24e09c76649..22f48bb8975 100644 --- a/src/include/catalog/pg_publication.h +++ b/src/include/catalog/pg_publication.h @@ -170,7 +170,7 @@ typedef enum PublicationPartOpt extern List *GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt); extern List *GetAllTablesPublications(void); -extern List *GetAllTablesPublicationRelations(bool pubviaroot); +extern List *GetAllPublicationRelations(char relkind, bool pubviaroot); extern List *GetPublicationSchemas(Oid pubid); extern List *GetSchemaPublications(Oid schemaid); extern List *GetSchemaPublicationRelations(Oid schemaid, diff --git a/src/include/catalog/pg_subscription_rel.h b/src/include/catalog/pg_subscription_rel.h index 61b63c6bb7a..3d6e31a0d6c 100644 --- a/src/include/catalog/pg_subscription_rel.h +++ b/src/include/catalog/pg_subscription_rel.h @@ -22,6 +22,7 @@ #include "catalog/genbki.h" #include "catalog/pg_subscription_rel_d.h" /* IWYU pragma: export */ #include "nodes/pg_list.h" +#include "nodes/primnodes.h" /* ---------------- * pg_subscription_rel definition. cpp turns this into @@ -82,6 +83,12 @@ typedef struct SubscriptionRelState char state; } SubscriptionRelState; +typedef struct SubscriptionRelKind +{ + RangeVar *rv; + char relkind; +} SubscriptionRelKind; + extern void AddSubscriptionRelState(Oid subid, Oid relid, char state, XLogRecPtr sublsn, bool retain_lock); extern void UpdateSubscriptionRelState(Oid subid, Oid relid, char state, @@ -90,7 +97,9 @@ extern char GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn); extern void RemoveSubscriptionRel(Oid subid, Oid relid); extern bool HasSubscriptionTables(Oid subid); -extern List *GetSubscriptionRelations(Oid subid, bool not_ready); +extern List *GetSubscriptionRelations(Oid subid, bool get_tables, + bool get_sequences, + bool not_ready); extern void UpdateDeadTupleRetentionStatus(Oid subid, bool active); diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index 68ee5670124..8d8487c2454 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -4360,6 +4360,7 @@ typedef enum AlterSubscriptionType ALTER_SUBSCRIPTION_ADD_PUBLICATION, ALTER_SUBSCRIPTION_DROP_PUBLICATION, ALTER_SUBSCRIPTION_REFRESH_PUBLICATION, + ALTER_SUBSCRIPTION_REFRESH_PUBLICATION_SEQ, ALTER_SUBSCRIPTION_ENABLED, ALTER_SUBSCRIPTION_SKIP, } AlterSubscriptionType; diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 35e8aad7701..4e2d6b693c6 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1462,6 +1462,14 @@ pg_prepared_xacts| SELECT p.transaction, FROM ((pg_prepared_xact() p(transaction, gid, prepared, ownerid, dbid) LEFT JOIN pg_authid u ON ((p.ownerid = u.oid))) LEFT JOIN pg_database d ON ((p.dbid = d.oid))); +pg_publication_sequences| SELECT p.pubname, + n.nspname AS schemaname, + c.relname AS sequencename + FROM pg_publication p, + LATERAL pg_get_publication_sequences((p.pubname)::text) gps(relid), + (pg_class c + JOIN pg_namespace n ON ((n.oid = c.relnamespace))) + WHERE (c.oid = gps.relid); pg_publication_tables| SELECT p.pubname, n.nspname AS schemaname, c.relname AS tablename, diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index a7ff6601054..a3f02884404 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2899,6 +2899,7 @@ SubscriptingRef SubscriptingRefState Subscription SubscriptionInfo +SubscriptionRelKind SubscriptionRelState SummarizerReadLocalXLogPrivate SupportRequestCost -- 2.43.0