public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: Amit Kapila <[email protected]>
Cc: Peter Smith <[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: Mon, 23 Jan 2023 08:05:06 +0000
Message-ID: <OS0PR01MB5716B68E724D9A321238483294C89@OS0PR01MB5716.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAA4eK1+EYODkcHTs1F6TmQdYeuT8WkzZGbnr8yAv5WfTFeV7-Q@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>
On Monday, January 23, 2023 11:17 AM Amit Kapila <[email protected]> wrote:
>
> On Fri, Jan 20, 2023 at 11:48 AM Masahiko Sawada <[email protected]>
> wrote:
> >
> > >
> > > Yet another way is to use the existing parameter logical_decode_mode
> > > [1]. If the value of logical_decoding_mode is 'immediate', then we
> > > can immediately switch to partial serialize mode. This will
> > > eliminate the dependency on timing. The one argument against using
> > > this is that it won't be as clear as a separate parameter like
> > > 'stream_serialize_threshold' proposed by the patch but OTOH we
> > > already have a few parameters that serve a different purpose when
> > > used on the subscriber. For example, 'max_replication_slots' is used
> > > to define the maximum number of replication slots on the publisher
> > > and the maximum number of origins on subscribers. Similarly,
> > > wal_retrieve_retry_interval' is used for different purposes on
> > > subscriber and standby nodes.
> >
> > Using the existing parameter makes sense to me. But if we use
> > logical_decoding_mode also on the subscriber, as Shveta Malik also
> > suggested, probably it's better to rename it so as not to confuse. For
> > example, logical_replication_mode or something.
> >
>
> +1. Among the options discussed, this sounds better.
OK, here is patch set which does the same.
The first patch set only renames the GUC name, and the second patch uses
the GUC to test the partial serialization.
Best Regards,
Hou zj
Attachments:
[application/octet-stream] v86-0002-Extend-the-logical_replication_mode-to-test-the-stre.patch (11.2K, ../OS0PR01MB5716B68E724D9A321238483294C89@OS0PR01MB5716.jpnprd01.prod.outlook.com/2-v86-0002-Extend-the-logical_replication_mode-to-test-the-stre.patch)
download | inline diff:
From 5fe8f54ad16c3b8da0bf926293d7c4f1ce3fec8d Mon Sep 17 00:00:00 2001
From: Hou Zhijie <[email protected]>
Date: Fri, 20 Jan 2023 16:59:37 +0800
Subject: [PATCH] Extend the logical_replication_mode to test the stream
parallel mode
Use the use the existing developer option logical_replication_mode to test the
parallel apply of large transaction on 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.
---
doc/src/sgml/config.sgml | 12 ++++
.../replication/logical/applyparallelworker.c | 12 ++--
src/test/subscription/t/015_stream.pl | 27 ++++++++
.../t/018_stream_subxact_abort.pl | 65 ++++++++++++++++++-
.../subscription/t/023_twophase_stream.pl | 45 ++++++++++++-
5 files changed, 154 insertions(+), 7 deletions(-)
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 275dcd55ba..87df627751 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -11708,6 +11708,18 @@ LOG: CleanUpLock: deleting: lock(0xb7acd844) id(24688,24696,0,0,0,1)
or serialize changes when <varname>logical_decoding_work_mem</varname>
is reached.
</para>
+
+ <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>
+
<para>
This parameter is intended to be used to test logical decoding and
replication of large transactions for which otherwise we need to
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 3579e704fe..39e8f2d36d 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -1149,6 +1149,9 @@ pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
Assert(!IsTransactionState());
Assert(!winfo->serialize_changes);
+ 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 +1190,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 +1204,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/test/subscription/t/015_stream.pl b/src/test/subscription/t/015_stream.pl
index 91e8aa8c0a..93086052f6 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 messages 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..1fa0ae0b05 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,67 @@ $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 messages 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), 'all changes are replayed from file');
+
+# Clean up test data from the environment.
+$node_publisher->safe_psql('postgres', "TRUNCATE TABLE test_tab_2");
+$node_publisher->wait_for_catchup($appname);
+
+# 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 messages 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), 'all changes are replayed from file');
+
$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..e1bf63b97c 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 messages 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), 'all changes are replayed from file');
+
###############################
# check all the cleanup
###############################
--
2.28.0.windows.1
[application/octet-stream] v86-0001-Rename-logical_decoding_mode-to-logical_replication_.patch (9.5K, ../OS0PR01MB5716B68E724D9A321238483294C89@OS0PR01MB5716.jpnprd01.prod.outlook.com/3-v86-0001-Rename-logical_decoding_mode-to-logical_replication_.patch)
download | inline diff:
From b53dd94b2fb23f556a4bea8a2be4e1890e1db930 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <[email protected]>
Date: Fri, 20 Jan 2023 14:39:38 +0800
Subject: [PATCH] Rename logical_decoding_mode to logical_replication_mode
Since we may extend the developer option logical_decoding_mode to to test the
parallel apply of large transaction on subscriber, rename this option to
logical_replication_mode to make it easier to understand.
---
doc/src/sgml/config.sgml | 8 ++++----
src/backend/replication/logical/reorderbuffer.c | 12 ++++++------
src/backend/utils/misc/guc_tables.c | 10 +++++-----
src/include/replication/reorderbuffer.h | 10 +++++-----
src/test/subscription/t/016_stream_subxact.pl | 2 +-
src/test/subscription/t/018_stream_subxact_abort.pl | 2 +-
src/test/subscription/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, 25 insertions(+), 25 deletions(-)
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 77574e2..638c99d 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -11655,16 +11655,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 54ee824..42a137d 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,7 +3552,7 @@ 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
+ * 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
@@ -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 5025e80..705920c 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -396,8 +396,8 @@ static const struct config_enum_entry ssl_protocol_versions_info[] = {
};
static const struct config_enum_entry logical_decoding_mode_options[] = {
- {"buffered", LOGICAL_DECODING_MODE_BUFFERED, false},
- {"immediate", LOGICAL_DECODING_MODE_IMMEDIATE, false},
+ {"buffered", LOGICAL_REP_MODE_BUFFERED, false},
+ {"immediate", LOGICAL_REP_MODE_IMMEDIATE, false},
{NULL, 0, false}
};
@@ -4908,13 +4908,13 @@ struct config_enum ConfigureNamesEnum[] =
},
{
- {"logical_decoding_mode", PGC_USERSET, DEVELOPER_OPTIONS,
+ {"logical_replication_mode", PGC_USERSET, DEVELOPER_OPTIONS,
gettext_noop("Allows streaming or serializing each change in logical decoding."),
NULL,
GUC_NOT_IN_SAMPLE
},
- &logical_decoding_mode,
- LOGICAL_DECODING_MODE_BUFFERED, logical_decoding_mode_options,
+ &logical_replication_mode,
+ LOGICAL_REP_MODE_BUFFERED, logical_decoding_mode_options,
NULL, NULL, NULL
},
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index f6c4dd7..5a26ec9 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -18,14 +18,14 @@
#include "utils/timestamp.h"
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 2f0148c..d830f26 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 dce14b1..814daf4 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 b30223d..d0e556c 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 75ca837..497245a 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 23bafec..d0d61d8 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1456,7 +1456,6 @@ LogicalDecodeStreamStopCB
LogicalDecodeStreamTruncateCB
LogicalDecodeTruncateCB
LogicalDecodingContext
-LogicalDecodingMode
LogicalErrorCallbackState
LogicalOutputPluginInit
LogicalOutputPluginWriterPrepareWrite
@@ -1466,6 +1465,7 @@ LogicalRepBeginData
LogicalRepCommitData
LogicalRepCommitPreparedTxnData
LogicalRepCtxStruct
+LogicalRepMode
LogicalRepMsgType
LogicalRepPartMapEntry
LogicalRepPreparedTxnData
--
2.7.2.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: <OS0PR01MB5716B68E724D9A321238483294C89@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