public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: Amit Kapila <[email protected]>
To: Peter Smith <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Cc: Hayato Kuroda (Fujitsu) <[email protected]>
Cc: Dilip Kumar <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: RE: Force streaming every change in logical decoding
Date: Wed, 21 Dec 2022 13:14:09 +0000
Message-ID: <OSZPR01MB6310AAE12BC281158880380DFDEB9@OSZPR01MB6310.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAA4eK1LbHVsheSYu6uLfVREiNXFWRkS+ps8rfXKUhDBpk+FizA@mail.gmail.com>
References: <OSZPR01MB63104E7449DBE41932DB19F1FD1B9@OSZPR01MB6310.jpnprd01.prod.outlook.com>
<CAFiTN-v9r17Ae5uFJbFPQqdQyFtJb5=nZXBu8h1HW6xA9msrtg@mail.gmail.com>
<OSZPR01MB63102546180646AEE1C66C34FDE09@OSZPR01MB6310.jpnprd01.prod.outlook.com>
<CAA4eK1+vE1_iTty_2Z73_GgA6d26Kh9Oh+kY5uLi6kdterMV1Q@mail.gmail.com>
<TYAPR01MB586604F7A7E27A6A1A695E32F5EA9@TYAPR01MB5866.jpnprd01.prod.outlook.com>
<CAA4eK1+b-Y=4a39Ux3mc7h8Y+ea+0qjYjMbKkGNwg9y=kf3fSQ@mail.gmail.com>
<CAD21AoAAjhpS71y3xhZ8sAnX9GK4X31-+h-9RroTmMx4fpocSw@mail.gmail.com>
<CAHut+Pvn-b0dC2_NO5Vu+8Nfh+MrTuuw_fi1nOo1oLdDPVNNWA@mail.gmail.com>
<CAA4eK1LbHVsheSYu6uLfVREiNXFWRkS+ps8rfXKUhDBpk+FizA@mail.gmail.com>
On Wed, Dec 21, 2022 4:54 PM Amit Kapila <[email protected]> wrote:
>
> On Wed, Dec 21, 2022 at 1:55 PM Peter Smith <[email protected]>
> wrote:
> >
> > On Wed, Dec 21, 2022 at 6:22 PM Masahiko Sawada
> <[email protected]> wrote:
> > >
> > > On Tue, Dec 20, 2022 at 7:49 PM Amit Kapila <[email protected]>
> wrote:
> > > >
> > > > On Tue, Dec 20, 2022 at 2:46 PM Hayato Kuroda (Fujitsu)
> > > > <[email protected]> wrote:
> > > > >
> > > > > Dear hackers,
> > > > >
> > > > > > We have discussed three different ways to provide GUC for these
> > > > > > features. (1) Have separate GUCs like force_server_stream_mode,
> > > > > > force_server_serialize_mode, force_client_serialize_mode (we can
> use
> > > > > > different names for these) for each of these; (2) Have two sets of
> > > > > > GUCs for server and client. We can have logical_decoding_mode with
> > > > > > values as 'stream' and 'serialize' for the server and then
> > > > > > logical_apply_serialize = true/false for the client. (3) Have one GUC
> > > > > > like logical_replication_mode with values as 'server_stream',
> > > > > > 'server_serialize', 'client_serialize'.
> > > > >
> > > > > I also agreed for adding new GUC parameters (and I have already done
> partially
> > > > > in parallel apply[1]), and basically options 2 made sense for me. But is
> it OK
> > > > > that we can choose "serialize" mode even if subscribers require
> streaming?
> > > > >
> > > > > Currently the reorder buffer transactions are serialized on publisher
> only when
> > > > > the there are no streamable transaction. So what happen if the
> > > > > logical_decoding_mode = "serialize" but streaming option streaming is
> on? If we
> > > > > break the first one and serialize changes on publisher anyway, it may
> be not
> > > > > suitable for testing the normal operation.
> > > > >
> > > >
> > > > I think the change will be streamed as soon as the next change is
> > > > processed even if we serialize based on this option. See
> > > > ReorderBufferProcessPartialChange. However, I see your point that
> when
> > > > the streaming option is given, the value 'serialize' for this GUC may
> > > > not make much sense.
> > > >
> > > > > Therefore, I came up with the variant of (2): logical_decoding_mode
> can be
> > > > > "normal" or "immediate".
> > > > >
> > > > > "normal" is a default value, which is same as current HEAD. Changes
> are streamed
> > > > > or serialized when the buffered size exceeds
> logical_decoding_work_mem.
> > > > >
> > > > > When users set to "immediate", the walsenders starts to stream or
> serialize all
> > > > > changes. The choice is depends on the subscription option.
> > > > >
> > > >
> > > > The other possibility to achieve what you are saying is that we allow
> > > > a minimum value of logical_decoding_work_mem as 0 which would
> mean
> > > > stream or serialize each change depending on whether the streaming
> > > > option is enabled. I think we normally don't allow a minimum value
> > > > below a certain threshold for other *_work_mem parameters (like
> > > > maintenance_work_mem, work_mem), so we have followed the same
> here.
> > > > And, I think it makes sense from the user's perspective because below
> > > > a certain threshold it will just add overhead by either writing small
> > > > changes to the disk or by sending those over the network. However, it
> > > > can be quite useful for testing/debugging. So, not sure, if we should
> > > > restrict setting logical_decoding_work_mem below a certain threshold.
> > > > What do you think?
> > >
> > > I agree with (2), having separate GUCs for publisher side and
> > > subscriber side. Also, on the publisher side, Amit's idea, controlling
> > > the logical decoding behavior by changing logical_decoding_work_mem,
> > > seems like a good idea.
> > >
> > > But I'm not sure it's a good idea if we lower the minimum value of
> > > logical_decoding_work_mem to 0. I agree it's helpful for testing and
> > > debugging but setting logical_decoding_work_mem = 0 doesn't benefit
> > > users at all, rather brings risks.
> > >
> > > I prefer the idea Kuroda-san previously proposed; setting
> > > logical_decoding_mode = 'immediate' means setting
> > > logical_decoding_work_mem = 0. We might not need to have it as an
> enum
> > > parameter since it has only two values, though.
> >
> > Did you mean one GUC (logical_decoding_mode) will cause a side-effect
> > implicit value change on another GUC value
> > (logical_decoding_work_mem)?
> >
>
> I don't think that is required. The only value that can be allowed for
> logical_decoding_mode will be "immediate", something like we do for
> recovery_target. The default will be "". The "immediate" value will
> mean that stream each change if the "streaming" option is enabled
> ("on" of "parallel") or if "streaming" is not enabled then that would
> mean serializing each change.
>
I agreed and updated the patch as Amit suggested.
Please see the attached patch.
Regards,
Shi yu
Attachments:
[application/octet-stream] v3-0001-Allow-streaming-or-serializing-each-change-in-log.patch (8.9K, ../OSZPR01MB6310AAE12BC281158880380DFDEB9@OSZPR01MB6310.jpnprd01.prod.outlook.com/2-v3-0001-Allow-streaming-or-serializing-each-change-in-log.patch)
download | inline diff:
From 155aa2c3bce8b2df8f8d7bc9751cd32d4bae2452 Mon Sep 17 00:00:00 2001
From: Shi Yu <[email protected]>
Date: Wed, 21 Dec 2022 18:46:57 +0800
Subject: [PATCH v3] Allow streaming or serializing each change in logical
decoding.
Add a new GUC 'logical_decoding_mode'. The default is the empty string, which
disables this feature. When set to "immediate", stream each change immediately
if "streaming" option is enabled, and if "streaming" option is disabled,
serialize each change.
---
contrib/test_decoding/expected/stream.out | 13 +++++++
contrib/test_decoding/sql/stream.sql | 8 ++++
doc/src/sgml/config.sgml | 21 +++++++++++
.../replication/logical/reorderbuffer.c | 37 +++++++++++++------
src/backend/utils/misc/guc_tables.c | 17 +++++++++
src/include/replication/reorderbuffer.h | 8 ++++
6 files changed, 92 insertions(+), 12 deletions(-)
diff --git a/contrib/test_decoding/expected/stream.out b/contrib/test_decoding/expected/stream.out
index 0f21dcb8e0..b7af8147a3 100644
--- a/contrib/test_decoding/expected/stream.out
+++ b/contrib/test_decoding/expected/stream.out
@@ -106,6 +106,19 @@ SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL, 'incl
committing streamed transaction
(17 rows)
+-- streaming test with logical_decoding_mode=immediate
+SET logical_decoding_mode=immediate;
+TRUNCATE table stream_test;
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1');
+ data
+------------------------------------------
+ opening a streamed block for transaction
+ streaming truncate for transaction
+ closing a streamed block for transaction
+ committing streamed transaction
+(4 rows)
+
+RESET logical_decoding_mode;
DROP TABLE stream_test;
SELECT pg_drop_replication_slot('regression_slot');
pg_drop_replication_slot
diff --git a/contrib/test_decoding/sql/stream.sql b/contrib/test_decoding/sql/stream.sql
index 4feec62972..b184c13e6b 100644
--- a/contrib/test_decoding/sql/stream.sql
+++ b/contrib/test_decoding/sql/stream.sql
@@ -44,5 +44,13 @@ toasted-123456789012345678901234567890123456789012345678901234567890123456789012
SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1');
+-- streaming test with logical_decoding_mode=immediate
+SET logical_decoding_mode=immediate;
+
+TRUNCATE table stream_test;
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1');
+
+RESET logical_decoding_mode;
+
DROP TABLE stream_test;
SELECT pg_drop_replication_slot('regression_slot');
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 9eedab652d..28d78473d7 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -11597,6 +11597,27 @@ 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>)
+ <indexterm>
+ <primary><varname>logical_decoding_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
+ empty string and <literal>immediate</literal>. When set to
+ <literal>immediate</literal>, stream each change if
+ <literal>streaming</literal> option is enabled, and if
+ <literal>streaming</literal> option is disabled, serialize each change.
+ The default is the empty string, which disables this feature (stream or
+ serialize until <varname>logical_decoding_work_mem</varname> is
+ exceeded).
+ </para>
+ </listitem>
+ </varlistentry>
+
</variablelist>
</sect1>
<sect1 id="runtime-config-short">
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index b567b8b59e..ae6fe97cf9 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -209,6 +209,9 @@ typedef struct ReorderBufferDiskChange
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_DEFAULT;
+
/* ---------------------------------------
* primary reorderbuffer support routines
* ---------------------------------------
@@ -3540,32 +3543,42 @@ ReorderBufferLargestStreamableTopTXN(ReorderBuffer *rb)
/*
* Check whether the logical_decoding_work_mem limit was reached, and if yes
* pick the largest (sub)transaction at-a-time to evict and spill its changes to
- * disk until we reach under the memory limit.
+ * 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.
*
* 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
- * evicting enough transactions to free certain fraction (e.g. 50%) of the memory
- * limit.
+ * limit, but we might also adapt a more elaborate eviction strategy - for
+ * example evicting enough transactions to free certain fraction (e.g. 50%) of
+ * the memory limit.
*/
static void
ReorderBufferCheckMemoryLimit(ReorderBuffer *rb)
{
ReorderBufferTXN *txn;
- /* bail out if we haven't exceeded the memory limit */
- if (rb->size < logical_decoding_work_mem * 1024L)
+ /*
+ * Bail out if logical_decoding_mode is disabled and we haven't exceeded the
+ * memory limit.
+ */
+ if (logical_decoding_mode == LOGICAL_DECODING_MODE_DEFAULT
+ && rb->size < logical_decoding_work_mem * 1024L)
return;
/*
- * 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 at least as
- * large as the most recent change (which caused us to go over the memory
- * limit). However, that is not true because a user can reduce the
+ * If logical_decoding_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 at
+ * least as large as the most recent change (which caused us to go over the
+ * memory limit). However, that is not true because a user can reduce the
* logical_decoding_work_mem to a smaller value before the most recent
* change.
*/
- while (rb->size >= logical_decoding_work_mem * 1024L)
+ while (rb->size >= logical_decoding_work_mem * 1024L ||
+ (logical_decoding_mode == LOGICAL_DECODING_MODE_IMMEDIATE
+ && rb->size > 0))
{
/*
* Pick the largest transaction (or subtransaction) and evict it from
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index 436afe1d21..58f46da500 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -395,6 +395,12 @@ static const struct config_enum_entry ssl_protocol_versions_info[] = {
{NULL, 0, false}
};
+static const struct config_enum_entry logical_decoding_mode_options[] = {
+ {"", LOGICAL_DECODING_MODE_DEFAULT, false},
+ {"immediate", LOGICAL_DECODING_MODE_IMMEDIATE, false},
+ {NULL, 0, false}
+};
+
StaticAssertDecl(lengthof(ssl_protocol_versions_info) == (PG_TLS1_3_VERSION + 2),
"array length mismatch");
@@ -4877,6 +4883,17 @@ struct config_enum ConfigureNamesEnum[] =
NULL, NULL, NULL
},
+ {
+ {"logical_decoding_mode", PGC_USERSET, DEVELOPER_OPTIONS,
+ gettext_noop("Allows sending or serializing each change in logical decoding."),
+ NULL,
+ GUC_NOT_IN_SAMPLE
+ },
+ &logical_decoding_mode,
+ LOGICAL_DECODING_MODE_DEFAULT, logical_decoding_mode_options,
+ NULL, NULL, NULL
+ },
+
/* End-of-list marker */
{
{NULL, 0, 0, NULL, NULL}, NULL, 0, NULL, NULL, NULL, NULL
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index c700b55b1c..046833f1a5 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -18,6 +18,14 @@
#include "utils/timestamp.h"
extern PGDLLIMPORT int logical_decoding_work_mem;
+extern PGDLLIMPORT int logical_decoding_mode;
+
+/* possible values for logical_decoding_mode */
+typedef enum
+{
+ LOGICAL_DECODING_MODE_DEFAULT,
+ LOGICAL_DECODING_MODE_IMMEDIATE
+} LogicalDecodingMode;
/* an individual tuple, stored in one chunk of memory */
typedef struct ReorderBufferTupleBuf
--
2.31.1
view thread (61+ 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]
Subject: RE: Force streaming every change in logical decoding
In-Reply-To: <OSZPR01MB6310AAE12BC281158880380DFDEB9@OSZPR01MB6310.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