public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: Peter Smith <[email protected]>
Cc: Amit Kapila <[email protected]>
Cc: shveta malik <[email protected]>
Cc: [email protected] <[email protected]>
Cc: [email protected] <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Cc: Dilip Kumar <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Subject: RE: Perform streaming logical transactions by background workers and parallel apply
Date: Wed, 25 Jan 2023 14:24:50 +0000
Message-ID: <OS0PR01MB57164C52100F6B68D2E3874194CE9@OS0PR01MB5716.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAHut+Pu8LCtzdd_PMbUoueZ9W8D7-Hd6nTH30kFEGzhfTgtFEg@mail.gmail.com>
References: <CAA4eK1LGmZsevrqJra0V4O8oBU_eKyzm2VMpSAYQaDgC6n4fkA@mail.gmail.com>
<CAFiTN-uhnJkrUsQByBHMK268T-GEx_D8DQ6b2T+aW6RiU75pbQ@mail.gmail.com>
<OS0PR01MB571621ED532C2D7C3E01625894FB9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAFiTN-t5+_zTf86EzgCwObs1ED-P_5Ab1KMw340d3oPY_+Cwpw@mail.gmail.com>
<OS3PR01MB5718AE486227CE3ACB844C5E94F89@OS3PR01MB5718.jpnprd01.prod.outlook.com>
<CAFiTN-sgQ-K9BnMWKLR7Hm36QNjS7ZOBF6hsXJmQz5ChC0Desw@mail.gmail.com>
<CAA4eK1+P9WOFBupaTZCzYtranSH=8TSUYV=kk5xQJ-VNKnvgmQ@mail.gmail.com>
<OS0PR01MB571685A645E509267AC5F30894F99@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB57168A5A1C31BC316CB63D3C94F99@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1KK7Xty7SaCReY_b5moPURowMGOsfvhOb6xQ1EAaJAh6A@mail.gmail.com>
<OS0PR01MB5716863B23DBAD186F64ADF194FF9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAHut+PvA10Bp9Jaw9OS2+puKHr7ry_xB3Tf2-bbv5gyxD5E_gw@mail.gmail.com>
<CAA4eK1KYUbnthSPyo4VjnhMygB0c1DZtp0XC-V2-GSETQ743ww@mail.gmail.com>
<CAJpy0uC0JamKT6pTtux3NDhcqSMJnDoQXZxx32pq=2yyxQ1c5Q@mail.gmail.com>
<CAA4eK1KueUT_vr1cg43MpvRcaVvW03i6zDECQNAwFBK5PhXOVA@mail.gmail.com>
<OS0PR01MB5716C663C85687E76672327094FD9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAHut+PuKYdA4047wATm8WpkkqQX2CPNCa++dSQKr6dD1O5nz_Q@mail.gmail.com>
<CAA4eK1LotEuPsteuJMNpixxTj6R4B8k93q-6ruRmDzCxKzMNpA@mail.gmail.com>
<CAA4eK1JdDrFxcw20YHifFrfX1o_e=9sXpNMfNC5vMYvYcebwNg@mail.gmail.com>
<CAD21AoAy2c=Mx=FTCs+EwUsf2kQL5MmU3N18X84k0EmCXntK4g@mail.gmail.com>
<CAA4eK1+EYODkcHTs1F6TmQdYeuT8WkzZGbnr8yAv5WfTFeV7-Q@mail.gmail.com>
<OS0PR01MB5716B68E724D9A321238483294C89@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAHut+Pvw0eBGu+pe9jRBdC+29MO5Fep4CTf92muhV5Ji+P9tcw@mail.gmail.com>
<OS0PR01MB57167F14EED6B95582155D9194C99@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB57168ADE4F156E1D65795ECF94C99@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAHut+Pu8LCtzdd_PMbUoueZ9W8D7-Hd6nTH30kFEGzhfTgtFEg@mail.gmail.com>
On Wednesday, January 25, 2023 7:30 AM Peter Smith <[email protected]> wrote:
>
> Here are my review comments for patch v87-0002.
Thanks for your comments.
> ======
> doc/src/sgml/config.sgml
>
> 1.
> <para>
> - Allows streaming or serializing changes immediately in
> logical decoding.
> The allowed values of
> <varname>logical_replication_mode</varname> are
> - <literal>buffered</literal> and <literal>immediate</literal>. When
> set
> - to <literal>immediate</literal>, stream each change if
> + <literal>buffered</literal> and <literal>immediate</literal>.
> The default
> + is <literal>buffered</literal>.
> + </para>
>
> I didn't think it was necessary to say “of logical_replication_mode”.
> IMO that much is already obvious because this is the first sentence of the
> description for logical_replication_mode.
>
Changed.
> ~~~
>
> 2.
> + <para>
> + On the publisher side, it allows streaming or serializing changes
> + immediately in logical decoding. When set to
> + <literal>immediate</literal>, stream each change if
> <literal>streaming</literal> option (see optional parameters set by
> <link linkend="sql-createsubscription"><command>CREATE
> SUBSCRIPTION</command></link>)
> is enabled, otherwise, serialize each change. When set to
> - <literal>buffered</literal>, which is the default, decoding will stream
> - or serialize changes when
> <varname>logical_decoding_work_mem</varname>
> - is reached.
> + <literal>buffered</literal>, decoding will stream or serialize changes
> + when <varname>logical_decoding_work_mem</varname> is
> reached.
> </para>
>
> 2a.
> "it allows" --> "logical_replication_mode allows"
>
> 2b.
> "decoding" --> "the decoding"
Changed.
> ~~~
>
> 3.
> + <para>
> + On the subscriber side, if <literal>streaming</literal> option is set
> + to <literal>parallel</literal>, this parameter also allows the leader
> + apply worker to send changes to the shared memory queue or to
> serialize
> + changes. When set to <literal>buffered</literal>, the leader sends
> + changes to parallel apply workers via shared memory queue. When
> set to
> + <literal>immediate</literal>, the leader serializes all changes to
> + files and notifies the parallel apply workers to read and apply them at
> + the end of the transaction.
> + </para>
>
> "this parameter also allows" --> "logical_replication_mode also allows"
Changed.
> ~~~
>
> 4.
> <para>
> This parameter is intended to be used to test logical decoding and
> replication of large transactions for which otherwise we need to
> generate the changes till
> <varname>logical_decoding_work_mem</varname>
> - is reached.
> + is reached. Moreover, this can also be used to test the transmission of
> + changes between the leader and parallel apply workers.
> </para>
>
> "Moreover, this can also" --> "It can also"
>
> I am wondering would this sentence be better put at the top of the GUC
> description. So then the first paragraph becomes like this:
>
>
> SUGGESTION (I've also added another sentence "The effect of...")
>
> The allowed values are buffered and immediate. The default is buffered. This
> parameter is intended to be used to test logical decoding and replication of large
> transactions for which otherwise we need to generate the changes till
> logical_decoding_work_mem is reached. It can also be used to test the
> transmission of changes between the leader and parallel apply workers. The
> effect of logical_replication_mode is different for the publisher and
> subscriber:
>
> On the publisher side...
>
> On the subscriber side...
I think your suggestion makes sense, so changed as suggested.
> ======
> .../replication/logical/applyparallelworker.c
>
> 5.
> + /*
> + * In immeidate mode, directly return false so that we can switch to
> + * PARTIAL_SERIALIZE mode and serialize remaining changes to files.
> + */
> + if (logical_replication_mode == LOGICAL_REP_MODE_IMMEDIATE) return
> + false;
>
> Typo "immediate"
>
> Also, I felt "directly" is not needed. "return false" and "directly return false" is the
> same.
>
> SUGGESTION
> Using ‘immediate’ mode returns false to cause a switch to PARTIAL_SERIALIZE
> mode so that the remaining changes will be serialized.
Changed.
> ======
> src/backend/utils/misc/guc_tables.c
>
> 6.
> {
> {"logical_replication_mode", PGC_USERSET, DEVELOPER_OPTIONS,
> - gettext_noop("Allows streaming or serializing each change in logical
> decoding."),
> - NULL,
> + gettext_noop("Controls the behavior of logical replication publisher
> and subscriber"),
> + gettext_noop("If set to immediate, on the publisher side, it "
> + "allows streaming or serializing each change in "
> + "logical decoding. On the subscriber side, in "
> + "parallel streaming mode, it allows the leader apply "
> + "worker to serialize changes to files and notifies "
> + "the parallel apply workers to read and apply them at "
> + "the end of the transaction."),
> GUC_NOT_IN_SAMPLE
> },
>
> 6a. short description
>
> User PoV behaviour should be the same. Instead, maybe say "controls the
> internal behavior" or something like that?
Changed to "internal behavior xxx"
> ~
>
> 6b. long description
>
> IMO the long description shouldn’t mention ‘immediate’ mode first as it does.
>
> BEFORE
> If set to immediate, on the publisher side, ...
>
> AFTER
> On the publisher side, ...
Changed.
Attach the new version patch set.
The 0001 patch is the same as the v88-0001 posted by Amit[1],
attach it here to make cfbot happy.
[1] https://www.postgresql.org/message-id/CAA4eK1JpWoaB63YULpQa1KDw_zBW-QoRMuNxuiP1KafPJzuVuw%40mail.gma...
Best Regards,
Hou zj
Attachments:
[application/octet-stream] v88-0001-Rename-GUC-logical_decoding_mode-to-logical_repl.patch (10.9K, ../OS0PR01MB57164C52100F6B68D2E3874194CE9@OS0PR01MB5716.jpnprd01.prod.outlook.com/2-v88-0001-Rename-GUC-logical_decoding_mode-to-logical_repl.patch)
download | inline diff:
From bc9691d5be9bf37fd546d8e6567df0869502dd97 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <[email protected]>
Date: Fri, 20 Jan 2023 14:39:38 +0800
Subject: [PATCH v88 1/2] Rename GUC logical_decoding_mode to
logical_replication_mode.
Rename the developer option 'logical_decoding_mode' to the more flexible
name 'logical_replication_mode' because doing so will make it easier to
extend this option in the future to help test other areas of logical
replication.
Currently, it is used on the publisher side to allow streaming or
serializing each change in logical decoding. In the upcoming patch, we are
planning to use it on the subscriber. On the subscriber, it will allow
serializing the changes to file and notifies the parallel apply workers to
read and apply them at the end of the transaction.
We have discussed having this parameter as a subscription option but
exposing a parameter that is primarily used for testing/debugging to users
didn't seem advisable and there is no other such parameter. The other
option we have discussed is to have a separate GUC for subscriber-side
testing but it appears that for the current testing existing parameter is
sufficient and avoids adding another GUC.
Author: Hou Zhijie
Reviewed-by: Pater Smith, Kuroda Hayato, Amit Kapila
Discussion: https://postgr.es/m/CAD21AoAy2c=Mx=FTCs+EwUsf2kQL5MmU3N18X84k0EmCXntK4g@mail.gmail.com
Discussion: https://postgr.es/m/CAA4eK1+wyN6zpaHUkCLorEWNx75MG0xhMwcFhvjqm2KURZEAGw@mail.gmail.com
---
doc/src/sgml/config.sgml | 8 ++++----
src/backend/replication/logical/reorderbuffer.c | 14 +++++++-------
src/backend/utils/misc/guc_tables.c | 16 ++++++++--------
src/include/replication/reorderbuffer.h | 11 ++++++-----
src/test/subscription/t/016_stream_subxact.pl | 2 +-
.../subscription/t/018_stream_subxact_abort.pl | 2 +-
.../t/019_stream_subxact_ddl_abort.pl | 2 +-
src/test/subscription/t/023_twophase_stream.pl | 2 +-
src/tools/pgindent/typedefs.list | 2 +-
9 files changed, 30 insertions(+), 29 deletions(-)
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index f985afc009..1cf53c74ea 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -11693,16 +11693,16 @@ LOG: CleanUpLock: deleting: lock(0xb7acd844) id(24688,24696,0,0,0,1)
</listitem>
</varlistentry>
- <varlistentry id="guc-logical-decoding-mode" xreflabel="logical_decoding_mode">
- <term><varname>logical_decoding_mode</varname> (<type>enum</type>)
+ <varlistentry id="guc-logical-replication-mode" xreflabel="logical_replication_mode">
+ <term><varname>logical_replication_mode</varname> (<type>enum</type>)
<indexterm>
- <primary><varname>logical_decoding_mode</varname> configuration parameter</primary>
+ <primary><varname>logical_replication_mode</varname> configuration parameter</primary>
</indexterm>
</term>
<listitem>
<para>
Allows streaming or serializing changes immediately in logical decoding.
- The allowed values of <varname>logical_decoding_mode</varname> are
+ The allowed values of <varname>logical_replication_mode</varname> are
<literal>buffered</literal> and <literal>immediate</literal>. When set
to <literal>immediate</literal>, stream each change if
<literal>streaming</literal> option (see optional parameters set by
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 54ee824e6c..efe057b4de 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -210,7 +210,7 @@ int logical_decoding_work_mem;
static const Size max_changes_in_memory = 4096; /* XXX for restore only */
/* GUC variable */
-int logical_decoding_mode = LOGICAL_DECODING_MODE_BUFFERED;
+int logical_replication_mode = LOGICAL_REP_MODE_BUFFERED;
/* ---------------------------------------
* primary reorderbuffer support routines
@@ -3552,8 +3552,8 @@ ReorderBufferLargestStreamableTopTXN(ReorderBuffer *rb)
* pick the largest (sub)transaction at-a-time to evict and spill its changes to
* disk or send to the output plugin until we reach under the memory limit.
*
- * If logical_decoding_mode is set to "immediate", stream or serialize the changes
- * immediately.
+ * If logical_replication_mode is set to "immediate", stream or serialize the
+ * changes immediately.
*
* XXX At this point we select the transactions until we reach under the memory
* limit, but we might also adapt a more elaborate eviction strategy - for example
@@ -3566,15 +3566,15 @@ ReorderBufferCheckMemoryLimit(ReorderBuffer *rb)
ReorderBufferTXN *txn;
/*
- * Bail out if logical_decoding_mode is buffered and we haven't exceeded
+ * Bail out if logical_replication_mode is buffered and we haven't exceeded
* the memory limit.
*/
- if (logical_decoding_mode == LOGICAL_DECODING_MODE_BUFFERED &&
+ if (logical_replication_mode == LOGICAL_REP_MODE_BUFFERED &&
rb->size < logical_decoding_work_mem * 1024L)
return;
/*
- * If logical_decoding_mode is immediate, loop until there's no change.
+ * If logical_replication_mode is immediate, loop until there's no change.
* Otherwise, loop until we reach under the memory limit. One might think
* that just by evicting the largest (sub)transaction we will come under
* the memory limit based on assumption that the selected transaction is
@@ -3584,7 +3584,7 @@ ReorderBufferCheckMemoryLimit(ReorderBuffer *rb)
* change.
*/
while (rb->size >= logical_decoding_work_mem * 1024L ||
- (logical_decoding_mode == LOGICAL_DECODING_MODE_IMMEDIATE &&
+ (logical_replication_mode == LOGICAL_REP_MODE_IMMEDIATE &&
rb->size > 0))
{
/*
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index 4ac808ed22..c5a95f5dcc 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -395,9 +395,9 @@ static const struct config_enum_entry ssl_protocol_versions_info[] = {
{NULL, 0, false}
};
-static const struct config_enum_entry logical_decoding_mode_options[] = {
- {"buffered", LOGICAL_DECODING_MODE_BUFFERED, false},
- {"immediate", LOGICAL_DECODING_MODE_IMMEDIATE, false},
+static const struct config_enum_entry logical_replication_mode_options[] = {
+ {"buffered", LOGICAL_REP_MODE_BUFFERED, false},
+ {"immediate", LOGICAL_REP_MODE_IMMEDIATE, false},
{NULL, 0, false}
};
@@ -4919,13 +4919,13 @@ struct config_enum ConfigureNamesEnum[] =
},
{
- {"logical_decoding_mode", PGC_USERSET, DEVELOPER_OPTIONS,
- gettext_noop("Allows streaming or serializing each change in logical decoding."),
- NULL,
+ {"logical_replication_mode", PGC_USERSET, DEVELOPER_OPTIONS,
+ gettext_noop("Controls when to replicate each change."),
+ gettext_noop("On the publisher, it allows streaming or serializing each change in logical decoding."),
GUC_NOT_IN_SAMPLE
},
- &logical_decoding_mode,
- LOGICAL_DECODING_MODE_BUFFERED, logical_decoding_mode_options,
+ &logical_replication_mode,
+ LOGICAL_REP_MODE_BUFFERED, logical_replication_mode_options,
NULL, NULL, NULL
},
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index f6c4dd75db..e5db041df1 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -17,15 +17,16 @@
#include "utils/snapshot.h"
#include "utils/timestamp.h"
+/* GUC variables */
extern PGDLLIMPORT int logical_decoding_work_mem;
-extern PGDLLIMPORT int logical_decoding_mode;
+extern PGDLLIMPORT int logical_replication_mode;
-/* possible values for logical_decoding_mode */
+/* possible values for logical_replication_mode */
typedef enum
{
- LOGICAL_DECODING_MODE_BUFFERED,
- LOGICAL_DECODING_MODE_IMMEDIATE
-} LogicalDecodingMode;
+ LOGICAL_REP_MODE_BUFFERED,
+ LOGICAL_REP_MODE_IMMEDIATE
+} LogicalRepMode;
/* an individual tuple, stored in one chunk of memory */
typedef struct ReorderBufferTupleBuf
diff --git a/src/test/subscription/t/016_stream_subxact.pl b/src/test/subscription/t/016_stream_subxact.pl
index 2f0148c3a8..d830f26e06 100644
--- a/src/test/subscription/t/016_stream_subxact.pl
+++ b/src/test/subscription/t/016_stream_subxact.pl
@@ -79,7 +79,7 @@ sub test_streaming
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
$node_publisher->append_conf('postgresql.conf',
- 'logical_decoding_mode = immediate');
+ 'logical_replication_mode = immediate');
$node_publisher->start;
# Create subscriber node
diff --git a/src/test/subscription/t/018_stream_subxact_abort.pl b/src/test/subscription/t/018_stream_subxact_abort.pl
index dce14b150a..814daf4d2f 100644
--- a/src/test/subscription/t/018_stream_subxact_abort.pl
+++ b/src/test/subscription/t/018_stream_subxact_abort.pl
@@ -130,7 +130,7 @@ sub test_streaming
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
$node_publisher->append_conf('postgresql.conf',
- 'logical_decoding_mode = immediate');
+ 'logical_replication_mode = immediate');
$node_publisher->start;
# Create subscriber node
diff --git a/src/test/subscription/t/019_stream_subxact_ddl_abort.pl b/src/test/subscription/t/019_stream_subxact_ddl_abort.pl
index b30223de51..d0e556c8b8 100644
--- a/src/test/subscription/t/019_stream_subxact_ddl_abort.pl
+++ b/src/test/subscription/t/019_stream_subxact_ddl_abort.pl
@@ -16,7 +16,7 @@ use Test::More;
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
$node_publisher->append_conf('postgresql.conf',
- 'logical_decoding_mode = immediate');
+ 'logical_replication_mode = immediate');
$node_publisher->start;
# Create subscriber node
diff --git a/src/test/subscription/t/023_twophase_stream.pl b/src/test/subscription/t/023_twophase_stream.pl
index 75ca83765e..497245a209 100644
--- a/src/test/subscription/t/023_twophase_stream.pl
+++ b/src/test/subscription/t/023_twophase_stream.pl
@@ -301,7 +301,7 @@ $node_publisher->init(allows_streaming => 'logical');
$node_publisher->append_conf(
'postgresql.conf', qq(
max_prepared_transactions = 10
-logical_decoding_mode = immediate
+logical_replication_mode = immediate
));
$node_publisher->start;
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 51484ca7e2..07fbb7ccf6 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1458,7 +1458,6 @@ LogicalDecodeStreamStopCB
LogicalDecodeStreamTruncateCB
LogicalDecodeTruncateCB
LogicalDecodingContext
-LogicalDecodingMode
LogicalErrorCallbackState
LogicalOutputPluginInit
LogicalOutputPluginWriterPrepareWrite
@@ -1468,6 +1467,7 @@ LogicalRepBeginData
LogicalRepCommitData
LogicalRepCommitPreparedTxnData
LogicalRepCtxStruct
+LogicalRepMode
LogicalRepMsgType
LogicalRepPartMapEntry
LogicalRepPreparedTxnData
--
2.28.0.windows.1
[application/octet-stream] v88-0002-Extend-the-logical_replication_mode-to-test-the-.patch (14.3K, ../OS0PR01MB57164C52100F6B68D2E3874194CE9@OS0PR01MB5716.jpnprd01.prod.outlook.com/3-v88-0002-Extend-the-logical_replication_mode-to-test-the-.patch)
download | inline diff:
From 047ff210b77a253658222f58e8b8c8a83419f1c6 Mon Sep 17 00:00:00 2001
From: Hou <[email protected]>
Date: Wed, 25 Jan 2023 22:00:31 +0800
Subject: [PATCH v88 2/2] Extend the logical_replication_mode to test the
stream parallel mode
Give additional functionality to the existing developer option
'logical_replication_mode' to help test parallel apply of large
transactions on the subscriber.
When set to 'buffered', the leader sends changes to parallel apply workers via
shared memory queue. When set to 'immediate', the leader serializes all changes
to files and notifies the parallel apply workers to read and apply them at the
end of the transaction.
This helps in adding tests to cover the serialization code path in parallel
streaming mode.
---
doc/src/sgml/config.sgml | 36 +++++++----
.../replication/logical/applyparallelworker.c | 16 +++--
src/backend/utils/misc/guc_tables.c | 9 ++-
src/test/subscription/t/015_stream.pl | 27 ++++++++
.../t/018_stream_subxact_abort.pl | 61 ++++++++++++++++++-
.../subscription/t/023_twophase_stream.pl | 45 +++++++++++++-
6 files changed, 174 insertions(+), 20 deletions(-)
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 1cf53c74ea..c00ec90b56 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -11701,22 +11701,36 @@ LOG: CleanUpLock: deleting: lock(0xb7acd844) id(24688,24696,0,0,0,1)
</term>
<listitem>
<para>
- Allows streaming or serializing changes immediately in logical decoding.
- The allowed values of <varname>logical_replication_mode</varname> are
- <literal>buffered</literal> and <literal>immediate</literal>. When set
- to <literal>immediate</literal>, stream each change if
+ The allowed values are <literal>buffered</literal> and
+ <literal>immediate</literal>. The default is <literal>buffered</literal>.
+ This parameter is intended to be used to test logical decoding and
+ replication of large transactions for which otherwise we need to generate
+ the changes till logical_decoding_work_mem is reached. It can also be
+ used to test the transmission of changes between the leader and parallel
+ apply workers. The effect of <varname>logical_replication_mode</varname>
+ is different for the publisher and subscriber:
+ </para>
+
+ <para>
+ On the publisher side, <varname>logical_replication_mode</varname> allows
+ allows streaming or serializing changes immediately in logical decoding.
+ When set to <literal>immediate</literal>, stream each change if
<literal>streaming</literal> option (see optional parameters set by
<link linkend="sql-createsubscription"><command>CREATE SUBSCRIPTION</command></link>)
is enabled, otherwise, serialize each change. When set to
- <literal>buffered</literal>, which is the default, decoding will stream
- or serialize changes when <varname>logical_decoding_work_mem</varname>
- is reached.
+ <literal>buffered</literal>, the decoding will stream or serialize
+ changes when <varname>logical_decoding_work_mem</varname> is reached.
</para>
+
<para>
- This parameter is intended to be used to test logical decoding and
- replication of large transactions for which otherwise we need to
- generate the changes till <varname>logical_decoding_work_mem</varname>
- is reached.
+ On the subscriber side, if <literal>streaming</literal> option is set to
+ <literal>parallel</literal>, <varname>logical_replication_mode</varname>
+ also allows the leader apply worker to send changes to the shared memory
+ queue or to serialize changes. When set to <literal>buffered</literal>,
+ the leader sends changes to parallel apply workers via shared memory
+ queue. When set to <literal>immediate</literal>, the leader serializes
+ all changes to files and notifies the parallel apply workers to read and
+ apply them at the end of the transaction.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 3579e704fe..f0b1ec2df3 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -1149,6 +1149,13 @@ pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
Assert(!IsTransactionState());
Assert(!winfo->serialize_changes);
+ /*
+ * Using 'immediate' mode returns false to cause a switch to
+ * PARTIAL_SERIALIZE mode so that the remaining changes will be serialized.
+ */
+ if (logical_replication_mode == LOGICAL_REP_MODE_IMMEDIATE)
+ return false;
+
/*
* This timeout is a bit arbitrary but testing revealed that it is sufficient
* to send the message unless the parallel apply worker is waiting on some
@@ -1187,12 +1194,7 @@ pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
startTime = GetCurrentTimestamp();
else if (TimestampDifferenceExceeds(startTime, GetCurrentTimestamp(),
SHM_SEND_TIMEOUT_MS))
- {
- ereport(LOG,
- (errmsg("logical replication apply worker will serialize the remaining changes of remote transaction %u to a file",
- winfo->shared->xid)));
return false;
- }
}
}
@@ -1206,6 +1208,10 @@ void
pa_switch_to_partial_serialize(ParallelApplyWorkerInfo *winfo,
bool stream_locked)
{
+ ereport(LOG,
+ (errmsg("logical replication apply worker will serialize the remaining changes of remote transaction %u to a file",
+ winfo->shared->xid)));
+
/*
* The parallel apply worker could be stuck for some reason (say waiting
* on some lock by other backend), so stop trying to send data directly to
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index c5a95f5dcc..2e352d50ba 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -4920,8 +4920,13 @@ struct config_enum ConfigureNamesEnum[] =
{
{"logical_replication_mode", PGC_USERSET, DEVELOPER_OPTIONS,
- gettext_noop("Controls when to replicate each change."),
- gettext_noop("On the publisher, it allows streaming or serializing each change in logical decoding."),
+ gettext_noop("Controls the internal behavior of logical replication publisher and subscriber"),
+ gettext_noop("On the publisher, it allows streaming or "
+ "serializing each change in logical decoding. On the "
+ "subscriber, in parallel streaming mode, it allows "
+ "the leader apply worker to serialize changes to "
+ "files and notifies the parallel apply workers to "
+ "read and apply them at the end of the transaction."),
GUC_NOT_IN_SAMPLE
},
&logical_replication_mode,
diff --git a/src/test/subscription/t/015_stream.pl b/src/test/subscription/t/015_stream.pl
index 91e8aa8c0a..c591cfea5a 100644
--- a/src/test/subscription/t/015_stream.pl
+++ b/src/test/subscription/t/015_stream.pl
@@ -312,6 +312,33 @@ $result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_2");
is($result, qq(10000), 'data replicated to subscriber after dropping index');
+# Test serializing changes to files and notify the parallel apply worker to
+# apply them at the end of transaction.
+$node_subscriber->append_conf('postgresql.conf',
+ 'logical_replication_mode = immediate');
+$node_subscriber->append_conf('postgresql.conf', "log_min_messages = warning");
+$node_subscriber->reload;
+
+# Run a query to make sure that the reload has taken effect.
+$node_subscriber->safe_psql('postgres', q{SELECT 1});
+
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO test_tab_2 SELECT i FROM generate_series(1, 5000) s(i)");
+
+# Ensure that the changes are serialized.
+$node_subscriber->wait_for_log(
+ qr/LOG: ( [A-Z0-9]+:)? logical replication apply worker will serialize the remaining changes of remote transaction \d+ to a file/,
+ $offset);
+
+$node_publisher->wait_for_catchup($appname);
+
+# Check that transaction is committed on subscriber
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_2");
+is($result, qq(15000), 'all changes are replayed from file');
+
$node_subscriber->stop;
$node_publisher->stop;
diff --git a/src/test/subscription/t/018_stream_subxact_abort.pl b/src/test/subscription/t/018_stream_subxact_abort.pl
index 814daf4d2f..455fde2a7c 100644
--- a/src/test/subscription/t/018_stream_subxact_abort.pl
+++ b/src/test/subscription/t/018_stream_subxact_abort.pl
@@ -143,15 +143,17 @@ $node_publisher->safe_psql('postgres',
"CREATE TABLE test_tab (a int primary key, b varchar)");
$node_publisher->safe_psql('postgres',
"INSERT INTO test_tab VALUES (1, 'foo'), (2, 'bar')");
+$node_publisher->safe_psql('postgres', "CREATE TABLE test_tab_2 (a int)");
# Setup structure on subscriber
$node_subscriber->safe_psql('postgres',
"CREATE TABLE test_tab (a int primary key, b text, c INT, d INT, e INT)");
+$node_subscriber->safe_psql('postgres', "CREATE TABLE test_tab_2 (a int)");
# Setup logical replication
my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
$node_publisher->safe_psql('postgres',
- "CREATE PUBLICATION tap_pub FOR TABLE test_tab");
+ "CREATE PUBLICATION tap_pub FOR TABLE test_tab, test_tab_2");
my $appname = 'tap_sub';
@@ -198,6 +200,63 @@ $node_subscriber->safe_psql('postgres', q{SELECT 1});
test_streaming($node_publisher, $node_subscriber, $appname, 1);
+# Test serializing changes to files and notify the parallel apply worker to
+# apply them at the end of transaction.
+$node_subscriber->append_conf('postgresql.conf',
+ 'logical_replication_mode = immediate');
+$node_subscriber->append_conf('postgresql.conf', "log_min_messages = warning");
+$node_subscriber->reload;
+
+# Run a query to make sure that the reload has taken effect.
+$node_subscriber->safe_psql('postgres', q{SELECT 1});
+
+my $offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab_2 values(1);
+ ROLLBACK;
+ });
+
+# Ensure that the changes are serialized.
+$node_subscriber->wait_for_log(
+ qr/LOG: ( [A-Z0-9]+:)? logical replication apply worker will serialize the remaining changes of remote transaction \d+ to a file/,
+ $offset);
+
+$node_publisher->wait_for_catchup($appname);
+
+# Check that transaction is aborted on subscriber
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_2");
+is($result, qq(0), 'check rollback was reflected on subscriber');
+
+# Serialize the ABORT sub-transaction.
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab_2 values(1);
+ SAVEPOINT sp;
+ INSERT INTO test_tab_2 values(1);
+ ROLLBACK TO sp;
+ COMMIT;
+ });
+
+# Ensure that the changes are serialized.
+$node_subscriber->wait_for_log(
+ qr/LOG: ( [A-Z0-9]+:)? logical replication apply worker will serialize the remaining changes of remote transaction \d+ to a file/,
+ $offset);
+
+$node_publisher->wait_for_catchup($appname);
+
+# Check that only sub-transaction is aborted on subscriber.
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_2");
+is($result, qq(1), 'check rollback to savepoint was reflected on subscriber');
+
$node_subscriber->stop;
$node_publisher->stop;
diff --git a/src/test/subscription/t/023_twophase_stream.pl b/src/test/subscription/t/023_twophase_stream.pl
index 497245a209..9f046e13b8 100644
--- a/src/test/subscription/t/023_twophase_stream.pl
+++ b/src/test/subscription/t/023_twophase_stream.pl
@@ -319,16 +319,18 @@ $node_publisher->safe_psql('postgres',
"CREATE TABLE test_tab (a int primary key, b varchar)");
$node_publisher->safe_psql('postgres',
"INSERT INTO test_tab VALUES (1, 'foo'), (2, 'bar')");
+$node_publisher->safe_psql('postgres', "CREATE TABLE test_tab_2 (a int)");
# Setup structure on subscriber (columns a and b are compatible with same table name on publisher)
$node_subscriber->safe_psql('postgres',
"CREATE TABLE test_tab (a int primary key, b text, c timestamptz DEFAULT now(), d bigint DEFAULT 999)"
);
+$node_subscriber->safe_psql('postgres', "CREATE TABLE test_tab_2 (a int)");
# Setup logical replication (streaming = on)
my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
$node_publisher->safe_psql('postgres',
- "CREATE PUBLICATION tap_pub FOR TABLE test_tab");
+ "CREATE PUBLICATION tap_pub FOR TABLE test_tab, test_tab_2");
my $appname = 'tap_sub';
@@ -384,6 +386,47 @@ $node_subscriber->safe_psql('postgres', q{SELECT 1});
test_streaming($node_publisher, $node_subscriber, $appname, 1);
+# Test serializing changes to files and notify the parallel apply worker to
+# apply them at the end of transaction.
+$node_subscriber->append_conf('postgresql.conf',
+ 'logical_replication_mode = immediate');
+$node_subscriber->append_conf('postgresql.conf', "log_min_messages = warning");
+$node_subscriber->reload;
+
+# Run a query to make sure that the reload has taken effect.
+$node_subscriber->safe_psql('postgres', q{SELECT 1});
+
+my $offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab_2 values(1);
+ PREPARE TRANSACTION 'xact';
+ });
+
+# Ensure that the changes are serialized.
+$node_subscriber->wait_for_log(
+ qr/LOG: ( [A-Z0-9]+:)? logical replication apply worker will serialize the remaining changes of remote transaction \d+ to a file/,
+ $offset);
+
+$node_publisher->wait_for_catchup($appname);
+
+# Check that transaction is in prepared state on subscriber
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+is($result, qq(1), 'transaction is prepared on subscriber');
+
+# Check that 2PC gets committed on subscriber
+$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'xact';");
+
+$node_publisher->wait_for_catchup($appname);
+
+# Check that transaction is committed on subscriber
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab_2");
+is($result, qq(1), 'transaction is committed on subscriber');
+
###############################
# check all the cleanup
###############################
--
2.28.0.windows.1
view thread (105+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: RE: Perform streaming logical transactions by background workers and parallel apply
In-Reply-To: <OS0PR01MB57164C52100F6B68D2E3874194CE9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox