From 4b0d31d9462d4c18a508be5461f2cc16894501a1 Mon Sep 17 00:00:00 2001 From: Ajin Cherian Date: Tue, 9 Mar 2021 04:38:22 -0500 Subject: [PATCH v55] Add support for apply at prepare time to built-in logical replication. To add support for streaming transactions at prepare time into the built-in logical replication, we need to do the below things: * Modify the output plugin (pgoutput) to implement the new two-phase API callbacks, by leveraging the extended replication protocol. * Modify the replication apply worker, to properly handle two-phase transactions by replaying them on prepare. * prepare API for streaming transactions is not supported. We however must explicitly disable replication of two-phase transactions during replication slot creation, even if the plugin supports it. We don't need to replicate the changes accumulated during this phase, and moreover, we don't have a replication connection open so we don't know where to send the data anyway. * This patch also adds new option to enable two_phase while creating a slot. Author: Peter Smith, Ajin Cherian and Amit Kapila based on previous work by Nikhil Sontakke and Stas Kelvich Reviewed-by: Amit Kapila, Sawada Masahiko, and Dilip Kumar Discussion: https://postgr.es/m/02DA5F5E-CECE-4D9C-8B4B-418077E2C010@postgrespro.ru https://postgr.es/m/CAMGcDxeqEpWj3fTXwqhSwBdXd2RS9jzwWscO-XbeCfso6ts3+Q@mail.gmail.com --- doc/src/sgml/protocol.sgml | 14 +- src/backend/access/transam/twophase.c | 68 ++ src/backend/commands/subscriptioncmds.c | 2 +- .../libpqwalreceiver/libpqwalreceiver.c | 6 +- src/backend/replication/logical/origin.c | 7 +- src/backend/replication/logical/proto.c | 206 ++++++ src/backend/replication/logical/tablesync.c | 180 ++++- src/backend/replication/logical/worker.c | 767 ++++++++++++++++++++- src/backend/replication/pgoutput/pgoutput.c | 157 ++++- src/backend/replication/repl_gram.y | 16 +- src/backend/replication/repl_scanner.l | 1 + src/backend/replication/walreceiver.c | 2 +- src/include/access/twophase.h | 2 + src/include/replication/logicalproto.h | 69 +- src/include/replication/reorderbuffer.h | 12 + src/include/replication/walreceiver.h | 5 +- src/include/replication/worker_internal.h | 3 + src/tools/pgindent/typedefs.list | 5 + 18 files changed, 1438 insertions(+), 84 deletions(-) diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml index 43092fe..9694713 100644 --- a/doc/src/sgml/protocol.sgml +++ b/doc/src/sgml/protocol.sgml @@ -1914,7 +1914,7 @@ The commands accepted in replication mode are: - CREATE_REPLICATION_SLOT slot_name [ TEMPORARY ] { PHYSICAL [ RESERVE_WAL ] | LOGICAL output_plugin [ EXPORT_SNAPSHOT | NOEXPORT_SNAPSHOT | USE_SNAPSHOT ] } + CREATE_REPLICATION_SLOT slot_name [ TEMPORARY ] [ TWO_PHASE ] { PHYSICAL [ RESERVE_WAL ] | LOGICAL output_plugin [ EXPORT_SNAPSHOT | NOEXPORT_SNAPSHOT | USE_SNAPSHOT ] } CREATE_REPLICATION_SLOT @@ -1956,6 +1956,18 @@ The commands accepted in replication mode are: + TWO_PHASE + + + Specify that this replication slot supports decode of two-phase transactions. + Two-phase commands like PREPARE TRANSACTION, COMMIT PREPARED and ROLLBACK PREPARED + are also decoded and transmitted. In two-phase transactions, the transaction is + decoded and transmitted at PREPARE TRANSACTION time. + + + + + RESERVE_WAL diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index 6023e7c..81cb765 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -2445,3 +2445,71 @@ PrepareRedoRemove(TransactionId xid, bool giveWarning) RemoveTwoPhaseFile(xid, giveWarning); RemoveGXact(gxact); } + +/* + * LookupGXact + * Check if the prepared transaction with the given GID and lsn is around. + * + * Note that we always compare with the LSN where prepare ends because that is + * what is stored as origin_lsn in the 2PC file. + * + * This function is primarily used to check if the prepared transaction + * received from the upstream (remote node) already exists. Checking only GID + * is not sufficient because a different prepared xact with the same GID can + * exist on the same node. So, we are ensuring to match origin_lsn and + * origin_timestamp of prepared xact to avoid the possibility of a match of + * prepared xact from two different nodes. + */ +bool +LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn, + TimestampTz origin_prepare_timestamp) +{ + int i; + bool found = false; + + LWLockAcquire(TwoPhaseStateLock, LW_SHARED); + for (i = 0; i < TwoPhaseState->numPrepXacts; i++) + { + GlobalTransaction gxact = TwoPhaseState->prepXacts[i]; + + /* Ignore not-yet-valid GIDs. */ + if ((gxact->valid && strcmp(gxact->gid, gid) == 0)) + { + char* buf; + TwoPhaseFileHeader* hdr; + + /* + * We are neither expecting the collisions of GXACTs (same gid) + * between publisher and subscribers nor the apply worker restarts + * after prepared xacts, so we perform all I/O while holding + * TwoPhaseStateLock for simplicity. + * + * To move the I/O out of the lock, we need to ensure that no other + * backend commits the prepared xact in the meantime. We can do + * this optimization if we encounter many collisions in GID between + * publisher and subscriber. + */ + if (gxact->ondisk) + buf = ReadTwoPhaseFile(gxact->xid, false); + else + { + Assert(gxact->prepare_start_lsn); + XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL); + } + + hdr = (TwoPhaseFileHeader *) buf; + + if (hdr->origin_lsn == prepare_end_lsn && + hdr->origin_timestamp == origin_prepare_timestamp) + { + found = true; + pfree(buf); + break; + } + + pfree(buf); + } + } + LWLockRelease(TwoPhaseStateLock); + return found; +} 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/origin.c b/src/backend/replication/logical/origin.c index 39471fd..b258174 100644 --- a/src/backend/replication/logical/origin.c +++ b/src/backend/replication/logical/origin.c @@ -973,8 +973,11 @@ replorigin_advance(RepOriginId node, /* * Due to - harmless - race conditions during a checkpoint we could see - * values here that are older than the ones we already have in memory. - * Don't overwrite those. + * values here that are older than the ones we already have in memory. We + * could also see older values for prepared transactions when the prepare + * is sent at a later point of time along with commit prepared and there + * are other transactions commits between prepare and commit prepared. See + * ReorderBufferFinishPrepared. Don't overwrite those. */ if (go_backward || replication_state->remote_lsn < remote_commit) replication_state->remote_lsn = remote_commit; diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c index f2c85ca..488b2a2 100644 --- a/src/backend/replication/logical/proto.c +++ b/src/backend/replication/logical/proto.c @@ -106,6 +106,212 @@ logicalrep_read_commit(StringInfo in, LogicalRepCommitData *commit_data) } /* + * Write BEGIN PREPARE to the output stream. + */ +void +logicalrep_write_begin_prepare(StringInfo out, ReorderBufferTXN *txn) +{ + pq_sendbyte(out, LOGICAL_REP_MSG_BEGIN_PREPARE); + + /* fixed fields */ + pq_sendint64(out, txn->final_lsn); + pq_sendint64(out, txn->end_lsn); + pq_sendint64(out, txn->commit_time); + pq_sendint32(out, txn->xid); + + /* send gid */ + pq_sendstring(out, txn->gid); +} + +/* + * Read transaction BEGIN PREPARE from the stream. + */ +void +logicalrep_read_begin_prepare(StringInfo in, LogicalRepBeginPrepareData *begin_data) +{ + /* read fields */ + begin_data->final_lsn = pq_getmsgint64(in); + if (begin_data->final_lsn == InvalidXLogRecPtr) + elog(ERROR, "final_lsn not set in begin message"); + begin_data->end_lsn = pq_getmsgint64(in); + if (begin_data->end_lsn == InvalidXLogRecPtr) + elog(ERROR, "end_lsn not set in begin message"); + begin_data->committime = pq_getmsgint64(in); + begin_data->xid = pq_getmsgint(in, 4); + + /* read gid (copy it into a pre-allocated buffer) */ + strcpy(begin_data->gid, pq_getmsgstring(in)); +} + +/* + * Write PREPARE to the output stream. + */ +void +logicalrep_write_prepare(StringInfo out, ReorderBufferTXN *txn, + XLogRecPtr prepare_lsn) +{ + uint8 flags = 0; + + pq_sendbyte(out, LOGICAL_REP_MSG_PREPARE); + + /* + * This should only ever happen for two-phase commit transactions. In + * which case we expect to have a valid GID. + */ + Assert(txn->gid != NULL); + Assert(rbtxn_prepared(txn)); + + /* send the flags field */ + pq_sendbyte(out, flags); + + /* send fields */ + pq_sendint64(out, prepare_lsn); + pq_sendint64(out, txn->end_lsn); + pq_sendint64(out, txn->commit_time); + + /* send gid */ + pq_sendstring(out, txn->gid); +} + +/* + * Read transaction PREPARE from the stream. + */ +void +logicalrep_read_prepare(StringInfo in, LogicalRepPreparedTxnData *prepare_data) +{ + /* read flags */ + uint8 flags = pq_getmsgbyte(in); + + if (flags != 0) + elog(ERROR, "unrecognized flags %u in prepare message", flags); + + /* read fields */ + prepare_data->prepare_lsn = pq_getmsgint64(in); + if (prepare_data->prepare_lsn == InvalidXLogRecPtr) + elog(ERROR, "prepare_lsn is not set in prepare message"); + prepare_data->end_lsn = pq_getmsgint64(in); + if (prepare_data->end_lsn == InvalidXLogRecPtr) + elog(ERROR, "end_lsn is not set in prepare message"); + prepare_data->preparetime = pq_getmsgint64(in); + + /* read gid (copy it into a pre-allocated buffer) */ + strcpy(prepare_data->gid, pq_getmsgstring(in)); +} + +/* + * Write COMMIT PREPARED to the output stream. + */ +void +logicalrep_write_commit_prepared(StringInfo out, ReorderBufferTXN *txn, + XLogRecPtr commit_lsn) +{ + uint8 flags = 0; + + pq_sendbyte(out, LOGICAL_REP_MSG_COMMIT_PREPARED); + + /* + * This should only ever happen for two-phase commit transactions. In + * which case we expect to have a valid GID. Additionally, the transaction + * must be prepared. See ReorderBufferFinishPrepared. + */ + Assert(txn->gid != NULL); + + /* send the flags field */ + pq_sendbyte(out, flags); + + /* send fields */ + pq_sendint64(out, commit_lsn); + pq_sendint64(out, txn->end_lsn); + pq_sendint64(out, txn->commit_time); + + /* send gid */ + pq_sendstring(out, txn->gid); +} + +/* + * Read transaction COMMIT PREPARED from the stream. + */ +void +logicalrep_read_commit_prepared(StringInfo in, LogicalRepPreparedTxnData *prepare_data) +{ + /* read flags */ + uint8 flags = pq_getmsgbyte(in); + + if (flags != 0) + elog(ERROR, "unrecognized flags %u in commit prepare message", flags); + + /* read fields */ + prepare_data->prepare_lsn = pq_getmsgint64(in); + if (prepare_data->prepare_lsn == InvalidXLogRecPtr) + elog(ERROR, "prepare_lsn is not set in commit prepared message"); + prepare_data->end_lsn = pq_getmsgint64(in); + if (prepare_data->end_lsn == InvalidXLogRecPtr) + elog(ERROR, "end_lsn is not set in commit prepared message"); + prepare_data->preparetime = pq_getmsgint64(in); + + /* read gid (copy it into a pre-allocated buffer) */ + strcpy(prepare_data->gid, pq_getmsgstring(in)); +} + +/* + * Write ROLLBACK PREPARED to the output stream. + */ +void +logicalrep_write_rollback_prepared(StringInfo out, ReorderBufferTXN *txn, + XLogRecPtr prepare_end_lsn, + TimestampTz prepare_time) +{ + uint8 flags = 0; + + pq_sendbyte(out, LOGICAL_REP_MSG_ROLLBACK_PREPARED); + + /* + * This should only ever happen for two-phase commit transactions. In + * which case we expect to have a valid GID. + */ + Assert(txn->gid != NULL); + + /* send the flags field */ + pq_sendbyte(out, flags); + + /* send fields */ + pq_sendint64(out, prepare_end_lsn); + pq_sendint64(out, txn->end_lsn); + pq_sendint64(out, prepare_time); + pq_sendint64(out, txn->commit_time); + + /* send gid */ + pq_sendstring(out, txn->gid); +} + +/* + * Read transaction ROLLBACK PREPARED from the stream. + */ +void +logicalrep_read_rollback_prepared(StringInfo in, + LogicalRepRollbackPreparedTxnData *rollback_data) +{ + /* read flags */ + uint8 flags = pq_getmsgbyte(in); + + if (flags != 0) + elog(ERROR, "unrecognized flags %u in rollback prepare message", flags); + + /* read fields */ + rollback_data->prepare_end_lsn = pq_getmsgint64(in); + if (rollback_data->prepare_end_lsn == InvalidXLogRecPtr) + elog(ERROR, "prepare_end_lsn is not set in commit prepared message"); + rollback_data->rollback_end_lsn = pq_getmsgint64(in); + if (rollback_data->rollback_end_lsn == InvalidXLogRecPtr) + elog(ERROR, "rollback_end_lsn is not set in commit prepared message"); + rollback_data->preparetime = pq_getmsgint64(in); + rollback_data->rollbacktime = pq_getmsgint64(in); + + /* read gid (copy it into a pre-allocated buffer) */ + strcpy(rollback_data->gid, pq_getmsgstring(in)); +} + +/* * Write ORIGIN to the output stream. */ void diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index feb634e..97fc399 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -116,6 +116,9 @@ #include "utils/snapmgr.h" static bool table_states_valid = false; +static List *table_states_not_ready = NIL; +static List *table_states_all = NIL; +static void FetchTableStates(bool *started_tx); StringInfo copybuf = NULL; @@ -359,7 +362,6 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) Oid relid; TimestampTz last_start_time; }; - static List *table_states = NIL; static HTAB *last_start_times = NULL; ListCell *lc; bool started_tx = false; @@ -367,42 +369,14 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) Assert(!IsTransactionState()); /* We need up-to-date sync state info for subscription tables here. */ - if (!table_states_valid) - { - MemoryContext oldctx; - List *rstates; - ListCell *lc; - SubscriptionRelState *rstate; - - /* Clean the old list. */ - list_free_deep(table_states); - table_states = NIL; - - StartTransactionCommand(); - started_tx = true; - - /* Fetch all non-ready tables. */ - rstates = GetSubscriptionNotReadyRelations(MySubscription->oid); - - /* Allocate the tracking info in a permanent memory context. */ - oldctx = MemoryContextSwitchTo(CacheMemoryContext); - foreach(lc, rstates) - { - rstate = palloc(sizeof(SubscriptionRelState)); - memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState)); - table_states = lappend(table_states, rstate); - } - MemoryContextSwitchTo(oldctx); - - table_states_valid = true; - } + FetchTableStates(&started_tx); /* * Prepare a hash table for tracking last start times of workers, to avoid * immediate restarts. We don't need it if there are no tables that need * syncing. */ - if (table_states && !last_start_times) + if (table_states_not_ready && !last_start_times) { HASHCTL ctl; @@ -416,7 +390,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) * Clean up the hash table when we're done with all tables (just to * release the bit of memory). */ - else if (!table_states && last_start_times) + else if (!table_states_not_ready && last_start_times) { hash_destroy(last_start_times); last_start_times = NULL; @@ -425,7 +399,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) /* * Process all tables that are being synchronized. */ - foreach(lc, table_states) + foreach(lc, table_states_not_ready) { SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); @@ -1052,7 +1026,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); /* @@ -1137,3 +1111,141 @@ copy_table_done: wait_for_worker_state_change(SUBREL_STATE_CATCHUP); return slotname; } + +/* + * Common code to fetch the up-to-date sync state info into the static lists. + */ +static void +FetchTableStates(bool *started_tx) +{ + *started_tx = false; + + if (!table_states_valid) + { + MemoryContext oldctx; + List *rstates; + ListCell *lc; + SubscriptionRelState *rstate; + + /* Clean the old lists. */ + list_free_deep(table_states_all); + table_states_all = NIL; + list_free_deep(table_states_not_ready); + table_states_not_ready = NIL; + + StartTransactionCommand(); + *started_tx = true; + + /* Fetch all tables. */ + rstates = GetSubscriptionRelations(MySubscription->oid); + + /* Allocate the tracking info in a permanent memory context. */ + oldctx = MemoryContextSwitchTo(CacheMemoryContext); + foreach(lc, rstates) + { + SubscriptionRelState *cur_rstate = (SubscriptionRelState *) lfirst(lc); + + /* List of all states */ + rstate = palloc(sizeof(SubscriptionRelState)); + memcpy(rstate, cur_rstate, sizeof(SubscriptionRelState)); + table_states_all = lappend(table_states_all, rstate); + + /* List of only not-ready states */ + if (cur_rstate->state != SUBREL_STATE_READY) + { + rstate = palloc(sizeof(SubscriptionRelState)); + memcpy(rstate, cur_rstate, sizeof(SubscriptionRelState)); + table_states_not_ready = lappend(table_states_not_ready, rstate); + } + } + MemoryContextSwitchTo(oldctx); + + table_states_valid = true; + } +} + +/* + * Are there any tablesyncs which have still not yet reached SYNCDONE/READY state? + */ +bool +AnyTablesyncInProgress() +{ + bool found_busy = false; + bool started_tx = false; + int count = 0; + ListCell *lc; + + /* We need up-to-date sync state info for subscription tables here. */ + FetchTableStates(&started_tx); + + /* + * Process all not-READY tables to see if any are also not-SYNCDONE + */ + foreach(lc, table_states_not_ready) + { + SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); + + count++; + elog(DEBUG1, + "AnyTablesyncInProgress?: #%d. Table relid %u has state '%c'", + count, + rstate->relid, + rstate->state); + + /* + * When the process_syncing_tables_for_apply changes the state + * from SYNCDONE to READY, that change is actually written directly + * into the list element of table_states_not_ready. + * + * So the "table_states_not_ready" list might end up having a READY + * state in it even though there was none when it was initially + * created. This is reason why we need to check for READY below. + */ + if (rstate->state != SUBREL_STATE_SYNCDONE && + rstate->state != SUBREL_STATE_READY) + { + found_busy = true; + break; + } + } + + if (started_tx) + { + CommitTransactionCommand(); + pgstat_report_stat(false); + } + + elog(DEBUG1, + "AnyTablesyncInProgress?: Scanned %d tables, and found busy = %s", + count, + found_busy ? "true" : "false"); + + return found_busy; +} + +/* + * What is the biggest LSN from the all the known tablesyncs? + */ +XLogRecPtr +BiggestTablesyncLSN() +{ + XLogRecPtr biggest_lsn = InvalidXLogRecPtr; + ListCell *lc; + int count = 0; + + foreach(lc, table_states_all) + { + SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); + + count++; + if (rstate->lsn > biggest_lsn) + biggest_lsn = rstate->lsn; + } + + elog(DEBUG1, + "BiggestTablesyncLSN: Scanned %d tables. Biggest lsn found = %X/%X", + count, + LSN_FORMAT_ARGS(biggest_lsn)); + + return biggest_lsn; +} diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 18d0528..1cdfc91 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -59,6 +59,7 @@ #include "access/table.h" #include "access/tableam.h" +#include "access/twophase.h" #include "access/xact.h" #include "access/xlog_internal.h" #include "catalog/catalog.h" @@ -208,6 +209,54 @@ static void subxact_info_add(TransactionId xid); static inline void cleanup_subxact_info(void); /* + * The following are for the support of a spoolfile for prepared messages. + */ + +/* + * A Prepare spoolfile hash entry. We create this entry in the psf_hash. This is + * for maintaining a mapping between the name of the prepared spoolfile, and the + * corresponding fileset handles of same. + */ +typedef struct PsfHashEntry +{ + char name[MAXPGPATH]; /* Hash key --- must be first */ + bool allow_delete; /* ok to delete? */ +} PsfHashEntry; + +/* + * Information about the "current" psf spoolfile. + */ +typedef struct PsfFile +{ + char name[MAXPGPATH];/* psf name - same as the HTAB key. */ + bool is_spooling; /* are we currently spooling to this file? */ + File vfd; /* -1 when the file is closed. */ + off_t cur_offset; /* offset for the next write or read. Reset to 0 + * when file is opened. */ +} PsfFile; + +/* + * Hash table for storing the Prepared spoolfile info along with shared fileset. + */ +static HTAB *psf_hash = NULL; + +/* + * Information about the 'current' open spoolfile is only valid when spooling. + * This is flagged as 'is_spooling' only between begin_prepare and prepare. + */ +static PsfFile psf_cur = { .is_spooling = false, .vfd = -1, .cur_offset = 0 }; + +static void prepare_spoolfile_create(char *path); +static void prepare_spoolfile_write(char action, StringInfo s); +static void prepare_spoolfile_close(void); +static void prepare_spoolfile_delete(char *path); +static bool prepare_spoolfile_exists(char *path); +static void prepare_spoolfile_name(char *path, int szpath, Oid subid, char *gid); +static int prepare_spoolfile_replay_messages(char *path, XLogRecPtr final_lsn); +static bool prepare_spoolfile_handler(LogicalRepMsgType action, StringInfo s); +static void prepare_spoolfile_on_proc_exit(int status, Datum arg); + +/* * Serialize and deserialize changes for a toplevel transaction. */ static void stream_cleanup_files(Oid subid, TransactionId xid); @@ -720,6 +769,338 @@ apply_handle_commit(StringInfo s) } /* + * Handle BEGIN PREPARE message. + */ +static void +apply_handle_begin_prepare(StringInfo s) +{ + LogicalRepBeginPrepareData begin_data; + + /* Tablesync should never receive prepare. */ + Assert(!am_tablesync_worker()); + + logicalrep_read_begin_prepare(s, &begin_data); + + /* The gid must not already be prepared. */ + if (LookupGXact(begin_data.gid, begin_data.end_lsn, begin_data.committime)) + ereport(ERROR, + (errcode(ERRCODE_DUPLICATE_OBJECT), + errmsg("transaction identifier \"%s\" is already in use", + begin_data.gid))); + + /* + * By sad timing of apply/tablesync workers it is possible to have a + * “consistent snapshot” that spans prepare/commit in such a way that + * the tablesync did not do the prepare (because snapshot not consistent) + * and the apply worker does the begin prepare (‘b’) but it skips all + * the prepared operations [e.g. inserts] while the tablesync was still + * busy (see the condition of should_apply_changes_for_rel). + * + * This can lead to an "empty prepare", because later when the apply + * worker does the commit prepare (‘K’), there is nothing in it (the + * inserts were skipped earlier). + * + * We avoid this using the 2 part logic: (1) Wait for all tablesync workers + * to reach SYNCDONE/READY state; (2) If the begin_prepare lsn is now + * behind any tablesync lsn then spool the prepared messages to a file + * to be replayed later at commit_prepared time. + * + * ----- + * + * XXX - The 2PC protocol needs the publisher to be aware when the PREPARE + * has been successfully acted on. But because of this "empty prepare" + * case now the prepared messages may be spooled to a file and, when + * that happens the PREPARE would not happen at the usual time, but would + * be deferred until COMMIT PREPARED time. This quirk could only happen + * immediately after the initial table synchronization phase; once all + * tables have acheived READY state the 2PC protocol will behave normally. + * + * A future release may be able to detect when all tables are READY and set + * a flag to indicate this subscription/slot is ready for two_phase + * decoding. Then at the publisher-side, we could enable wait-for-prepares + * only when all the slots of WALSender have that flag set. + */ + if (!am_tablesync_worker()) + { + /* + * Part 1 of 2: + * + * Make sure every tablesync has reached at least SYNCDONE state + * before letting the apply worker proceed. + */ + elog(DEBUG1, + "apply_handle_begin_prepare, end_lsn = %X/%X, final_lsn = %X/%X, lstate_lsn = %X/%X", + LSN_FORMAT_ARGS(begin_data.end_lsn), + LSN_FORMAT_ARGS(begin_data.final_lsn), + LSN_FORMAT_ARGS(MyLogicalRepWorker->relstate_lsn)); + + while (AnyTablesyncInProgress()) + { + process_syncing_tables(begin_data.final_lsn); + + /* This latch is to prevent 100% CPU looping. */ + (void) WaitLatch(MyLatch, + WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, + 1000L, WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE); + ResetLatch(MyLatch); + } + + /* + * Part 2 of 2: + * + * If (when getting to SYNCDONE/READY state) some tablesync went + * beyond this begin_prepare LSN then set all messages (until + * prepared) will be saved to a spoolfile for replay later at + * commit_prepared time. + */ + if (begin_data.final_lsn < BiggestTablesyncLSN()) + { + char psfpath[MAXPGPATH]; + + /* + * Create the spoolfile. + */ + prepare_spoolfile_name(psfpath, sizeof(psfpath), + MyLogicalRepWorker->subid, begin_data.gid); + prepare_spoolfile_create(psfpath); + + /* + * From now, until the handle_prepare we are spooling to the + * current psf. + */ + psf_cur.is_spooling = true; + } + } + + remote_final_lsn = begin_data.final_lsn; + + in_remote_transaction = true; + + pgstat_report_activity(STATE_RUNNING, NULL); +} + +/* + * Handle PREPARE message. + */ +static void +apply_handle_prepare(StringInfo s) +{ + LogicalRepPreparedTxnData prepare_data; + + /* + * If we were using a psf spoolfile, then write the PREPARE as the final + * message. This prepare information will be used at commit_prepared time. + */ + if (psf_cur.is_spooling) + { + PsfHashEntry *hentry; + + /* Write the PREPARE info to the psf file. */ + prepare_spoolfile_handler(LOGICAL_REP_MSG_PREPARE, s); + + /* + * Flush the spoolfile, so changes can survive a restart. + * + * If the publisher resends the same data again after a restart (e.g. + * if subscriber origin has not moved past this prepare), then the same + * named psf file will be overwritten with the same data. See + * prepare_spoolfile_create. + */ + FileSync(psf_cur.vfd, WAIT_EVENT_DATA_FILE_SYNC); + + /* We are finished spooling to the current psf. */ + psf_cur.is_spooling = false; + + /* + * The commit_prepare will need the spoolfile, so unregister it for + * removal on proc-exit just in case there is an unexpected restart + * between now and when commit_prepared happens. + */ + hentry = (PsfHashEntry *) hash_search(psf_hash, psf_cur.name, + HASH_FIND, NULL); + Assert(hentry); + hentry->allow_delete = false; + + /* + * The psf_cur.vfd is meaningful only between begin_prepare and prepared. + * So close it now. Any messages written to the psf will be applied + * later during handle_commit_prepared. + */ + prepare_spoolfile_close(); + + in_remote_transaction = false; + return; + } + + logicalrep_read_prepare(s, &prepare_data); + + /* + * Normally, prepare_lsn == remote_final_lsn, but if this prepare message + * was dispatched via the psf spoolfile replay then the remote_final_lsn + * is set to commit lsn instead. Hence the <= instead of == check below. + */ + Assert(prepare_data.prepare_lsn <= remote_final_lsn); + + if (IsTransactionState()) + { + /* + * BeginTransactionBlock is necessary to balance the + * EndTransactionBlock called within the PrepareTransactionBlock + * below. + */ + BeginTransactionBlock(); + CommitTransactionCommand(); + + /* + * Update origin state so we can restart streaming from correct + * position in case of crash. + */ + replorigin_session_origin_lsn = prepare_data.end_lsn; + replorigin_session_origin_timestamp = prepare_data.preparetime; + + PrepareTransactionBlock(prepare_data.gid); + CommitTransactionCommand(); + pgstat_report_stat(false); + + store_flush_position(prepare_data.end_lsn); + } + else + { + /* Process any invalidation messages that might have accumulated. */ + AcceptInvalidationMessages(); + maybe_reread_subscription(); + } + + in_remote_transaction = false; + + /* Process any tables that are being synchronized in parallel. */ + process_syncing_tables(prepare_data.end_lsn); + + pgstat_report_activity(STATE_IDLE, NULL); +} + +/* + * Handle a COMMIT PREPARED of a previously PREPARED transaction. + */ +static void +apply_handle_commit_prepared(StringInfo s) +{ + LogicalRepPreparedTxnData prepare_data; + char psfpath[MAXPGPATH]; + + logicalrep_read_commit_prepared(s, &prepare_data); + + /* + * If this prepare's messages were being spooled to a file, then replay + * them all now. + */ + prepare_spoolfile_name(psfpath, sizeof(psfpath), + MyLogicalRepWorker->subid, prepare_data.gid); + if (prepare_spoolfile_exists(psfpath)) + { + int nchanges; + + /* + * Replay/dispatch the spooled messages. + */ + + ensure_transaction(); + + nchanges = prepare_spoolfile_replay_messages(psfpath, prepare_data.prepare_lsn); + elog(DEBUG1, + "apply_handle_commit_prepared: replayed %d (all) changes.", + nchanges); + + /* After replaying the psf it is no longer needed. Just delete it. */ + prepare_spoolfile_delete(psfpath); + } + + /* there is no transaction when COMMIT PREPARED is called */ + ensure_transaction(); + + /* + * Update origin state so we can restart streaming from correct position + * in case of crash. + */ + replorigin_session_origin_lsn = prepare_data.end_lsn; + replorigin_session_origin_timestamp = prepare_data.preparetime; + + FinishPreparedTransaction(prepare_data.gid, true); + CommitTransactionCommand(); + pgstat_report_stat(false); + + store_flush_position(prepare_data.end_lsn); + in_remote_transaction = false; + + /* Process any tables that are being synchronized in parallel. */ + process_syncing_tables(prepare_data.end_lsn); + + pgstat_report_activity(STATE_IDLE, NULL); +} + +/* + * Handle a ROLLBACK PREPARED of a previously PREPARED TRANSACTION. + */ +static void +apply_handle_rollback_prepared(StringInfo s) +{ + LogicalRepRollbackPreparedTxnData rollback_data; + bool using_psf; + char psfpath[MAXPGPATH]; + + logicalrep_read_rollback_prepared(s, &rollback_data); + + /* + * If this prepare's messages were being spooled to a file, then cleanup + * the file. + */ + prepare_spoolfile_name(psfpath, sizeof(psfpath), + MyLogicalRepWorker->subid, rollback_data.gid); + using_psf = prepare_spoolfile_exists(psfpath); + if (using_psf) + { + /* We are finished with this spoolfile. Delete it. */ + prepare_spoolfile_delete(psfpath); + } + + /* + * It is possible that we haven't received prepare because it occurred + * before walsender reached a consistent point in which case we need to + * skip rollback prepared. + * + * And we also skip the FinishPreparedTransaction if we're using the + * Prepare Spoolfile (using_psf) because in that case there is no matching + * PrepareTransactionBlock done yet. + */ + if (!using_psf && + LookupGXact(rollback_data.gid, rollback_data.prepare_end_lsn, + rollback_data.preparetime)) + { + /* + * Update origin state so we can restart streaming from correct + * position in case of crash. + */ + replorigin_session_origin_lsn = rollback_data.rollback_end_lsn; + replorigin_session_origin_timestamp = rollback_data.rollbacktime; + + /* there is no transaction when ABORT/ROLLBACK PREPARED is called */ + ensure_transaction(); + FinishPreparedTransaction(rollback_data.gid, false); + CommitTransactionCommand(); + } + + pgstat_report_stat(false); + + store_flush_position(rollback_data.rollback_end_lsn); + in_remote_transaction = false; + + /* Process any tables that are being synchronized in parallel. */ + process_syncing_tables(rollback_data.rollback_end_lsn); + + pgstat_report_activity(STATE_IDLE, NULL); +} + +/* * Handle ORIGIN message. * * TODO, support tracking of multiple origins @@ -732,6 +1113,7 @@ apply_handle_origin(StringInfo s) * remote transaction and before any actual writes. */ if (!in_streamed_transaction && + !psf_cur.is_spooling && (!in_remote_transaction || (IsTransactionState() && !am_tablesync_worker()))) ereport(ERROR, @@ -1150,6 +1532,9 @@ apply_handle_insert(StringInfo s) TupleTableSlot *remoteslot; MemoryContext oldctx; + if (prepare_spoolfile_handler(LOGICAL_REP_MSG_INSERT, s)) + return; + if (handle_streamed_transaction(LOGICAL_REP_MSG_INSERT, s)) return; @@ -1271,6 +1656,9 @@ apply_handle_update(StringInfo s) RangeTblEntry *target_rte; MemoryContext oldctx; + if (prepare_spoolfile_handler(LOGICAL_REP_MSG_UPDATE, s)) + return; + if (handle_streamed_transaction(LOGICAL_REP_MSG_UPDATE, s)) return; @@ -1429,6 +1817,9 @@ apply_handle_delete(StringInfo s) TupleTableSlot *remoteslot; MemoryContext oldctx; + if (prepare_spoolfile_handler(LOGICAL_REP_MSG_DELETE, s)) + return; + if (handle_streamed_transaction(LOGICAL_REP_MSG_DELETE, s)) return; @@ -1798,6 +2189,9 @@ apply_handle_truncate(StringInfo s) List *relids_logged = NIL; ListCell *lc; + if (prepare_spoolfile_handler(LOGICAL_REP_MSG_TRUNCATE, s)) + return; + if (handle_streamed_transaction(LOGICAL_REP_MSG_TRUNCATE, s)) return; @@ -1954,6 +2348,28 @@ apply_dispatch(StringInfo s) case LOGICAL_REP_MSG_STREAM_COMMIT: apply_handle_stream_commit(s); return; + + case LOGICAL_REP_MSG_BEGIN_PREPARE: + apply_handle_begin_prepare(s); + return; + + case LOGICAL_REP_MSG_PREPARE: + apply_handle_prepare(s); + return; + + case LOGICAL_REP_MSG_COMMIT_PREPARED: + apply_handle_commit_prepared(s); + return; + + case LOGICAL_REP_MSG_ROLLBACK_PREPARED: + apply_handle_rollback_prepared(s); + return; + + case LOGICAL_REP_MSG_STREAM_PREPARE: + /* Streaming with two-phase is not supported */ + ereport(ERROR, + (errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg("invalid logical replication message type \"%c\"", action))); } ereport(ERROR, @@ -2061,6 +2477,23 @@ LogicalRepApplyLoop(XLogRecPtr last_received) TimeLineID tli; /* + * Initialize the psf_hash table if we haven't yet. This will be used for + * the entire duration of the apply worker so create it in permanent + * context. + */ + if (psf_hash == NULL) + { + HASHCTL hash_ctl; + PsfHashEntry *hentry; + + hash_ctl.keysize = sizeof(hentry->name); + hash_ctl.entrysize = sizeof(PsfHashEntry); + hash_ctl.hcxt = ApplyContext; + psf_hash = hash_create("PrepareSpoolfileHash", 1024, &hash_ctl, + HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); + } + + /* * Init the ApplyMessageContext which we clean up after each replication * protocol message. */ @@ -2180,7 +2613,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received) /* confirm all writes so far */ send_feedback(last_received, false, false); - if (!in_remote_transaction && !in_streamed_transaction) + if (!in_remote_transaction && !in_streamed_transaction && !psf_cur.is_spooling) { /* * If we didn't get any transactions for a while there might be @@ -2927,6 +3360,9 @@ ApplyWorkerMain(Datum main_arg) /* Attach to slot */ logicalrep_worker_attach(worker_slot); + /* Arrange to delete any unwanted psf file(s) at proc-exit */ + on_proc_exit(prepare_spoolfile_on_proc_exit, 0); + /* Setup signal handling */ pqsignal(SIGHUP, SignalHandlerForConfigReload); pqsignal(SIGTERM, die); @@ -3103,3 +3539,332 @@ IsLogicalWorker(void) { return MyLogicalRepWorker != NULL; } + +/* + * Handle the PREPARE spoolfile (if any) + * + * It can be necessary to redirect the PREPARE messages to a spoolfile (see + * apply_handle_begin_prepare) and then replay them back at the COMMIT PREPARED + * time. If needed, this is the common function to do that file redirection. + * + * Returns true if the message was redirected to the spoolfile, false + * otherwise (regular mode). + */ +static bool +prepare_spoolfile_handler(LogicalRepMsgType action, StringInfo s) +{ + elog(DEBUG1, + "prepare_spoolfile_handler for action '%c'. %s write to spool file", + action, + psf_cur.is_spooling ? "Do" : "Don't"); + + if (!psf_cur.is_spooling) + return false; + + Assert(!in_streamed_transaction); + + /* write the change to the current file */ + prepare_spoolfile_write(action, s); + + return true; +} + +/* + * Create the spoolfile used to serialize the prepare messages. + */ +static void +prepare_spoolfile_create(char *path) +{ + bool file_found; + PsfHashEntry *hentry; + + elog(DEBUG1, "creating file \"%s\" for prepare changes", path); + + Assert(!psf_cur.is_spooling); + + /* check if the file already exists. */ + file_found = prepare_spoolfile_exists(path); + + if (!file_found) + { + elog(DEBUG1, "Not found file \"%s\". Create it.", path); + psf_cur.vfd = PathNameOpenFile(path, O_RDWR | O_CREAT | O_TRUNC | PG_BINARY); + if (psf_cur.vfd < 0) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not create file \"%s\": %m", path))); + } + else + { + /* + * Open the file and seek to the beginning because we always want to + * create/overwrite this file. + */ + elog(DEBUG1, "Found file \"%s\". Overwrite it.", path); + psf_cur.vfd = PathNameOpenFile(path, O_RDWR | O_CREAT | O_TRUNC | PG_BINARY); + if (psf_cur.vfd < 0) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not open file \"%s\": %m", path))); + } + + /* Create/Find the spoolfile entry in the psf_hash */ + hentry = (PsfHashEntry *) hash_search(psf_hash, path, + HASH_ENTER | HASH_FIND, NULL); + Assert(hentry); + memcpy(psf_cur.name, path, sizeof(psf_cur.name)); + psf_cur.cur_offset = 0; + hentry->allow_delete = true; + + /* Sanity checks */ + Assert(psf_cur.vfd >= 0); + Assert(psf_cur.cur_offset == 0); + Assert(prepare_spoolfile_exists(path)); +} + +/* + * Close the "current" spoolfile and unset the fd. + */ +static void +prepare_spoolfile_close() +{ + if (psf_cur.vfd >= 0) + FileClose(psf_cur.vfd); + + /* Mark this fd as not valid to use anymore. */ + psf_cur.is_spooling = false; + psf_cur.vfd = -1; + psf_cur.cur_offset = 0; +} + +/* + * Delete the specified psf spoolfile, and any HTAB associated with it. + */ +static void +prepare_spoolfile_delete(char *path) +{ + /* The current psf should be closed already, but make sure anyway. */ + prepare_spoolfile_close(); + + /* Delete the file off the disk. */ + unlink(path); + + /* Remove any entry from the psf_hash, if present */ + hash_search(psf_hash, path, HASH_REMOVE, NULL); +} + +/* + * Serialize a change to the prepare spoolfile for the current toplevel transaction. + * + * The change is serialized in a simple format, with length (not including + * the length), action code (identifying the message type) and message + * contents (without the subxact TransactionId value). + */ +static void +prepare_spoolfile_write(char action, StringInfo s) +{ + int len; + int bytes_written; + + Assert(psf_cur.is_spooling); + + elog(DEBUG1, "prepare_spoolfile_write: writing action '%c'", action); + + /* total on-disk size, including the action type character */ + len = (s->len - s->cursor) + sizeof(char); + + /* first write the size */ + bytes_written = FileWrite(psf_cur.vfd, (char *)&len, sizeof(len), + psf_cur.cur_offset, WAIT_EVENT_DATA_FILE_WRITE); + Assert(bytes_written == sizeof(len)); + psf_cur.cur_offset += bytes_written; + + /* then the action */ + bytes_written = FileWrite(psf_cur.vfd, &action, sizeof(action), + psf_cur.cur_offset, WAIT_EVENT_DATA_FILE_WRITE); + Assert(bytes_written == sizeof(action)); + psf_cur.cur_offset += bytes_written; + + /* and finally the remaining part of the buffer (after the XID) */ + len = (s->len - s->cursor); + + bytes_written = FileWrite(psf_cur.vfd, &s->data[s->cursor], len, + psf_cur.cur_offset, WAIT_EVENT_DATA_FILE_WRITE); + Assert(bytes_written == len); + psf_cur.cur_offset += bytes_written; +} + +/* + * Is there a prepare spoolfile for the specified path? + */ +static bool +prepare_spoolfile_exists(char *path) +{ + bool found; + + File fd = PathNameOpenFile(path, O_RDONLY | PG_BINARY); + + found = fd >= 0; + if (fd >= 0) + FileClose(fd); + + return found; +} + +/* + * Replay (apply) all the prepared messages that are in the prepare spoolfile. + */ +static int +prepare_spoolfile_replay_messages(char *path, XLogRecPtr final_lsn) +{ + StringInfoData s2; + int nchanges = 0; + char *buffer = NULL; + MemoryContext oldctx, + oldctx2; + PsfFile psf = { .is_spooling = false, .vfd = -1, .cur_offset = 0 }; + + elog(DEBUG1, + "prepare_spoolfile_replay_messages: replaying changes from file \"%s\"", + path); + + /* + * Allocate memory required to process all the messages in + * TopTransactionContext to avoid it getting reset after each message is + * processed. + */ + oldctx = MemoryContextSwitchTo(TopTransactionContext); + + psf.vfd = PathNameOpenFile(path, O_RDONLY | PG_BINARY); + if (psf.vfd < 0) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not read from prepared spoolfile \"%s\": %m", + path))); + + buffer = palloc(BLCKSZ); + initStringInfo(&s2); + + MemoryContextSwitchTo(oldctx); + + /* + * Make sure the handle apply_dispatch methods are aware we're in a remote + * transaction. + */ + remote_final_lsn = final_lsn; + in_remote_transaction = true; + pgstat_report_activity(STATE_RUNNING, NULL); + + /* + * Read the entries one by one and pass them through the same logic as in + * apply_dispatch. + */ + while (true) + { + int nbytes; + int len; + + CHECK_FOR_INTERRUPTS(); + + /* read length of the on-disk record */ + nbytes = FileRead(psf.vfd, (char *) &len, sizeof(len), + psf.cur_offset, WAIT_EVENT_DATA_FILE_READ); + psf.cur_offset += nbytes; + + /* have we reached end of the file? */ + if (nbytes == 0) + break; + + /* do we have a correct length? */ + if (nbytes != sizeof(len)) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not read from prepared spoolfile \"%s\": %m", + path))); + + Assert(len > 0); + + /* make sure we have sufficiently large buffer */ + buffer = repalloc(buffer, len); + + /* and finally read the data into the buffer */ + nbytes = FileRead(psf.vfd, buffer, len, + psf.cur_offset, WAIT_EVENT_DATA_FILE_READ); + psf.cur_offset += nbytes; + if (nbytes != len) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not read from prepared spoolfile \"%s\": %m", + path))); + + /* copy the buffer to the stringinfo and call apply_dispatch */ + resetStringInfo(&s2); + appendBinaryStringInfo(&s2, buffer, len); + + /* Ensure we are reading the data into our memory context. */ + oldctx2 = MemoryContextSwitchTo(ApplyMessageContext); + + apply_dispatch(&s2); + + MemoryContextReset(ApplyMessageContext); + + MemoryContextSwitchTo(oldctx2); + + nchanges++; + + if (nchanges % 1000 == 0) + elog(DEBUG1, "replayed %d changes from file '%s'", + nchanges, path); + } + + FileClose(psf.vfd); + + elog(DEBUG1, "replayed %d (all) changes from file \"%s\"", + nchanges, path); + + return nchanges; +} + +/* + * Format the filename for the prepare spoolfile. + */ +static void +prepare_spoolfile_name(char *path, int szpath, Oid subid, char *gid) +{ + PsfHashEntry *hentry; + + /* + * This name is used as the key in the psf_hash HTAB. + * + * Therefore, the name and the key must be exactly same lengths and padded + * with '\0' so garbage does not impact the HTAB lookups. + */ + Assert(sizeof(hentry->name) == MAXPGPATH); + Assert(szpath == MAXPGPATH); + memset(path, '\0', MAXPGPATH); + + snprintf(path, MAXPGPATH, "pg_twophase/%u-%s.prep_changes", subid, gid); +} + +/* + * proc_exit callback to remove unwanted psf files. + */ +static void +prepare_spoolfile_on_proc_exit(int status, Datum arg) +{ + HASH_SEQ_STATUS seq_status; + PsfHashEntry *hentry; + + /* Iterate the HTAB looking for what file can be deleted. */ + if (psf_hash) + { + hash_seq_init(&seq_status, psf_hash); + while ((hentry = (PsfHashEntry *) hash_seq_search(&seq_status)) != NULL) + { + char *path = hentry->name; + + if (hentry->allow_delete) + prepare_spoolfile_delete(path); + } + } +} diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c index 1b993fb..2e4b39f 100644 --- a/src/backend/replication/pgoutput/pgoutput.c +++ b/src/backend/replication/pgoutput/pgoutput.c @@ -47,6 +47,16 @@ static void pgoutput_truncate(LogicalDecodingContext *ctx, ReorderBufferChange *change); static bool pgoutput_origin_filter(LogicalDecodingContext *ctx, RepOriginId origin_id); +static void pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx, + ReorderBufferTXN *txn); +static void pgoutput_prepare_txn(LogicalDecodingContext *ctx, + ReorderBufferTXN *txn, XLogRecPtr prepare_lsn); +static void pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, + ReorderBufferTXN *txn, XLogRecPtr commit_lsn); +static void pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx, + ReorderBufferTXN *txn, + XLogRecPtr prepare_end_lsn, + TimestampTz prepare_time); static void pgoutput_stream_start(struct LogicalDecodingContext *ctx, ReorderBufferTXN *txn); static void pgoutput_stream_stop(struct LogicalDecodingContext *ctx, @@ -66,6 +76,9 @@ static void publication_invalidation_cb(Datum arg, int cacheid, uint32 hashvalue); static void send_relation_and_attrs(Relation relation, TransactionId xid, LogicalDecodingContext *ctx); +static void send_repl_origin(LogicalDecodingContext* ctx, + RepOriginId origin_id, XLogRecPtr origin_lsn, + bool send_origin); /* * Entry in the map used to remember which relation schemas we sent. @@ -143,6 +156,11 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb) cb->change_cb = pgoutput_change; cb->truncate_cb = pgoutput_truncate; cb->commit_cb = pgoutput_commit_txn; + + cb->begin_prepare_cb = pgoutput_begin_prepare_txn; + cb->prepare_cb = pgoutput_prepare_txn; + cb->commit_prepared_cb = pgoutput_commit_prepared_txn; + cb->rollback_prepared_cb = pgoutput_rollback_prepared_txn; cb->filter_by_origin_cb = pgoutput_origin_filter; cb->shutdown_cb = pgoutput_shutdown; @@ -153,6 +171,8 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb) cb->stream_commit_cb = pgoutput_stream_commit; cb->stream_change_cb = pgoutput_change; cb->stream_truncate_cb = pgoutput_truncate; + /* transaction streaming - two-phase commit */ + cb->stream_prepare_cb = NULL; } static void @@ -322,8 +342,12 @@ pgoutput_startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt, } else { - /* Disable the streaming during the slot initialization mode. */ + /* + * Disable the streaming and prepared transactions during the slot + * initialization mode. + */ ctx->streaming = false; + ctx->twophase = false; } } @@ -338,29 +362,8 @@ pgoutput_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn) OutputPluginPrepareWrite(ctx, !send_replication_origin); logicalrep_write_begin(ctx->out, txn); - if (send_replication_origin) - { - char *origin; - - /*---------- - * XXX: which behaviour do we want here? - * - * Alternatives: - * - don't send origin message if origin name not found - * (that's what we do now) - * - throw error - that will break replication, not good - * - send some special "unknown" origin - *---------- - */ - if (replorigin_by_oid(txn->origin_id, true, &origin)) - { - /* Message boundary */ - OutputPluginWrite(ctx, false); - OutputPluginPrepareWrite(ctx, true); - logicalrep_write_origin(ctx->out, origin, txn->origin_lsn); - } - - } + send_repl_origin(ctx, txn->origin_id, txn->origin_lsn, + send_replication_origin); OutputPluginWrite(ctx, true); } @@ -380,6 +383,68 @@ pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, } /* + * BEGIN PREPARE callback + */ +static void +pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn) +{ + bool send_replication_origin = txn->origin_id != InvalidRepOriginId; + + OutputPluginPrepareWrite(ctx, !send_replication_origin); + logicalrep_write_begin_prepare(ctx->out, txn); + + send_repl_origin(ctx, txn->origin_id, txn->origin_lsn, + send_replication_origin); + + OutputPluginWrite(ctx, true); +} + +/* + * PREPARE callback + */ +static void +pgoutput_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, + XLogRecPtr prepare_lsn) +{ + OutputPluginUpdateProgress(ctx); + + OutputPluginPrepareWrite(ctx, true); + logicalrep_write_prepare(ctx->out, txn, prepare_lsn); + OutputPluginWrite(ctx, true); +} + +/* + * COMMIT PREPARED callback + */ +static void +pgoutput_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, + XLogRecPtr commit_lsn) +{ + OutputPluginUpdateProgress(ctx); + + OutputPluginPrepareWrite(ctx, true); + logicalrep_write_commit_prepared(ctx->out, txn, commit_lsn); + OutputPluginWrite(ctx, true); +} + +/* + * ROLLBACK PREPARED callback + */ +static void +pgoutput_rollback_prepared_txn(LogicalDecodingContext *ctx, + ReorderBufferTXN *txn, + XLogRecPtr prepare_end_lsn, + TimestampTz prepare_time) +{ + OutputPluginUpdateProgress(ctx); + + OutputPluginPrepareWrite(ctx, true); + logicalrep_write_rollback_prepared(ctx->out, txn, prepare_end_lsn, + prepare_time); + OutputPluginWrite(ctx, true); +} + +/* * Write the current schema of the relation and its ancestor (if any) if not * done yet. */ @@ -778,18 +843,8 @@ pgoutput_stream_start(struct LogicalDecodingContext *ctx, OutputPluginPrepareWrite(ctx, !send_replication_origin); logicalrep_write_stream_start(ctx->out, txn->xid, !rbtxn_is_streamed(txn)); - if (send_replication_origin) - { - char *origin; - - if (replorigin_by_oid(txn->origin_id, true, &origin)) - { - /* Message boundary */ - OutputPluginWrite(ctx, false); - OutputPluginPrepareWrite(ctx, true); - logicalrep_write_origin(ctx->out, origin, InvalidXLogRecPtr); - } - } + send_repl_origin(ctx, txn->origin_id, InvalidXLogRecPtr, + send_replication_origin); OutputPluginWrite(ctx, true); @@ -1195,3 +1250,33 @@ rel_sync_cache_publication_cb(Datum arg, int cacheid, uint32 hashvalue) entry->pubactions.pubtruncate = false; } } + +/* Send Replication origin */ +static void +send_repl_origin(LogicalDecodingContext *ctx, RepOriginId origin_id, + XLogRecPtr origin_lsn, bool send_origin) +{ + if (send_origin) + { + char *origin; + + /*---------- + * XXX: which behaviour do we want here? + * + * Alternatives: + * - don't send origin message if origin name not found + * (that's what we do now) + * - throw error - that will break replication, not good + * - send some special "unknown" origin + *---------- + */ + if (replorigin_by_oid(origin_id, true, &origin)) + { + /* Message boundary */ + OutputPluginWrite(ctx, false); + OutputPluginPrepareWrite(ctx, true); + + logicalrep_write_origin(ctx->out, origin, origin_lsn); + } + } +} diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y index eb283a8..8c1f353 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 @@ -241,16 +243,17 @@ create_replication_slot: cmd->options = $5; $$ = (Node *) cmd; } - /* CREATE_REPLICATION_SLOT slot TEMPORARY LOGICAL plugin */ - | K_CREATE_REPLICATION_SLOT IDENT opt_temporary K_LOGICAL IDENT create_slot_opt_list + /* CREATE_REPLICATION_SLOT slot TEMPORARY TWO_PHASE LOGICAL plugin */ + | 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/access/twophase.h b/src/include/access/twophase.h index 91786da..e27e1a8 100644 --- a/src/include/access/twophase.h +++ b/src/include/access/twophase.h @@ -58,4 +58,6 @@ extern void PrepareRedoAdd(char *buf, XLogRecPtr start_lsn, XLogRecPtr end_lsn, RepOriginId origin_id); extern void PrepareRedoRemove(TransactionId xid, bool giveWarning); extern void restoreTwoPhaseData(void); +extern bool LookupGXact(const char *gid, XLogRecPtr prepare_at_lsn, + TimestampTz origin_prepare_timestamp); #endif /* TWOPHASE_H */ diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h index fa4c372..232af01 100644 --- a/src/include/replication/logicalproto.h +++ b/src/include/replication/logicalproto.h @@ -13,6 +13,7 @@ #ifndef LOGICAL_PROTO_H #define LOGICAL_PROTO_H +#include "access/xact.h" #include "replication/reorderbuffer.h" #include "utils/rel.h" @@ -54,10 +55,15 @@ typedef enum LogicalRepMsgType LOGICAL_REP_MSG_TRUNCATE = 'T', LOGICAL_REP_MSG_RELATION = 'R', LOGICAL_REP_MSG_TYPE = 'Y', + LOGICAL_REP_MSG_BEGIN_PREPARE = 'b', + LOGICAL_REP_MSG_PREPARE = 'P', + LOGICAL_REP_MSG_COMMIT_PREPARED = 'K', + LOGICAL_REP_MSG_ROLLBACK_PREPARED = 'r', LOGICAL_REP_MSG_STREAM_START = 'S', LOGICAL_REP_MSG_STREAM_END = 'E', LOGICAL_REP_MSG_STREAM_COMMIT = 'c', - LOGICAL_REP_MSG_STREAM_ABORT = 'A' + LOGICAL_REP_MSG_STREAM_ABORT = 'A', + LOGICAL_REP_MSG_STREAM_PREPARE = 'p' } LogicalRepMsgType; /* @@ -114,6 +120,7 @@ typedef struct LogicalRepBeginData TransactionId xid; } LogicalRepBeginData; +/* Commit (and abort) information */ typedef struct LogicalRepCommitData { XLogRecPtr commit_lsn; @@ -121,6 +128,48 @@ typedef struct LogicalRepCommitData TimestampTz committime; } LogicalRepCommitData; + +/* Begin Prepare information */ +typedef struct LogicalRepBeginPrepareData +{ + XLogRecPtr final_lsn; + XLogRecPtr end_lsn; + TimestampTz committime; + TransactionId xid; + char gid[GIDSIZE]; +} LogicalRepBeginPrepareData; + +/* + * Prepared transaction protocol information. This same structure is used to + * hold the information for prepare, and commit prepared transaction. + * prepare_lsn and preparetime are used to store commit lsn and + * commit time for commit prepared. + */ +typedef struct LogicalRepPreparedTxnData +{ + XLogRecPtr prepare_lsn; + XLogRecPtr end_lsn; + TimestampTz preparetime; + char gid[GIDSIZE]; +} LogicalRepPreparedTxnData; + +/* + * Rollback Prepared transaction protocol information. The prepare information + * prepare_end_lsn and prepare_time are used to check if the downstream has + * received this prepared transaction in which case it can apply the rollback, + * otherwise, it can skip the rollback operation. The gid alone is not + * sufficient because the downstream node can have a prepared transaction with + * same identifier. + */ +typedef struct LogicalRepRollbackPreparedTxnData +{ + XLogRecPtr prepare_end_lsn; + XLogRecPtr rollback_end_lsn; + TimestampTz preparetime; + TimestampTz rollbacktime; + char gid[GIDSIZE]; +} LogicalRepRollbackPreparedTxnData; + extern void logicalrep_write_begin(StringInfo out, ReorderBufferTXN *txn); extern void logicalrep_read_begin(StringInfo in, LogicalRepBeginData *begin_data); @@ -128,6 +177,24 @@ extern void logicalrep_write_commit(StringInfo out, ReorderBufferTXN *txn, XLogRecPtr commit_lsn); extern void logicalrep_read_commit(StringInfo in, LogicalRepCommitData *commit_data); +extern void logicalrep_write_begin_prepare(StringInfo out, ReorderBufferTXN *txn); +extern void logicalrep_read_begin_prepare(StringInfo in, + LogicalRepBeginPrepareData *begin_data); +extern void logicalrep_write_prepare(StringInfo out, ReorderBufferTXN *txn, + XLogRecPtr prepare_lsn); +extern void logicalrep_read_prepare(StringInfo in, + LogicalRepPreparedTxnData *prepare_data); +extern void logicalrep_write_commit_prepared(StringInfo out, ReorderBufferTXN *txn, + XLogRecPtr commit_lsn); +extern void logicalrep_read_commit_prepared(StringInfo in, + LogicalRepPreparedTxnData *prepare_data); +extern void logicalrep_write_rollback_prepared(StringInfo out, ReorderBufferTXN *txn, + XLogRecPtr prepare_end_lsn, + TimestampTz prepare_time); +extern void logicalrep_read_rollback_prepared(StringInfo in, + LogicalRepRollbackPreparedTxnData *rollback_data); + + extern void logicalrep_write_origin(StringInfo out, const char *origin, XLogRecPtr origin_lsn); extern char *logicalrep_read_origin(StringInfo in, XLogRecPtr *origin_lsn); diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h index 565a961..0c95dc6 100644 --- a/src/include/replication/reorderbuffer.h +++ b/src/include/replication/reorderbuffer.h @@ -247,6 +247,18 @@ typedef struct ReorderBufferChange ((txn)->txn_flags & RBTXN_SKIPPED_PREPARE) != 0 \ ) +/* Has this prepared transaction been committed? */ +#define rbtxn_commit_prepared(txn) \ +( \ + ((txn)->txn_flags & RBTXN_COMMIT_PREPARED) != 0 \ +) + +/* Has this prepared transaction been rollbacked? */ +#define rbtxn_rollback_prepared(txn) \ +( \ + ((txn)->txn_flags & RBTXN_ROLLBACK_PREPARED) != 0 \ +) + typedef struct ReorderBufferTXN { /* See above */ 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) \ diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 1cac75e..95d78e9 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -86,6 +86,9 @@ extern void ReplicationOriginNameForTablesync(Oid suboid, Oid relid, char *originname, int szorgname); extern char *LogicalRepSyncTableStart(XLogRecPtr *origin_startpos); +extern bool AnyTablesyncInProgress(void); +extern XLogRecPtr BiggestTablesyncLSN(void); + void process_syncing_tables(XLogRecPtr current_lsn); void invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 8bd95ae..4ffcef5 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1341,12 +1341,15 @@ LogicalOutputPluginWriterPrepareWrite LogicalOutputPluginWriterUpdateProgress LogicalOutputPluginWriterWrite LogicalRepBeginData +LogicalRepBeginPrepareData LogicalRepCommitData LogicalRepCtxStruct LogicalRepPartMapEntry +LogicalRepPreparedTxnData LogicalRepRelId LogicalRepRelMapEntry LogicalRepRelation +LogicalRepRollbackPreparedTxnData LogicalRepTupleData LogicalRepTyp LogicalRepWorker @@ -1955,6 +1958,8 @@ ProtocolVersion PrsStorage PruneState PruneStepResult +PsfFile +PsfHashEntry PsqlScanCallbacks PsqlScanQuoteType PsqlScanResult -- 1.8.3.1