diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y index e9177ca..aed9420 100644 --- a/src/backend/replication/repl_gram.y +++ b/src/backend/replication/repl_gram.y @@ -76,11 +76,12 @@ Node *replication_parse_result; %token K_PHYSICAL %token K_LOGICAL %token K_SLOT +%token K_LOGICAL_DECODING_SNAPSHOT %type command %type base_backup start_replication start_logical_replication create_replication_slot drop_replication_slot identify_system - timeline_history + timeline_history logical_decoding_snapshot %type base_backup_opt_list %type base_backup_opt %type opt_timeline @@ -109,6 +110,7 @@ command: | create_replication_slot | drop_replication_slot | timeline_history + | logical_decoding_snapshot ; /* @@ -202,6 +204,19 @@ create_replication_slot: } ; +/* LOGICAL_DECODING_SNAPSHOT plugin */ +logical_decoding_snapshot: + K_LOGICAL_DECODING_SNAPSHOT IDENT IDENT + { + LogicalDecodingSnapshotCmd *cmd; + cmd = makeNode(LogicalDecodingSnapshotCmd); + cmd->slotname = $2; + cmd->plugin = $3; + $$ = (Node *) cmd; + } + ; + + /* DROP_REPLICATION_SLOT slot */ drop_replication_slot: K_DROP_REPLICATION_SLOT IDENT diff --git a/src/backend/replication/repl_scanner.l b/src/backend/replication/repl_scanner.l index 056cc14..5c42710 100644 --- a/src/backend/replication/repl_scanner.l +++ b/src/backend/replication/repl_scanner.l @@ -97,6 +97,7 @@ TIMELINE_HISTORY { return K_TIMELINE_HISTORY; } PHYSICAL { return K_PHYSICAL; } LOGICAL { return K_LOGICAL; } SLOT { return K_SLOT; } +LOGICAL_DECODING_SNAPSHOT { return K_LOGICAL_DECODING_SNAPSHOT; } "," { return ','; } ";" { return ';'; } diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index eb1b89b..3c330a2 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -200,6 +200,7 @@ static void WalSndDone(WalSndSendDataCallback send_data); static XLogRecPtr GetStandbyFlushRecPtr(void); static void IdentifySystem(void); static void CreateReplicationSlot(CreateReplicationSlotCmd *cmd); +static void LogicalDecodingSnapshot(LogicalDecodingSnapshotCmd *cmd); static void DropReplicationSlot(DropReplicationSlotCmd *cmd); static void StartReplication(StartReplicationCmd *cmd); static void StartLogicalReplication(StartReplicationCmd *cmd); @@ -911,6 +912,97 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd) ReplicationSlotRelease(); } + +/* + * Export a new snapshot for logical decoding + */ +static void +LogicalDecodingSnapshot(LogicalDecodingSnapshotCmd *cmd) +{ + const char *snapshot_name = NULL; + char xpos[MAXFNAMELEN]; + StringInfoData buf; + LogicalDecodingContext *ctx; + + Assert(!MyReplicationSlot); + + /* setup state for XLogReadPage */ + sendTimeLineIsHistoric = false; + sendTimeLine = ThisTimeLineID; + + CheckLogicalDecodingRequirements(); + + /* + * create the slot as ephemeral, it'll get dropped at the end. + */ + ReplicationSlotCreate(cmd->slotname, true, RS_EPHEMERAL); + + initStringInfo(&output_message); + + ctx = CreateInitDecodingContext(cmd->plugin, NIL, + logical_read_xlog_page, + WalSndPrepareWrite, WalSndWriteData); + + /* + * See CreateReplicationSlot(...) for logical decoding + */ + last_reply_timestamp = 0; + + /* build initial snapshot, might take a while */ + DecodingContextFindStartpoint(ctx); + + /* + * Export a plain (not of the snapbuild.c type) snapshot to the user + * that can be imported into another session. + */ + snapshot_name = SnapBuildExportSnapshot(ctx->snapshot_builder); + + /* don't need the decoding context anymore */ + FreeDecodingContext(ctx); + + snprintf(xpos, sizeof(xpos), "%X/%X", + (uint32) (MyReplicationSlot->data.confirmed_flush >> 32), + (uint32) MyReplicationSlot->data.confirmed_flush); + + pq_beginmessage(&buf, 'T'); + pq_sendint(&buf, 2, 2); /* 2 fields */ + + /* first field: LSN at which we became consistent */ + pq_sendstring(&buf, "consistent_point"); /* col name */ + pq_sendint(&buf, 0, 4); /* table oid */ + pq_sendint(&buf, 0, 2); /* attnum */ + pq_sendint(&buf, TEXTOID, 4); /* type oid */ + pq_sendint(&buf, -1, 2); /* typlen */ + pq_sendint(&buf, 0, 4); /* typmod */ + pq_sendint(&buf, 0, 2); /* format code */ + + /* second field: exported snapshot's name */ + pq_sendstring(&buf, "snapshot_name"); /* col name */ + pq_sendint(&buf, 0, 4); /* table oid */ + pq_sendint(&buf, 0, 2); /* attnum */ + pq_sendint(&buf, TEXTOID, 4); /* type oid */ + pq_sendint(&buf, -1, 2); /* typlen */ + pq_sendint(&buf, 0, 4); /* typmod */ + pq_sendint(&buf, 0, 2); /* format code */ + + pq_endmessage(&buf); + + /* Send a DataRow message */ + pq_beginmessage(&buf, 'D'); + pq_sendint(&buf, 2, 2); /* # of columns */ + + /* consistent wal location */ + pq_sendint(&buf, strlen(xpos), 4); /* col1 len */ + pq_sendbytes(&buf, xpos, strlen(xpos)); + + /* snapshot name */ + pq_sendint(&buf, strlen(snapshot_name), 4); /* col2 len */ + pq_sendbytes(&buf, snapshot_name, strlen(snapshot_name)); + + pq_endmessage(&buf); +} + + /* * Get rid of a replication slot that is no longer wanted. */ @@ -1332,6 +1424,10 @@ exec_replication_command(const char *cmd_string) SendTimeLineHistory((TimeLineHistoryCmd *) cmd_node); break; + case T_LogicalDecodingSnapshotCmd: + LogicalDecodingSnapshot((LogicalDecodingSnapshotCmd *) cmd_node); + break; + default: elog(ERROR, "unrecognized replication command node tag: %u", cmd_node->type); diff --git a/src/bin/pg_basebackup/pg_recvlogical.c b/src/bin/pg_basebackup/pg_recvlogical.c index f189f71..31e8aa2 100644 --- a/src/bin/pg_basebackup/pg_recvlogical.c +++ b/src/bin/pg_basebackup/pg_recvlogical.c @@ -41,6 +41,7 @@ static bool do_create_slot = false; static bool slot_exists_ok = false; static bool do_start_slot = false; static bool do_drop_slot = false; +static bool do_snapshot = false; /* filled pairwise with option, value. value may be NULL */ static char **options; @@ -72,6 +73,7 @@ usage(void) printf(_(" --create-slot create a new replication slot (for the slot's name see --slot)\n")); printf(_(" --drop-slot drop the replication slot (for the slot's name see --slot)\n")); printf(_(" --start start streaming in a replication slot (for the slot's name see --slot)\n")); + printf(_(" --snapshot export a snapshot\n")); printf(_("\nOptions:\n")); printf(_(" -f, --file=FILE receive log into this file, - for stdout\n")); printf(_(" -F --fsync-interval=SECS\n" @@ -636,6 +638,7 @@ main(int argc, char **argv) {"start", no_argument, NULL, 2}, {"drop-slot", no_argument, NULL, 3}, {"if-not-exists", no_argument, NULL, 4}, + {"snapshot", no_argument, NULL, 5}, {NULL, 0, NULL, 0} }; int c; @@ -770,6 +773,9 @@ main(int argc, char **argv) case 4: slot_exists_ok = true; break; + case 5: + do_snapshot = true; + break; default: @@ -822,7 +828,7 @@ main(int argc, char **argv) exit(1); } - if (!do_drop_slot && !do_create_slot && !do_start_slot) + if (!do_drop_slot && !do_create_slot && !do_start_slot && !do_snapshot) { fprintf(stderr, _("%s: at least one action needs to be specified\n"), progname); fprintf(stderr, _("Try \"%s --help\" for more information.\n"), @@ -902,6 +908,23 @@ main(int argc, char **argv) startpos = InvalidXLogRecPtr; } + if (do_snapshot) + { + if (verbose) + fprintf(stderr, + _("%s: exporting a new snapshot for logical decoding\n"), + progname); + + if (!LogicalDecodingSnapshot(conn, replication_slot, plugin)) + disconnect_and_exit(1); + + while (!time_to_abort) + { + pg_usleep(2000); + } + disconnect_and_exit(0); + } + if (!do_start_slot) disconnect_and_exit(0); diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c index 91f919c..17436c6 100644 --- a/src/bin/pg_basebackup/streamutil.c +++ b/src/bin/pg_basebackup/streamutil.c @@ -375,6 +375,58 @@ CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin, } /* + * Export a new snapshot synchronized with logical decoding LSNs. + */ +bool +LogicalDecodingSnapshot(PGconn *conn, const char *slot_name, const char *plugin) +{ + PQExpBuffer query; + PGresult *res; + + query = createPQExpBuffer(); + + Assert(plugin != NULL); + Assert(slot_name != NULL); + + /* Build query */ + appendPQExpBuffer(query, "LOGICAL_DECODING_SNAPSHOT \"%s\" \"%s\"", + slot_name, plugin); + + res = PQexec(conn, query->data); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + fprintf(stderr, _("%s: could not export snapshot \"%s\": %s"), + progname, query->data, PQerrorMessage(conn)); + + destroyPQExpBuffer(query); + PQclear(res); + return false; + } + + if (PQntuples(res) != 1 || PQnfields(res) != 2) + { + fprintf(stderr, + _("%s: could not export snapshot: got %d rows and %d fields, expected %d rows and %d fields\n"), + progname, + PQntuples(res), PQnfields(res), 1, 2); + + destroyPQExpBuffer(query); + PQclear(res); + return false; + } + + /* Print LSN position*/ + printf("LSN: %s\n", PQgetvalue(res, 0, 0)); + + /* Print Snapshot name */ + printf("Snapshot: %s\n", PQgetvalue(res, 0, 1)); + + destroyPQExpBuffer(query); + PQclear(res); + return true; +} + +/* * Drop a replication slot for the given connection. This function * returns true in case of success. */ diff --git a/src/bin/pg_basebackup/streamutil.h b/src/bin/pg_basebackup/streamutil.h index b95f83f..2f8da8e 100644 --- a/src/bin/pg_basebackup/streamutil.h +++ b/src/bin/pg_basebackup/streamutil.h @@ -34,6 +34,8 @@ extern PGconn *GetConnection(void); extern bool CreateReplicationSlot(PGconn *conn, const char *slot_name, const char *plugin, bool is_physical, bool slot_exists_ok); +extern bool LogicalDecodingSnapshot(PGconn *conn, const char *slot_name, + const char *plugin); extern bool DropReplicationSlot(PGconn *conn, const char *slot_name); extern bool RunIdentifySystem(PGconn *conn, char **sysid, TimeLineID *starttli, diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h index 290cdb3..a7162ef 100644 --- a/src/include/nodes/nodes.h +++ b/src/include/nodes/nodes.h @@ -436,6 +436,7 @@ typedef enum NodeTag T_DropReplicationSlotCmd, T_StartReplicationCmd, T_TimeLineHistoryCmd, + T_LogicalDecodingSnapshotCmd, /* * TAGS FOR RANDOM OTHER STUFF diff --git a/src/include/nodes/replnodes.h b/src/include/nodes/replnodes.h index cac6419..f3cc739 100644 --- a/src/include/nodes/replnodes.h +++ b/src/include/nodes/replnodes.h @@ -59,6 +59,18 @@ typedef struct CreateReplicationSlotCmd /* ---------------------- + * LOGICAL_DECODING_SNAPSHOT command + * ---------------------- + */ +typedef struct LogicalDecodingSnapshotCmd +{ + NodeTag type; + char *slotname; + char *plugin; +} LogicalDecodingSnapshotCmd; + + +/* ---------------------- * DROP_REPLICATION_SLOT command * ---------------------- */