Skip to content

Commit a8fd13c

Browse files
author
Amit Kapila
committed
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/[email protected] Discussion: https://postgr.es/m/CAA4eK1+opiV4aFTmWWUF9h_32=HfPOW9vZASHarT0UA5oBrtGw@mail.gmail.com
1 parent 6c9c283 commit a8fd13c

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

43 files changed

+2382
-191
lines changed

‎contrib/test_decoding/test_decoding.c‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -339,7 +339,7 @@ pg_decode_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
339339

340340
if (data->include_timestamp)
341341
appendStringInfo(ctx->out, " (at %s)",
342-
timestamptz_to_str(txn->commit_time));
342+
timestamptz_to_str(txn->xact_time.commit_time));
343343

344344
OutputPluginWrite(ctx, true);
345345
}
@@ -382,7 +382,7 @@ pg_decode_prepare_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
382382

383383
if (data->include_timestamp)
384384
appendStringInfo(ctx->out, " (at %s)",
385-
timestamptz_to_str(txn->commit_time));
385+
timestamptz_to_str(txn->xact_time.prepare_time));
386386

387387
OutputPluginWrite(ctx, true);
388388
}
@@ -404,7 +404,7 @@ pg_decode_commit_prepared_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn
404404

405405
if (data->include_timestamp)
406406
appendStringInfo(ctx->out, " (at %s)",
407-
timestamptz_to_str(txn->commit_time));
407+
timestamptz_to_str(txn->xact_time.commit_time));
408408

409409
OutputPluginWrite(ctx, true);
410410
}
@@ -428,7 +428,7 @@ pg_decode_rollback_prepared_txn(LogicalDecodingContext *ctx,
428428

429429
if (data->include_timestamp)
430430
appendStringInfo(ctx->out, " (at %s)",
431-
timestamptz_to_str(txn->commit_time));
431+
timestamptz_to_str(txn->xact_time.commit_time));
432432

433433
OutputPluginWrite(ctx, true);
434434
}
@@ -853,7 +853,7 @@ pg_decode_stream_prepare(LogicalDecodingContext *ctx,
853853

854854
if (data->include_timestamp)
855855
appendStringInfo(ctx->out, " (at %s)",
856-
timestamptz_to_str(txn->commit_time));
856+
timestamptz_to_str(txn->xact_time.prepare_time));
857857

858858
OutputPluginWrite(ctx, true);
859859
}
@@ -882,7 +882,7 @@ pg_decode_stream_commit(LogicalDecodingContext *ctx,
882882

883883
if (data->include_timestamp)
884884
appendStringInfo(ctx->out, " (at %s)",
885-
timestamptz_to_str(txn->commit_time));
885+
timestamptz_to_str(txn->xact_time.commit_time));
886886

887887
OutputPluginWrite(ctx, true);
888888
}

‎doc/src/sgml/catalogs.sgml‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7641,6 +7641,18 @@ SCRAM-SHA-256$<replaceable>&lt;iteration count&gt;</replaceable>:<replaceable>&l
76417641
</para></entry>
76427642
</row>
76437643

7644+
<row>
7645+
<entry role="catalog_table_entry"><para role="column_definition">
7646+
<structfield>subtwophasestate</structfield> <type>char</type>
7647+
</para>
7648+
<para>
7649+
State codes for two-phase mode:
7650+
<literal>d</literal> = disabled,
7651+
<literal>p</literal> = pending enablement,
7652+
<literal>e</literal> = enabled
7653+
</para></entry>
7654+
</row>
7655+
76447656
<row>
76457657
<entry role="catalog_table_entry"><para role="column_definition">
76467658
<structfield>subconninfo</structfield> <type>text</type>

‎doc/src/sgml/protocol.sgml‎

Lines changed: 282 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -2811,11 +2811,17 @@ The commands accepted in replication mode are:
28112811
</term>
28122812
<listitem>
28132813
<para>
2814-
Protocol version. Currently versions <literal>1</literal> and
2815-
<literal>2</literal> are supported. The version <literal>2</literal>
2816-
is supported only for server version 14 and above, and it allows
2817-
streaming of large in-progress transactions.
2818-
</para>
2814+
Protocol version. Currently versions <literal>1</literal>, <literal>2</literal>,
2815+
and <literal>3</literal> are supported.
2816+
</para>
2817+
<para>
2818+
Version <literal>2</literal> is supported only for server version 14
2819+
and above, and it allows streaming of large in-progress transactions.
2820+
</para>
2821+
<para>
2822+
Version <literal>3</literal> is supported only for server version 15
2823+
and above, and it allows streaming of two-phase transactions.
2824+
</para>
28192825
</listitem>
28202826
</varlistentry>
28212827

