From a8fd13cab0ba815e9925dc9676e6309f699b5f72 Mon Sep 17 00:00:00 2001 From: Amit Kapila Date: Wed, 14 Jul 2021 07:33:50 +0530 Subject: Add support for prepared transactions 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 following 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. * Add a new SUBSCRIPTION option "two_phase" to allow users to enable two-phase transactions. We enable the two_phase once the initial data sync is over. 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. The streaming option is not allowed with this new two_phase option. This can be done as a separate patch. We don't allow to toggle two_phase option of a subscription because it can lead to an inconsistent replica. For the same reason, we don't allow to refresh the publication once the two_phase is enabled for a subscription unless copy_data option is false. Author: Peter Smith, Ajin Cherian and Amit Kapila based on previous work by Nikhil Sontakke and Stas Kelvich Reviewed-by: Amit Kapila, Sawada Masahiko, Vignesh C, Dilip Kumar, Takamichi Osumi, Greg Nancarrow Tested-By: Haiying Tang Discussion: https://postgr.es/m/02DA5F5E-CECE-4D9C-8B4B-418077E2C010@postgrespro.ru Discussion: https://postgr.es/m/CAA4eK1+opiV4aFTmWWUF9h_32=HfPOW9vZASHarT0UA5oBrtGw@mail.gmail.com --- src/backend/replication/logical/reorderbuffer.c | 25 +++++++++++++------------ 1 file changed, 13 insertions(+), 12 deletions(-) (limited to 'src/backend/replication/logical/reorderbuffer.c') diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 1b4f4a528aa..7378beb684d 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -2576,7 +2576,7 @@ ReorderBufferReplay(ReorderBufferTXN *txn, txn->final_lsn = commit_lsn; txn->end_lsn = end_lsn; - txn->commit_time = commit_time; + txn->xact_time.commit_time = commit_time; txn->origin_id = origin_id; txn->origin_lsn = origin_lsn; @@ -2667,7 +2667,7 @@ ReorderBufferRememberPrepareInfo(ReorderBuffer *rb, TransactionId xid, */ txn->final_lsn = prepare_lsn; txn->end_lsn = end_lsn; - txn->commit_time = prepare_time; + txn->xact_time.prepare_time = prepare_time; txn->origin_id = origin_id; txn->origin_lsn = origin_lsn; @@ -2714,7 +2714,7 @@ ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid, Assert(txn->final_lsn != InvalidXLogRecPtr); ReorderBufferReplay(txn, rb, xid, txn->final_lsn, txn->end_lsn, - txn->commit_time, txn->origin_id, txn->origin_lsn); + txn->xact_time.prepare_time, txn->origin_id, txn->origin_lsn); /* * We send the prepare for the concurrently aborted xacts so that later @@ -2734,7 +2734,7 @@ ReorderBufferPrepare(ReorderBuffer *rb, TransactionId xid, void ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid, XLogRecPtr commit_lsn, XLogRecPtr end_lsn, - XLogRecPtr initial_consistent_point, + XLogRecPtr two_phase_at, TimestampTz commit_time, RepOriginId origin_id, XLogRecPtr origin_lsn, char *gid, bool is_commit) { @@ -2753,19 +2753,20 @@ ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid, * be later used for rollback. */ prepare_end_lsn = txn->end_lsn; - prepare_time = txn->commit_time; + prepare_time = txn->xact_time.prepare_time; /* add the gid in the txn */ txn->gid = pstrdup(gid); /* * It is possible that this transaction is not decoded at prepare time - * either because by that time we didn't have a consistent snapshot or it - * was decoded earlier but we have restarted. We only need to send the - * prepare if it was not decoded earlier. We don't need to decode the xact - * for aborts if it is not done already. + * either because by that time we didn't have a consistent snapshot, or + * two_phase was not enabled, or it was decoded earlier but we have + * restarted. We only need to send the prepare if it was not decoded + * earlier. We don't need to decode the xact for aborts if it is not done + * already. */ - if ((txn->final_lsn < initial_consistent_point) && is_commit) + if ((txn->final_lsn < two_phase_at) && is_commit) { txn->txn_flags |= RBTXN_PREPARE; @@ -2783,12 +2784,12 @@ ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid, * prepared after the restart. */ ReorderBufferReplay(txn, rb, xid, txn->final_lsn, txn->end_lsn, - txn->commit_time, txn->origin_id, txn->origin_lsn); + txn->xact_time.prepare_time, txn->origin_id, txn->origin_lsn); } txn->final_lsn = commit_lsn; txn->end_lsn = end_lsn; - txn->commit_time = commit_time; + txn->xact_time.commit_time = commit_time; txn->origin_id = origin_id; txn->origin_lsn = origin_lsn; -- cgit v1.2.3