database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-08-24 09:27:52 +03:00
Code Activity

Add decoding of sequences to built-in replication

Browse Source 
This commit adds support for decoding of sequences to the built-in
replication (the infrastructure was added by commit 0da92dc530).

The syntax and behavior mostly mimics handling of tables, i.e. a
publication may be defined as FOR ALL SEQUENCES (replicating all
sequences in a database), FOR ALL SEQUENCES IN SCHEMA (replicating
all sequences in a particular schema) or individual sequences.

To publish sequence modifications, the publication has to include
'sequence' action. The protocol is extended with a new message,
describing sequence increments.

A new system view pg_publication_sequences lists all the sequences
added to a publication, both directly and indirectly. Various psql
commands (\d and \dRp) are improved to also display publications
including a given sequence, or sequences included in a publication.

Author: Tomas Vondra, Cary Huang
Reviewed-by: Peter Eisentraut, Amit Kapila, Hannu Krosing, Andres
             Freund, Petr Jelinek
Discussion: https://postgr.es/m/d045f3c2-6cfb-06d3-5540-e63c320df8bc@enterprisedb.com
Discussion: https://postgr.es/m/1710ed7e13b.cd7177461430746.3372264562543607781@highgo.ca
This commit is contained in:
Tomas Vondra
2022-03-24 18:20:21 +01:00
parent 0adb3dc68b
commit 75b1521dae
40 changed files with 3236 additions and 469 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -90,6 +90,7 @@ typedef struct SubOpts
} SubOpts;
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
static List *fetch_sequence_list(WalReceiverConn *wrconn, List *publications);
static void check_duplicates_in_publist(List *publist, Datum *datums);
static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname);
static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err);
@@ -541,9 +542,9 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
{
char *err;
WalReceiverConn *wrconn;
List *tables;
List *relations;
ListCell *lc;
char table_state;
char sync_state;
/* Try to connect to the publisher. */
wrconn = walrcv_connect(conninfo, true, stmt->subname, &err);
@@ -558,14 +559,17 @@ 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;
sync_state = opts.copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY;
/*
* Get the table list from publisher and build local table status
* info.
* Get the table and sequence list from publisher and build
* local relation sync status info.
*/
tables = fetch_table_list(wrconn, publications);
foreach(lc, tables)
relations = fetch_table_list(wrconn, publications);
relations = list_concat(relations,
fetch_sequence_list(wrconn, publications));
foreach(lc, relations)
{
RangeVar *rv = (RangeVar *) lfirst(lc);
Oid relid;
@@ -576,7 +580,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
CheckSubscriptionRelkind(get_rel_relkind(relid),
rv->schemaname, rv->relname);
AddSubscriptionRelState(subid, relid, table_state,
AddSubscriptionRelState(subid, relid, sync_state,
InvalidXLogRecPtr);
}
@@ -602,12 +606,12 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
*
* Note that if tables were specified but copy_data is false
* then it is safe to enable two_phase up-front because those
* tables are already initially in READY state. When the
* subscription has no tables, we leave the twophase state as
* PENDING, to allow ALTER SUBSCRIPTION ... REFRESH
* relations are already initially in READY state. When the
* subscription has no relations, we leave the twophase state
* as PENDING, to allow ALTER SUBSCRIPTION ... REFRESH
* PUBLICATION to work.
*/
if (opts.twophase && !opts.copy_data && tables != NIL)
if (opts.twophase && !opts.copy_data && relations != NIL)
twophase_enabled = true;
walrcv_create_slot(wrconn, opts.slot_name, false, twophase_enabled,
@@ -677,8 +681,10 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data)
PG_TRY();
{
/* Get the table list from publisher. */
/* Get the list of relations from publisher. */
pubrel_names = fetch_table_list(wrconn, sub->publications);
pubrel_names = list_concat(pubrel_names,
fetch_sequence_list(wrconn, sub->publications));
/* Get local table list. */
subrel_states = GetSubscriptionRelations(sub->oid);
@@ -1712,6 +1718,75 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications)
return tablelist;
}
/*
* Get the list of sequences which belong to specified publications on the
* publisher connection.
*/
static List *
fetch_sequence_list(WalReceiverConn *wrconn, List *publications)
{
WalRcvExecResult *res;
StringInfoData cmd;
TupleTableSlot *slot;
Oid tableRow[2] = {TEXTOID, TEXTOID};
ListCell *lc;
bool first;
List *tablelist = NIL;
Assert(list_length(publications) > 0);
initStringInfo(&cmd);
appendStringInfoString(&cmd, "SELECT DISTINCT s.schemaname, s.sequencename\n"
" FROM pg_catalog.pg_publication_sequences s\n"
" WHERE s.pubname IN (");
first = true;
foreach(lc, publications)
{
char *pubname = strVal(lfirst(lc));
if (first)
first = false;
else
appendStringInfoString(&cmd, ", ");
appendStringInfoString(&cmd, quote_literal_cstr(pubname));
}
appendStringInfoChar(&cmd, ')');
res = walrcv_exec(wrconn, cmd.data, 2, tableRow);
pfree(cmd.data);
if (res->status != WALRCV_OK_TUPLES)
ereport(ERROR,
(errmsg("could not receive list of replicated sequences from the publisher: %s",
res->err)));
/* Process sequences. */
slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
{
char *nspname;
char *relname;
bool isnull;
RangeVar *rv;
nspname = TextDatumGetCString(slot_getattr(slot, 1, &isnull));
Assert(!isnull);
relname = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
Assert(!isnull);
rv = makeRangeVar(nspname, relname, -1);
tablelist = lappend(tablelist, rv);
ExecClearTuple(slot);
}
ExecDropSingleTupleTableSlot(slot);
walrcv_clear_result(res);
return tablelist;
}
/*
* This is to report the connection failure while dropping replication slots.
* Here, we report the WARNING for all tablesync slots so that user can drop