@@ -2871,10 +2877,11 @@ The commands accepted in replication mode are:
28712877
<para>
28722878
The logical replication protocol sends individual transactions one by one.
28732879
This means that all messages between a pair of Begin and Commit messages
2874-
belong to the same transaction. It also sends changes of large in-progress
2875-
transactions between a pair of Stream Start and Stream Stop messages. The
2876-
last stream of such a transaction contains Stream Commit or Stream Abort
2877-
message.
2880+
belong to the same transaction. Similarly, all messages between a pair of
2881+
Begin Prepare and Prepare messages belong to the same transaction.
2882+
It also sends changes of large in-progress transactions between a pair of
2883+
Stream Start and Stream Stop messages. The last stream of such a transaction
2884+
contains a Stream Commit or Stream Abort message.
28782885
</para>
28792886

28802887
<para>
@@ -7390,6 +7397,272 @@ Stream Abort
73907397

73917398
</variablelist>
73927399

7400+
<para>
7401+
The following messages (Begin Prepare, Prepare, Commit Prepared, Rollback Prepared)
7402+
are available since protocol version 3.
7403+
</para>
7404+
7405+
<variablelist>
7406+
7407+
<varlistentry>
7408+
7409+
<term>Begin Prepare</term>
7410+
<listitem>
7411+
<para>
7412+
7413+
<variablelist>
7414+
7415+
<varlistentry>
7416+
<term>Byte1('b')</term>
7417+
<listitem><para>
7418+
Identifies the message as the beginning of a two-phase transaction message.
7419+
</para></listitem>
7420+
</varlistentry>
7421+
7422+
<varlistentry>
7423+
<term>Int64</term>
7424+
<listitem><para>
7425+
The LSN of the prepare.
7426+
</para></listitem>
7427+
</varlistentry>
7428+
7429+
<varlistentry>
7430+
<term>Int64</term>
7431+
<listitem><para>
7432+
The end LSN of the prepared transaction.
7433+
</para></listitem>
7434+
</varlistentry>
7435+
7436+
<varlistentry>
7437+
<term>Int64</term>
7438+
<listitem><para>
7439+
Prepare timestamp of the transaction. The value is in number
7440+
of microseconds since PostgreSQL epoch (2000-01-01).
7441+
</para></listitem>
7442+
</varlistentry>
7443+
7444+
<varlistentry>
7445+
<term>Int32</term>
7446+
<listitem><para>
7447+
Xid of the transaction.
7448+
</para></listitem>
7449+
</varlistentry>
7450+
7451+
<varlistentry>
7452+
<term>String</term>
7453+
<listitem><para>
7454+
The user defined GID of the two-phase transaction.
7455+
</para></listitem>
7456+
</varlistentry>
7457+
7458+
</variablelist>
7459+
7460+
</para>
7461+
</listitem>
7462+
</varlistentry>
7463+
7464+
<varlistentry>
7465+
7466+
<term>Prepare</term>
7467+
<listitem>
7468+
<para>
7469+
7470+
<variablelist>
7471+
7472+
<varlistentry>
7473+
<term>Byte1('P')</term>
7474+
<listitem><para>
7475+
Identifies the message as a two-phase prepared transaction message.
7476+
</para></listitem>
7477+
</varlistentry>
7478+
7479+
<varlistentry>
7480+
<term>Int8</term>
7481+
<listitem><para>
7482+
Flags; currently unused (must be 0).
7483+
</para></listitem>
7484+
</varlistentry>
7485+
7486+
<varlistentry>
7487+
<term>Int64</term>
7488+
<listitem><para>
7489+
The LSN of the prepare.
7490+
</para></listitem>
7491+
</varlistentry>
7492+
7493+
<varlistentry>
7494+
<term>Int64</term>
7495+
<listitem><para>
7496+
The end LSN of the prepared transaction.
7497+
</para></listitem>
7498+
</varlistentry>
7499+
7500+
<varlistentry>
7501+
<term>Int64</term>
7502+
<listitem><para>
7503+
Prepare timestamp of the transaction. The value is in number
7504+
of microseconds since PostgreSQL epoch (2000-01-01).
7505+
</para></listitem>
7506+
</varlistentry>
7507+
7508+
<varlistentry>
7509+
<term>Int32</term>
7510+
<listitem><para>
7511+
Xid of the transaction.
7512+
</para></listitem>
7513+
</varlistentry>
7514+
7515+
<varlistentry>
7516+
<term>String</term>
7517+
<listitem><para>
7518+
The user defined GID of the two-phase transaction.
7519+
</para></listitem>
7520+
</varlistentry>
7521+
7522+
</variablelist>
7523+
7524+
</para>
7525+
</listitem>
7526+
</varlistentry>
7527+
7528+
<varlistentry>
7529+
7530+
<term>Commit Prepared</term>
7531+
<listitem>
7532+
<para>
7533+
7534+
<variablelist>
7535+
7536+
<varlistentry>
7537+
<term>Byte1('K')</term>
7538+
<listitem><para>
7539+
Identifies the message as the commit of a two-phase transaction message.
7540+
</para></listitem>
7541+
</varlistentry>
7542+
7543+
<varlistentry>
7544+
<term>Int8</term>
7545+
<listitem><para>
7546+
Flags; currently unused (must be 0).
7547+
</para></listitem>
7548+
</varlistentry>
7549+
7550+
<varlistentry>
7551+
<term>Int64</term>
7552+
<listitem><para>
7553+
The LSN of the commit prepared.
7554+
</para></listitem>
7555+
</varlistentry>
7556+
7557+
<varlistentry>
7558+
<term>Int64</term>
7559+
<listitem><para>
7560+
The end LSN of the commit prepared transaction.
7561+
</para></listitem>
7562+
</varlistentry>
7563+
7564+
<varlistentry>
7565+
<term>Int64</term>
7566+
<listitem><para>
7567+
Commit timestamp of the transaction. The value is in number
7568+
of microseconds since PostgreSQL epoch (2000-01-01).
7569+
</para></listitem>
7570+
</varlistentry>
7571+
7572+
<varlistentry>
7573+
<term>Int32</term>
7574+
<listitem><para>
7575+
Xid of the transaction.
7576+
</para></listitem>
7577+
</varlistentry>
7578+
7579+
<varlistentry>
7580+
<term>String</term>
7581+
<listitem><para>
7582+
The user defined GID of the two-phase transaction.
7583+
</para></listitem>
7584+
</varlistentry>
7585+
7586+
</variablelist>
7587+
7588+
</para>
7589+
</listitem>
7590+
</varlistentry>
7591+
7592+
<varlistentry>
7593+
7594+
<term>Rollback Prepared</term>
7595+
<listitem>
7596+
<para>
7597+
7598+
<variablelist>
7599+
7600+
<varlistentry>
7601+
<term>Byte1('r')</term>
7602+
<listitem><para>
7603+
Identifies the message as the rollback of a two-phase transaction message.
7604+
</para></listitem>
7605+
</varlistentry>
7606+
7607+
<varlistentry>
7608+
<term>Int8</term>
7609+
<listitem><para>
7610+
Flags; currently unused (must be 0).
7611+
</para></listitem>
7612+
</varlistentry>
7613+
7614+
<varlistentry>
7615+
<term>Int64</term>
7616+
<listitem><para>
7617+
The end LSN of the prepared transaction.
7618+
</para></listitem>
7619+
</varlistentry>
7620+
7621+
<varlistentry>
7622+
<term>Int64</term>
7623+
<listitem><para>
7624+
The end LSN of the rollback prepared transaction.
7625+
</para></listitem>
7626+
</varlistentry>
7627+
7628+
<varlistentry>
7629+
<term>Int64</term>
7630+
<listitem><para>
7631+
Prepare timestamp of the transaction. The value is in number
7632+
of microseconds since PostgreSQL epoch (2000-01-01).
7633+
</para></listitem>
7634+
</varlistentry>
7635+
7636+
<varlistentry>
7637+
<term>Int64</term>
7638+
<listitem><para>
7639+
Rollback timestamp of the transaction. The value is in number
7640+
of microseconds since PostgreSQL epoch (2000-01-01).
7641+
</para></listitem>
7642+
</varlistentry>
7643+
7644+
<varlistentry>
7645+
<term>Int32</term>
7646+
<listitem><para>
7647+
Xid of the transaction.
7648+
</para></listitem>
7649+
</varlistentry>
7650+
7651+
<varlistentry>
7652+
<term>String</term>
7653+
<listitem><para>
7654+
The user defined GID of the two-phase transaction.
7655+
</para></listitem>
7656+
</varlistentry>
7657+
7658+
</variablelist>
7659+
7660+
</para>
7661+
</listitem>
7662+
</varlistentry>
7663+
7664+
</variablelist>
7665+
73937666
<para>
73947667

73957668
The following message parts are shared by the above messages.

‎doc/src/sgml/ref/alter_subscription.sgml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,11 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
6767
Commands <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command> and
6868
<command>ALTER SUBSCRIPTION ... {SET|ADD|DROP} PUBLICATION ...</command> with refresh
6969
option as true cannot be executed inside a transaction block.
70+
71+
These commands also cannot be executed when the subscription has
72+
<literal>two_phase</literal> commit enabled, unless <literal>copy_data = false</literal>.
73+
See column <literal>subtwophasestate</literal> of
74+
<xref linkend="catalog-pg-subscription"/> to know the actual two-phase state.
7075
</para>
7176
</refsect1>
7277

0 commit comments

Comments
 (0)