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 --- contrib/test_decoding/test_decoding.c | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) (limited to 'contrib/test_decoding/test_decoding.c') diff --git a/contrib/test_decoding/test_decoding.c b/contrib/test_decoding/test_decoding.c index de1b6926581..e5cd84e85e4 100644 --- a/contrib/test_decoding/test_decoding.c +++ b/contrib/test_decoding/test_decoding.c @@ -339,7 +339,7 @@ pg_decode_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, if (data->include_timestamp) appendStringInfo(ctx->out, " (at %s)", - timestamptz_to_str(txn->commit_time)); + timestamptz_to_str(txn->xact_time.commit_time)); OutputPluginWrite(ctx, true); } @@ -382,7 +382,7 @@ pg_decode_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, if (data->include_timestamp) appendStringInfo(ctx->out, " (at %s)", - timestamptz_to_str(txn->commit_time)); + timestamptz_to_str(txn->xact_time.prepare_time)); OutputPluginWrite(ctx, true); } @@ -404,7 +404,7 @@ pg_decode_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn if (data->include_timestamp) appendStringInfo(ctx->out, " (at %s)", - timestamptz_to_str(txn->commit_time)); + timestamptz_to_str(txn->xact_time.commit_time)); OutputPluginWrite(ctx, true); } @@ -428,7 +428,7 @@ pg_decode_rollback_prepared_txn(LogicalDecodingContext *ctx, if (data->include_timestamp) appendStringInfo(ctx->out, " (at %s)", - timestamptz_to_str(txn->commit_time)); + timestamptz_to_str(txn->xact_time.commit_time)); OutputPluginWrite(ctx, true); } @@ -853,7 +853,7 @@ pg_decode_stream_prepare(LogicalDecodingContext *ctx, if (data->include_timestamp) appendStringInfo(ctx->out, " (at %s)", - timestamptz_to_str(txn->commit_time)); + timestamptz_to_str(txn->xact_time.prepare_time)); OutputPluginWrite(ctx, true); } @@ -882,7 +882,7 @@ pg_decode_stream_commit(LogicalDecodingContext *ctx, if (data->include_timestamp) appendStringInfo(ctx->out, " (at %s)", - timestamptz_to_str(txn->commit_time)); + timestamptz_to_str(txn->xact_time.commit_time)); OutputPluginWrite(ctx, true); } -- cgit v1.2.3