From 719ac03dd963502381fbc210947faae472b4a95a Mon Sep 17 00:00:00 2001 From: Petr Jelinek Date: Sun, 30 Dec 2018 02:07:26 +0100 Subject: [PATCH 2/2] Add option for filtering which slots get synchronized --- .../libpqwalreceiver/libpqwalreceiver.c | 32 +++++- src/backend/replication/logical/launcher.c | 3 +- src/backend/replication/logical/slotsync.c | 104 +++++++++++++++++- src/backend/replication/repl_gram.y | 23 +++- src/backend/replication/walsender.c | 37 ++++++- src/backend/utils/misc/guc.c | 12 ++ src/include/nodes/replnodes.h | 1 + src/include/replication/logicalworker.h | 9 ++ src/include/replication/walreceiver.h | 6 +- src/include/replication/worker_internal.h | 4 + 10 files changed, 216 insertions(+), 15 deletions(-) diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index 3c77a76ea6..f564f288d0 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -59,7 +59,8 @@ static void libpqrcv_get_senderinfo(WalReceiverConn *conn, static char *libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli, int *server_version); -static List *libpqrcv_list_slots(WalReceiverConn *conn); +static List *libpqrcv_list_slots(WalReceiverConn *conn, int nslot_names, + NameData *slot_names); static void libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn, TimeLineID tli, char **filename, char **content, int *len); @@ -355,15 +356,34 @@ libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli, * Get list of slots from primary. */ static List * -libpqrcv_list_slots(WalReceiverConn *conn) +libpqrcv_list_slots(WalReceiverConn *conn, int nslot_names, + NameData *slot_names) { PGresult *res; int i; - List *slots = NIL; + List *slotlist = NIL; int ntuples; + StringInfoData s; WalRecvReplicationSlotData *slot_data; - res = libpqrcv_PQexec(conn->streamConn, "LIST_SLOTS"); + initStringInfo(&s); + appendStringInfoString(&s, "LIST_SLOTS"); + if (nslot_names > 0) + { + int i; + + appendStringInfoChar(&s, ' '); + for (i = 0; i < nslot_names; i++) + { + if (i > 0) + appendStringInfoChar(&s, ','); + appendStringInfo(&s, "%s", + quote_identifier(NameStr(slot_names[i]))); + } + } + + res = libpqrcv_PQexec(conn->streamConn, s.data); + pfree(s.data); if (PQresultStatus(res) != PGRES_TUPLES_OK) { PQclear(res); @@ -414,12 +434,12 @@ libpqrcv_list_slots(WalReceiverConn *conn) slot_data->confirmed_flush = pg_strtouint64(PQgetvalue(res, i, 9), NULL, 10); - slots = lappend(slots, slot_data); + slotlist = lappend(slotlist, slot_data); } PQclear(res); - return slots; + return slotlist; } /* diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index c47d4ee403..161205d9e4 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -1005,7 +1005,8 @@ ApplyLauncherStartSlotSync(TimestampTz *last_start_time, long *wait_time) ALLOCSET_DEFAULT_SIZES); oldctx = MemoryContextSwitchTo(tmpctx); - slots = walrcv_list_slots(wrconn); + slots = walrcv_list_slots(wrconn, numsynchronize_slot_names, + synchronize_slot_names); now = GetCurrentTimestamp(); diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index eda9ad0add..07f06a8fe9 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -26,7 +26,11 @@ #include "utils/builtins.h" #include "utils/guc.h" #include "utils/pg_lsn.h" +#include "utils/varlena.h" +char *synchronize_slot_names_string; +NameData *synchronize_slot_names; +int numsynchronize_slot_names; /* * Wait for remote slot to pass localy reserved position. @@ -215,12 +219,26 @@ synchronize_slots(void) (errmsg("could not connect to the primary server: %s", err))); resetStringInfo(&s); - /* TODO filter slot names? */ appendStringInfo(&s, "SELECT slot_name, plugin, confirmed_flush_lsn" " FROM pg_catalog.pg_replication_slots" " WHERE database = %s", quote_literal_cstr(database)); + if (numsynchronize_slot_names > 0) + { + int i; + + appendStringInfoString(&s, " AND slot_name IN ("); + for (i = 0; i < numsynchronize_slot_names; i++) + { + if (i > 0) + appendStringInfoChar(&s, ','); + appendStringInfo(&s, "%s", + quote_literal_cstr(NameStr(synchronize_slot_names[i]))); + } + appendStringInfoChar(&s, ')'); + } + res = walrcv_exec(wrconn, s.data, 3, slotRow); pfree(s.data); @@ -300,6 +318,9 @@ ReplSlotSyncMain(Datum main_arg) if (!RecoveryInProgress()) return; + if (numsynchronize_slot_names == 0) + return; + synchronize_slots(); rc = WaitLatch(MyLatch, @@ -314,3 +335,84 @@ ReplSlotSyncMain(Datum main_arg) proc_exit(1); } } + +/* + * Routines for handling the GUC variable(s) + */ + +typedef struct +{ + int numslots; + NameData slots[FLEXIBLE_ARRAY_MEMBER]; +} synchronize_slot_names_extra; + +bool +check_synchronize_slot_names(char **newval, void **extra, GucSource source) +{ + char *rawname; + List *namelist; + ListCell *lc; + synchronize_slot_names_extra *myextra; + NameData *slots; + int numslots; + + /* Need a modifiable copy of string */ + rawname = pstrdup(*newval); + + /* Parse string into list of identifiers */ + if (!SplitIdentifierString(rawname, ',', &namelist)) + { + /* syntax error in name list */ + GUC_check_errdetail("List syntax is invalid."); + pfree(rawname); + list_free(namelist); + return false; + } + + /* temporary workspace until we are done verifying the list */ + slots = (NameData *) palloc(list_length(namelist) * sizeof(NameData)); + numslots = 0; + foreach(lc, namelist) + { + char *curname = (char *) lfirst(lc); + + /* Special handling for "*" which means all. */ + if (strcmp(curname, "*") == 0) + { + numslots = -1; + break; + } + + /* For any other value, validate slot name. */ + ReplicationSlotValidateName(curname, ERROR); + + /* And add it to our array. */ + namestrcpy(&slots[numslots++], curname); + } + + /* Now prepare an "extra" struct for assign_temp_tablespaces */ + myextra = malloc(offsetof(synchronize_slot_names_extra, slots) + + numslots > 0 ? numslots * sizeof(NameData) : 0); + if (!myextra) + return false; + myextra->numslots = numslots; + if (numslots > 0) + memcpy(myextra->slots, slots, numslots * sizeof(NameData)); + *extra = (void *) myextra; + + pfree(slots); + pfree(rawname); + list_free(namelist); + + return true; +} + +void +assign_synchronize_slot_names(const char *newval, void *extra) +{ + synchronize_slot_names_extra *myextra = + (synchronize_slot_names_extra *) extra; + + synchronize_slot_names = myextra->slots; + numsynchronize_slot_names = myextra->numslots; +} diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y index 95d91bf236..f0988aed1c 100644 --- a/src/backend/replication/repl_gram.y +++ b/src/backend/replication/repl_gram.y @@ -103,6 +103,7 @@ static SQLCmd *make_sqlcmd(void); %type opt_temporary %type create_slot_opt_list %type create_slot_opt +%type slot_name_list slot_name_list_opt %% @@ -138,13 +139,31 @@ identify_system: $$ = (Node *) makeNode(IdentifySystemCmd); } ; + +slot_name_list: + IDENT + { + $$ = list_make1($1); + } + | slot_name_list ',' IDENT + { + $$ = lappend($1, $3); + } + +slot_name_list_opt: + slot_name_list { $$ = $1; } + | /* EMPTY */ { $$ = NIL; } + ; + /* * LIST_SLOTS */ list_slots: - K_LIST_SLOTS + K_LIST_SLOTS slot_name_list_opt { - $$ = (Node *) makeNode(ListSlotsCmd); + ListSlotsCmd *cmd = makeNode(ListSlotsCmd); + cmd->slot_names = $2; + $$ = (Node *) cmd; } ; diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 6ca304b8f5..bb9cec46a8 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -426,16 +426,41 @@ IdentifySystem(void) end_tup_output(tstate); } +static int +pg_qsort_namecmp(const void *a, const void *b) +{ + return strncmp(NameStr(*(Name)a), NameStr(*(Name)b), NAMEDATALEN); +} /* * Handle the LIST_SLOTS command. */ static void -ListSlots(void) +ListSlots(ListSlotsCmd *cmd) { DestReceiver *dest; TupOutputState *tstate; TupleDesc tupdesc; int slotno; + NameData *slot_names; + int numslot_names; + + numslot_names = list_length(cmd->slot_names); + if (numslot_names) + { + ListCell *lc; + int i = 0; + + slot_names = palloc(numslot_names * sizeof(NameData)); + foreach (lc, cmd->slot_names) + { + char *slot_name = lfirst(lc); + + ReplicationSlotValidateName(slot_name, ERROR); + namestrcpy(&slot_names[i++], slot_name); + } + + qsort(slot_names, numslot_names, sizeof(NameData), pg_qsort_namecmp); + } dest = CreateDestReceiver(DestRemoteSimple); @@ -501,6 +526,11 @@ ListSlots(void) SpinLockRelease(&slot->mutex); + if (numslot_names && + !bsearch((void *) &slot_name, (void *) slot_names, + numslot_names, sizeof(NameData), pg_qsort_namecmp)) + continue; + memset(nulls, 0, sizeof(nulls)); i = 0; @@ -1712,7 +1742,10 @@ exec_replication_command(const char *cmd_string) break; case T_ListSlotsCmd: - ListSlots(); + { + ListSlotsCmd *cmd = (ListSlotsCmd *) cmd_node; + ListSlots(cmd); + } break; case T_SQLCmd: diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index 6fe1939881..338232920b 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -64,6 +64,7 @@ #include "postmaster/syslogger.h" #include "postmaster/walwriter.h" #include "replication/logicallauncher.h" +#include "replication/logicalworker.h" #include "replication/slot.h" #include "replication/syncrep.h" #include "replication/walreceiver.h" @@ -4078,6 +4079,17 @@ static struct config_string ConfigureNamesString[] = NULL, NULL, NULL }, + { + {"synchronize_slot_names", PGC_SIGHUP, REPLICATION_STANDBY, + gettext_noop("Sets the names of replication slots which to synchronize from primary to standby."), + gettext_noop("Value of \"*\" means all."), + GUC_LIST_INPUT | GUC_LIST_QUOTE + }, + &synchronize_slot_names_string, + "", + check_synchronize_slot_names, assign_synchronize_slot_names, NULL + }, + /* End-of-list marker */ { {NULL, 0, 0, NULL, NULL}, NULL, NULL, NULL, NULL, NULL diff --git a/src/include/nodes/replnodes.h b/src/include/nodes/replnodes.h index 1101f71bcc..71f8f8ca8e 100644 --- a/src/include/nodes/replnodes.h +++ b/src/include/nodes/replnodes.h @@ -40,6 +40,7 @@ typedef struct IdentifySystemCmd typedef struct ListSlotsCmd { NodeTag type; + List *slot_names; } ListSlotsCmd; /* ---------------------- diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h index 6379aae7ce..1b09a1fb3e 100644 --- a/src/include/replication/logicalworker.h +++ b/src/include/replication/logicalworker.h @@ -12,8 +12,17 @@ #ifndef LOGICALWORKER_H #define LOGICALWORKER_H +#include "utils/guc.h" + +extern char *synchronize_slot_names_string; + extern void ApplyWorkerMain(Datum main_arg); +extern void ReplSlotSyncMain(Datum main_arg); extern bool IsLogicalWorker(void); +extern bool check_synchronize_slot_names(char **newval, void **extra, GucSource source); +extern void assign_synchronize_slot_names(const char *newval, void *extra); + + #endif /* LOGICALWORKER_H */ diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h index 9014f92bfa..2593af2291 100644 --- a/src/include/replication/walreceiver.h +++ b/src/include/replication/walreceiver.h @@ -219,7 +219,7 @@ typedef void (*walrcv_get_senderinfo_fn) (WalReceiverConn *conn, typedef char *(*walrcv_identify_system_fn) (WalReceiverConn *conn, TimeLineID *primary_tli, int *server_version); -typedef List *(*walrcv_list_slots_fn) (WalReceiverConn *conn); +typedef List *(*walrcv_list_slots_fn) (WalReceiverConn *conn, int nslots, NameData *slots); typedef void (*walrcv_readtimelinehistoryfile_fn) (WalReceiverConn *conn, TimeLineID tli, char **filename, @@ -272,8 +272,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions; WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port) #define walrcv_identify_system(conn, primary_tli, server_version) \ WalReceiverFunctions->walrcv_identify_system(conn, primary_tli, server_version) -#define walrcv_list_slots(conn) \ - WalReceiverFunctions->walrcv_list_slots(conn) +#define walrcv_list_slots(conn, nslots, slots) \ + WalReceiverFunctions->walrcv_list_slots(conn, nslots, slots) #define walrcv_readtimelinehistoryfile(conn, tli, filename, content, size) \ WalReceiverFunctions->walrcv_readtimelinehistoryfile(conn, tli, filename, content, size) #define walrcv_startstreaming(conn, options) \ diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index ba2d3f1fca..c2373fea47 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -68,6 +68,10 @@ extern LogicalRepWorker *MyLogicalRepWorker; extern bool in_remote_transaction; +/* Translated Gucs we share with launcher and worker. */ +extern NameData *synchronize_slot_names; +extern int numsynchronize_slot_names; + extern void logicalrep_worker_attach(int slot); extern LogicalRepWorker *logicalrep_worker_find(Oid dbid, Oid subid, Oid relid, bool only_running); -- 2.17.1