From 6d8c10d48ab2ac7dd9fea9a34f348202bac12e21 Mon Sep 17 00:00:00 2001 From: Ajin Cherian Date: Wed, 3 Mar 2021 04:58:18 -0500 Subject: [PATCH v47] Add two_phase option to CREATE REPLICATION SLOT. This patch adds new option to enable two_phase while creating a slot. --- src/backend/commands/subscriptioncmds.c | 2 +- .../replication/libpqwalreceiver/libpqwalreceiver.c | 6 +++++- src/backend/replication/logical/tablesync.c | 2 +- src/backend/replication/repl_gram.y | 14 +++++++++++--- src/backend/replication/repl_scanner.l | 1 + src/backend/replication/walreceiver.c | 2 +- src/include/replication/walreceiver.h | 5 +++-- 7 files changed, 23 insertions(+), 9 deletions(-) diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index bfd3514..f6793f0 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -528,7 +528,7 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) { Assert(slotname); - walrcv_create_slot(wrconn, slotname, false, + walrcv_create_slot(wrconn, slotname, false, true, CRS_NOEXPORT_SNAPSHOT, NULL); ereport(NOTICE, (errmsg("created replication slot \"%s\" on publisher", diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index 5272eed..9e822f9 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -73,6 +73,7 @@ static void libpqrcv_send(WalReceiverConn *conn, const char *buffer, static char *libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname, bool temporary, + bool two_phase, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn); static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn); @@ -827,7 +828,7 @@ libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes) */ static char * libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname, - bool temporary, CRSSnapshotAction snapshot_action, + bool temporary, bool two_phase, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn) { PGresult *res; @@ -841,6 +842,9 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname, if (temporary) appendStringInfoString(&cmd, " TEMPORARY"); + if (two_phase) + appendStringInfoString(&cmd, " TWO_PHASE"); + if (conn->logical) { appendStringInfoString(&cmd, " LOGICAL pgoutput"); diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index feb634e..50c3ea7 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -1052,7 +1052,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos) * for the catchup phase after COPY is done, so tell it to use the * snapshot to make the final data consistent. */ - walrcv_create_slot(wrconn, slotname, false /* permanent */ , + walrcv_create_slot(wrconn, slotname, false /* permanent */ , false /* two_phase */, CRS_USE_SNAPSHOT, origin_startpos); /* diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y index eb283a8..c5154ae 100644 --- a/src/backend/replication/repl_gram.y +++ b/src/backend/replication/repl_gram.y @@ -84,6 +84,7 @@ static SQLCmd *make_sqlcmd(void); %token K_SLOT %token K_RESERVE_WAL %token K_TEMPORARY +%token K_TWO_PHASE %token K_EXPORT_SNAPSHOT %token K_NOEXPORT_SNAPSHOT %token K_USE_SNAPSHOT @@ -102,6 +103,7 @@ static SQLCmd *make_sqlcmd(void); %type plugin_opt_arg %type opt_slot var_name %type opt_temporary +%type opt_two_phase %type create_slot_opt_list %type create_slot_opt @@ -242,15 +244,16 @@ create_replication_slot: $$ = (Node *) cmd; } /* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin */ - | K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_opt_list + | K_CREATE_REPLICATION_SLOT IDENT opt_temporary opt_two_phase K_LOGICAL IDENT create_slot_opt_list { CreateReplicationSlotCmd *cmd; cmd = makeNode(CreateReplicationSlotCmd); cmd->kind = REPLICATION_KIND_LOGICAL; cmd->slotname = $2; cmd->temporary = $3; - cmd->plugin = $5; - cmd->options = $6; + cmd->two_phase = $4; + cmd->plugin = $6; + cmd->options = $7; $$ = (Node *) cmd; } ; @@ -365,6 +368,11 @@ opt_temporary: | /* EMPTY */ { $$ = false; } ; +opt_two_phase: + K_TWO_PHASE { $$ = true; } + | /* EMPTY */ { $$ = false; } + ; + opt_slot: K_SLOT IDENT { $$ = $2; } diff --git a/src/backend/replication/repl_scanner.l b/src/backend/replication/repl_scanner.l index dcc3c3f..c038a63 100644 --- a/src/backend/replication/repl_scanner.l +++ b/src/backend/replication/repl_scanner.l @@ -103,6 +103,7 @@ RESERVE_WAL { return K_RESERVE_WAL; } LOGICAL { return K_LOGICAL; } SLOT { return K_SLOT; } TEMPORARY { return K_TEMPORARY; } +TWO_PHASE { return K_TWO_PHASE; } EXPORT_SNAPSHOT { return K_EXPORT_SNAPSHOT; } NOEXPORT_SNAPSHOT { return K_NOEXPORT_SNAPSHOT; } USE_SNAPSHOT { return K_USE_SNAPSHOT; } diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c index e5f8a06..e40d2d0 100644 --- a/src/backend/replication/walreceiver.c +++ b/src/backend/replication/walreceiver.c @@ -363,7 +363,7 @@ WalReceiverMain(void) "pg_walreceiver_%lld", (long long int) walrcv_get_backend_pid(wrconn)); - walrcv_create_slot(wrconn, slotname, true, 0, NULL); + walrcv_create_slot(wrconn, slotname, true, false, 0, NULL); SpinLockAcquire(&walrcv->mutex); strlcpy(walrcv->slotname, slotname, NAMEDATALEN); diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h index a97a59a..f55b07c 100644 --- a/src/include/replication/walreceiver.h +++ b/src/include/replication/walreceiver.h @@ -345,6 +345,7 @@ typedef void (*walrcv_send_fn) (WalReceiverConn *conn, typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn, const char *slotname, bool temporary, + bool two_phase, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn); @@ -418,8 +419,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions; WalReceiverFunctions->walrcv_receive(conn, buffer, wait_fd) #define walrcv_send(conn, buffer, nbytes) \ WalReceiverFunctions->walrcv_send(conn, buffer, nbytes) -#define walrcv_create_slot(conn, slotname, temporary, snapshot_action, lsn) \ - WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, snapshot_action, lsn) +#define walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn) \ + WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn) #define walrcv_get_backend_pid(conn) \ WalReceiverFunctions->walrcv_get_backend_pid(conn) #define walrcv_exec(conn, exec, nRetTypes, retTypes) \ -- 1.8.3.1