public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: Amit Kapila <[email protected]>
To: Masahiko Sawada <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: [email protected] <[email protected]>
Cc: [email protected] <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: RE: Perform streaming logical transactions by background workers and parallel apply
Date: Tue, 14 Jun 2022 03:37:04 +0000
Message-ID: <OS3PR01MB62756BD9482EB6BB1CA4CD4D9EAA9@OS3PR01MB6275.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <OS3PR01MB6275208A2F8ED832710F65E09EA49@OS3PR01MB6275.jpnprd01.prod.outlook.com>
References: <CAA4eK1+wyN6zpaHUkCLorEWNx75MG0xhMwcFhvjqm2KURZEAGw@mail.gmail.com>
<OS0PR01MB57160DFDEF5B1C4668346B7D94E99@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB57166BC3CAA873364A2CC07A94EF9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAA4eK1Kc0VAhJAQe4_5YGrnTpfKYA9=F6sxqEKPrrax5Hy0+jw@mail.gmail.com>
<OS0PR01MB571616192F70A18AF88F3B2894F59@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB5716A6C18E69FB1EA9A27EED94F59@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<CAHut+PuqYP5eD5wcSCtk=a6KuMjat2UCzqyGoE7sieCaBsVskQ@mail.gmail.com>
<OS0PR01MB57160D1B0D29DC66ED035F5694F89@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB5716E8D536552467EFB512EF94FC9@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OSZPR01MB6310F0FABB05F8E5BB31A5B0FDFC9@OSZPR01MB6310.jpnprd01.prod.outlook.com>
<CAHut+Pv3FV+WibRBuGCyXd_ri+O4L3iY+UnHO2SHRbmC+xR6dw@mail.gmail.com>
<OSZPR01MB63106EADF50E0E710D36CE5DFDCA9@OSZPR01MB6310.jpnprd01.prod.outlook.com>
<CAHut+PuAxW57fowiMrn=3=53sagmehiTSW0o1Q52MpR3phUmyw@mail.gmail.com>
<OS3PR01MB6275797F66EF0A47EEB2D8FC9EDD9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAA4eK1Jt08SYbRt_-rbSWNg=X9-m8+RdP5PosfnQgyF-z8bkxQ@mail.gmail.com>
<CAA4eK1+Z6ahpTQK2KzkvQ1kN-urVS9-N_RDM11MS+btqaB8Bpw@mail.gmail.com>
<CAD21AoC1ME-xQjdcTeMpVMADKJDjfGVZATdGNuR+Qcimov5TPA@mail.gmail.com>
<CAA4eK1L=KsaU4fCpM_xbwK08E+NpYMviH23+uD0Be_2pKNtw_g@mail.gmail.com>
<OS3PR01MB62758A881FF3240171B7B21B9EDE9@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<OS3PR01MB6275208A2F8ED832710F65E09EA49@OS3PR01MB6275.jpnprd01.prod.outlook.com>
On Wed, Jun 8, 2022 3:13 PM I wrote:
> Attach the new patches.(only changed 0001 and 0003)
I tried to improve the patches by following points:
1. Initialize variable include_abort_lsn to false. It reports a warning in
cfbot. (see patch v10-0001)
BTW, I merged the patch that added the new GUC (see v9-0004) into patch 0001.
2. Because of the improvement #2 in [1], the foreign key could not be detected
when checking trigger function. So added additional checks for the foreign key.
(see patch 0004)
3. Adding a check for the partition table when trying to apply changes in the
apply background worker. (see patch 0004)
In additional, the partition cache map on subscriber have several bugs (see
thread [2]). Because patch 0004 is developed based on the patches in [2], so I
merged the patches(v4-0001~v4-0003) in [2] into a temporary patch 0003 here.
After the patches in [2] is committed, I will delete patch 0003 and rebase
patch 0004.
4. Improve constraint checking in a separate patch as suggested by Amit-san in
[3] #6.(see patch 0005)
I added a new field "bool subretry" in catalog pg_subscription. I use this
field to indicate whether the transaction that we are going to process has
failed before.
If apply worker/bgworker was exit with an error, this field will be set to
true; If we successfully apply a transaction, this field will be set to false.
If we retry to apply a streaming transaction, whether the user sets the
streaming option to "on" or "apply", we will apply the transaction in the apply
worker.
Attach the new patches.
Only changed patches 0001, 0004 and added new separate patch 0005.
[1] - https://www.postgresql.org/message-id/OS3PR01MB6275208A2F8ED832710F65E09EA49%40OS3PR01MB6275.jpnprd0...
[2] - https://www.postgresql.org/message-id/flat/OSZPR01MB6310F46CD425A967E4AEF736FDA49%40OSZPR01MB6310.jp...
[3] - https://www.postgresql.org/message-id/CAA4eK1Jt08SYbRt_-rbSWNg%3DX9-m8%2BRdP5PosfnQgyF-z8bkxQ%40mail...
Regards,
Wang wei
Attachments:
[application/octet-stream] v10-0001-Perform-streaming-logical-transactions-by-backgr.patch (98.2K, ../OS3PR01MB62756BD9482EB6BB1CA4CD4D9EAA9@OS3PR01MB6275.jpnprd01.prod.outlook.com/2-v10-0001-Perform-streaming-logical-transactions-by-backgr.patch)
download | inline diff:
From 9f43f193f4c264693b1bbea5fa2188888ff0a1be Mon Sep 17 00:00:00 2001
From: "houzj.fnst" <[email protected]>
Date: Wed, 20 Apr 2022 16:45:07 +0800
Subject: [PATCH v10 1/5] Perform streaming logical transactions by background
workers
Currently, for large transactions, the publisher sends the data in multiple
streams (changes divided into chunks depending upon logical_decoding_work_mem),
and then on the subscriber-side, the apply worker writes the changes into
temporary files and once it receives the commit, it reads from the file and
applies the entire transaction. To improve the performance of such
transactions, we can instead allow them to be applied via background workers.
In this approach, we assign a new apply background worker (if available) as
soon as the xact's first stream is received and the main apply worker will send
changes to this new worker via shared memory. The apply background worker will
directly apply the change instead of writing it to temporary files. We keep
this worker assigned till the transaction commit is received and also wait for
the worker to finish at commit. This preserves commit ordering and avoids
writing to and reading from file in most cases. We still need to spill if there
is no worker available. We also need to allow stream_stop to complete by the
apply background worker to avoid deadlocks because T-1's current stream of
changes can update rows in conflicting order with T-2's next stream of changes.
This patch also extends the SUBSCRIPTION 'streaming' option so that the user
can control whether to apply the streaming transaction in an apply background
worker or spill the change to disk. User can set the streaming option to
'on/off', 'apply'. For now, 'apply' means the streaming will be applied via a
apply background worker if available. 'on' means the streaming transaction will
be spilled to disk.
---
doc/src/sgml/catalogs.sgml | 10 +-
doc/src/sgml/config.sgml | 25 +
doc/src/sgml/logical-replication.sgml | 7 +
doc/src/sgml/protocol.sgml | 19 +
doc/src/sgml/ref/create_subscription.sgml | 24 +-
src/backend/access/transam/xact.c | 13 +
src/backend/commands/subscriptioncmds.c | 66 +-
src/backend/postmaster/bgworker.c | 3 +
src/backend/replication/logical/Makefile | 3 +-
.../replication/logical/applybgwroker.c | 765 ++++++++++++++++++
src/backend/replication/logical/decode.c | 10 +-
src/backend/replication/logical/launcher.c | 112 ++-
src/backend/replication/logical/origin.c | 26 +-
src/backend/replication/logical/proto.c | 35 +-
.../replication/logical/reorderbuffer.c | 10 +-
src/backend/replication/logical/tablesync.c | 10 +-
src/backend/replication/logical/worker.c | 645 +++++++++++----
src/backend/replication/pgoutput/pgoutput.c | 2 +-
src/backend/utils/activity/wait_event.c | 3 +
src/backend/utils/misc/guc.c | 12 +
src/backend/utils/misc/postgresql.conf.sample | 1 +
src/bin/pg_dump/pg_dump.c | 6 +-
src/include/catalog/pg_subscription.h | 17 +-
src/include/replication/logicallauncher.h | 1 +
src/include/replication/logicalproto.h | 19 +-
src/include/replication/logicalworker.h | 1 +
src/include/replication/origin.h | 2 +-
src/include/replication/reorderbuffer.h | 7 +-
src/include/replication/worker_internal.h | 104 ++-
src/include/utils/wait_event.h | 1 +
src/test/regress/expected/subscription.out | 2 +-
src/tools/pgindent/typedefs.list | 5 +
32 files changed, 1751 insertions(+), 215 deletions(-)
create mode 100644 src/backend/replication/logical/applybgwroker.c
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index c00c93dd7b..5cb39b18bd 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -7873,11 +7873,15 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>substream</structfield> <type>bool</type>
+ <structfield>substream</structfield> <type>char</type>
</para>
<para>
- If true, the subscription will allow streaming of in-progress
- transactions
+ Controls how to handle the streaming of in-progress transactions:
+ <literal>f</literal> = disallow streaming of in-progress transactions,
+ <literal>t</literal> = spill the changes of in-progress transactions to
+ disk and apply at once after the transaction is committed on the
+ publisher,
+ <literal>a</literal> = apply changes directly using a background worker
</para></entry>
</row>
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 5b7ce6531d..ce7f6827f2 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -4970,6 +4970,31 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
</listitem>
</varlistentry>
+ <varlistentry id="guc-max-apply-bgworkers-per-subscription" xreflabel="max_apply_bgworkers_per_subscription">
+ <term><varname>max_apply_bgworkers_per_subscription</varname> (<type>integer</type>)
+ <indexterm>
+ <primary><varname>max_apply_bgworkers_per_subscription</varname> configuration parameter</primary>
+ </indexterm>
+ </term>
+ <listitem>
+ <para>
+ Maximum number of apply background workers per subscription. This
+ parameter controls the amount of parallelism of the streaming of
+ in-progress transactions if we set subscription option
+ <literal>streaming</literal> to <literal>apply</literal>.
+ </para>
+ <para>
+ The apply background workers are taken from the pool defined by
+ <varname>max_logical_replication_workers</varname>.
+ </para>
+ <para>
+ The default value is 3. This parameter can only be set in the
+ <filename>postgresql.conf</filename> file or on the server command
+ line.
+ </para>
+ </listitem>
+ </varlistentry>
+
</variablelist>
</sect2>
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 145ea71d61..cca6c97d79 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -948,6 +948,13 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
might not violate any constraint. This can easily make the subscriber
inconsistent.
</para>
+
+ <para>
+ Setting streaming mode to <literal>apply</literal> could export invalid LSN
+ as finish LSN of failed transaction. Changing the streaming mode and making
+ the same conflict writes the finish LSN of the failed transaction in the
+ server log if required.
+ </para>
</sect1>
<sect1 id="logical-replication-restrictions">
diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index bb30340892..34be1b2698 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -6809,6 +6809,25 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
</listitem>
</varlistentry>
+ <varlistentry>
+ <term>Int64 (XLogRecPtr)</term>
+ <listitem>
+ <para>
+ The LSN of the abort.
+ </para>
+ </listitem>
+ </varlistentry>
+
+ <varlistentry>
+ <term>Int64 (TimestampTz)</term>
+ <listitem>
+ <para>
+ Abort timestamp of the transaction. The value is in number
+ of microseconds since PostgreSQL epoch (2000-01-01).
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry>
<term>Int32 (TransactionId)</term>
<listitem>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 35b39c28da..d4caae0222 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -217,13 +217,29 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</varlistentry>
<varlistentry>
- <term><literal>streaming</literal> (<type>boolean</type>)</term>
+ <term><literal>streaming</literal> (<type>enum</type>)</term>
<listitem>
<para>
Specifies whether to enable streaming of in-progress transactions
- for this subscription. By default, all transactions
- are fully decoded on the publisher and only then sent to the
- subscriber as a whole.
+ for this subscription. The default value is <literal>off</literal>,
+ all transactions are fully decoded on the publisher and only then
+ sent to the subscriber as a whole.
+ </para>
+
+ <para>
+ If set to <literal>on</literal>, the changes of transaction are
+ written to temporary files and then applied at once after the
+ transaction is committed on the publisher.
+ </para>
+
+ <para>
+ If set to <literal>apply</literal> incoming
+ changes are directly applied via one of the background workers, if
+ available. If no background worker is free to handle streaming
+ transaction then the changes are written to a file and applied after
+ the transaction is committed. Note that if an error happens when
+ applying changes in a background worker, it might not report the
+ finish LSN of the remote transaction in the server log.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 47d80b0d25..678540ba25 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -1711,6 +1711,7 @@ RecordTransactionAbort(bool isSubXact)
int nchildren;
TransactionId *children;
TimestampTz xact_time;
+ bool replorigin;
/*
* If we haven't been assigned an XID, nobody will care whether we aborted
@@ -1741,6 +1742,13 @@ RecordTransactionAbort(bool isSubXact)
elog(PANIC, "cannot abort transaction %u, it was already committed",
xid);
+ /*
+ * Are we using the replication origins feature? Or, in other words,
+ * are we replaying remote actions?
+ */
+ replorigin = (replorigin_session_origin != InvalidRepOriginId &&
+ replorigin_session_origin != DoNotReplicateId);
+
/* Fetch the data we need for the abort record */
nrels = smgrGetPendingDeletes(false, &rels);
nchildren = xactGetCommittedChildren(&children);
@@ -1765,6 +1773,11 @@ RecordTransactionAbort(bool isSubXact)
MyXactFlags, InvalidTransactionId,
NULL);
+ if (replorigin)
+ /* Move LSNs forward for this replication origin */
+ replorigin_session_advance(replorigin_session_origin_lsn,
+ XactLastRecEnd);
+
/*
* Report the latest async abort LSN, so that the WAL writer knows to
* flush this abort. There's nothing to be gained by delaying this, since
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 83e6eae855..4080bba987 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -83,7 +83,7 @@ typedef struct SubOpts
bool copy_data;
bool refresh;
bool binary;
- bool streaming;
+ char streaming;
bool twophase;
bool disableonerr;
XLogRecPtr lsn;
@@ -95,6 +95,62 @@ static List *merge_publications(List *oldpublist, List *newpublist, bool addpub,
static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err);
+/*
+ * Extract the streaming mode value from a DefElem. This is like
+ * defGetBoolean() but also accepts the special value of "apply".
+ */
+static char
+defGetStreamingMode(DefElem *def)
+{
+ /*
+ * If no parameter given, assume "true" is meant.
+ */
+ if (def->arg == NULL)
+ return SUBSTREAM_ON;
+
+ /*
+ * Allow 0, 1, "true", "false", "on", "off" or "apply".
+ */
+ switch (nodeTag(def->arg))
+ {
+ case T_Integer:
+ switch (intVal(def->arg))
+ {
+ case 0:
+ return SUBSTREAM_OFF;
+ case 1:
+ return SUBSTREAM_ON;
+ default:
+ /* otherwise, error out below */
+ break;
+ }
+ break;
+ default:
+ {
+ char *sval = defGetString(def);
+
+ /*
+ * The set of strings accepted here should match up with the
+ * grammar's opt_boolean_or_string production.
+ */
+ if (pg_strcasecmp(sval, "false") == 0 ||
+ pg_strcasecmp(sval, "off") == 0)
+ return SUBSTREAM_OFF;
+ if (pg_strcasecmp(sval, "true") == 0 ||
+ pg_strcasecmp(sval, "on") == 0)
+ return SUBSTREAM_ON;
+ if (pg_strcasecmp(sval, "apply") == 0)
+ return SUBSTREAM_APPLY;
+ }
+ break;
+ }
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("%s requires a Boolean value or \"apply\"",
+ def->defname)));
+ return SUBSTREAM_OFF; /* keep compiler quiet */
+}
+
/*
* Common option parsing function for CREATE and ALTER SUBSCRIPTION commands.
*
@@ -132,7 +188,7 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
if (IsSet(supported_opts, SUBOPT_BINARY))
opts->binary = false;
if (IsSet(supported_opts, SUBOPT_STREAMING))
- opts->streaming = false;
+ opts->streaming = SUBSTREAM_OFF;
if (IsSet(supported_opts, SUBOPT_TWOPHASE_COMMIT))
opts->twophase = false;
if (IsSet(supported_opts, SUBOPT_DISABLE_ON_ERR))
@@ -233,7 +289,7 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
errorConflictingDefElem(defel, pstate);
opts->specified_opts |= SUBOPT_STREAMING;
- opts->streaming = defGetBoolean(defel);
+ opts->streaming = defGetStreamingMode(defel);
}
else if (strcmp(defel->defname, "two_phase") == 0)
{
@@ -601,7 +657,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
values[Anum_pg_subscription_subowner - 1] = ObjectIdGetDatum(owner);
values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(opts.enabled);
values[Anum_pg_subscription_subbinary - 1] = BoolGetDatum(opts.binary);
- values[Anum_pg_subscription_substream - 1] = BoolGetDatum(opts.streaming);
+ values[Anum_pg_subscription_substream - 1] = CharGetDatum(opts.streaming);
values[Anum_pg_subscription_subtwophasestate - 1] =
CharGetDatum(opts.twophase ?
LOGICALREP_TWOPHASE_STATE_PENDING :
@@ -1060,7 +1116,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
if (IsSet(opts.specified_opts, SUBOPT_STREAMING))
{
values[Anum_pg_subscription_substream - 1] =
- BoolGetDatum(opts.streaming);
+ CharGetDatum(opts.streaming);
replaces[Anum_pg_subscription_substream - 1] = true;
}
diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c
index 40601aefd9..40ccb8993c 100644
--- a/src/backend/postmaster/bgworker.c
+++ b/src/backend/postmaster/bgworker.c
@@ -128,6 +128,9 @@ static const struct
},
{
"ApplyWorkerMain", ApplyWorkerMain
+ },
+ {
+ "ApplyBgworkerMain", ApplyBgworkerMain
}
};
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index c4e2fdeb71..a24a41969e 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -26,6 +26,7 @@ OBJS = \
reorderbuffer.o \
snapbuild.o \
tablesync.o \
- worker.o
+ worker.o \
+ applybgwroker.o
include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/replication/logical/applybgwroker.c b/src/backend/replication/logical/applybgwroker.c
new file mode 100644
index 0000000000..1f52b155d7
--- /dev/null
+++ b/src/backend/replication/logical/applybgwroker.c
@@ -0,0 +1,765 @@
+/*-------------------------------------------------------------------------
+ * applybgwroker.c
+ * Support routines for applying xact by apply background worker
+ *
+ * Copyright (c) 2016-2022, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/backend/replication/logical/applybgwroker.c
+ *
+ * This file contains routines that are intended to support setting up, using,
+ * and tearing down a ApplyBgworkerState.
+ * Refer to the comments in file header of logical/worker.c to see more
+ * informations about apply background worker.
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "libpq/pqformat.h"
+#include "mb/pg_wchar.h"
+#include "pgstat.h"
+#include "postmaster/interrupt.h"
+#include "replication/logicallauncher.h"
+#include "replication/logicalworker.h"
+#include "replication/origin.h"
+#include "replication/walreceiver.h"
+#include "replication/worker_internal.h"
+#include "storage/ipc.h"
+#include "storage/procarray.h"
+#include "tcop/tcopprot.h"
+#include "utils/inval.h"
+#include "utils/memutils.h"
+#include "utils/resowner.h"
+#include "utils/syscache.h"
+
+#define PG_LOGICAL_APPLY_SHM_MAGIC 0x79fb2447 // TODO Consider change
+
+/*
+ * DSM keys for apply background worker. Unlike other parallel execution code,
+ * since we don't need to worry about DSM keys conflicting with plan_node_id we
+ * can use small integers.
+ */
+#define APPLY_BGWORKER_KEY_SHARED 1
+#define APPLY_BGWORKER_KEY_MQ 2
+
+/*
+ * entry for a hash table we use to map from xid to our apply background worker
+ * state.
+ */
+typedef struct ApplyBgworkerEntry
+{
+ TransactionId xid;
+ ApplyBgworkerState *wstate;
+} ApplyBgworkerEntry;
+
+/* Apply background workers hash table (initialized on first use) */
+static HTAB *ApplyWorkersHash = NULL;
+static List *ApplyWorkersFreeList = NIL;
+static List *ApplyWorkersList = NIL;
+
+/*
+ * Fields to record the share informations between main apply worker and apply
+ * background worker.
+ */
+volatile ApplyBgworkerShared *MyParallelState = NULL;
+
+List *subxactlist = NIL;
+
+/* apply background worker setup */
+static ApplyBgworkerState *apply_bgworker_setup(void);
+static void apply_bgworker_setup_dsm(ApplyBgworkerState *wstate);
+
+/*
+ * Look up worker inside ApplyWorkersHash for requested xid.
+ *
+ * If start flag is true, try to start a new worker if not found in hash table.
+ */
+ApplyBgworkerState *
+apply_bgworker_find_or_start(TransactionId xid, bool start)
+{
+ bool found;
+ ApplyBgworkerState *wstate;
+ ApplyBgworkerEntry *entry = NULL;
+
+ if (!TransactionIdIsValid(xid))
+ return NULL;
+
+ /*
+ * We don't start new background worker if we are not in streaming apply
+ * mode.
+ */
+ if (MySubscription->stream != SUBSTREAM_APPLY)
+ return NULL;
+
+ /*
+ * We don't start new background worker if user has set skiplsn as it's
+ * possible that user want to skip the streaming transaction. For
+ * streaming transaction, we need to spill the transaction to disk so that
+ * we can get the last LSN of the transaction to judge whether to skip
+ * before starting to apply the change.
+ */
+ if (start && !XLogRecPtrIsInvalid(MySubscription->skiplsn))
+ return NULL;
+
+ /*
+ * For streaming transactions that are being applied in apply background
+ * worker, we cannot decide whether to apply the change for a relation
+ * that is not in the READY state (see should_apply_changes_for_rel) as we
+ * won't know remote_final_lsn by that time. So, we don't start new apply
+ * background worker in this case.
+ */
+ if (start && !AllTablesyncsReady())
+ return NULL;
+
+ /* First time through, initialize apply workers hashtable */
+ if (ApplyWorkersHash == NULL)
+ {
+ HASHCTL ctl;
+
+ MemSet(&ctl, 0, sizeof(ctl));
+ ctl.keysize = sizeof(TransactionId);
+ ctl.entrysize = sizeof(ApplyBgworkerEntry);
+ ctl.hcxt = ApplyContext;
+
+ ApplyWorkersHash = hash_create("logical apply workers hash", 8, &ctl,
+ HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+ }
+
+ /*
+ * Find entry for requested transaction.
+ */
+ entry = hash_search(ApplyWorkersHash, &xid, start ? HASH_ENTER : HASH_FIND,
+ &found);
+ if (found)
+ {
+ entry->wstate->pstate->status = APPLY_BGWORKER_BUSY;
+ return entry->wstate;
+ }
+ else if (!start)
+ return NULL;
+
+ /*
+ * Now, we try to get a apply background worker. If there is at least one
+ * worker in the idle list, then take one. Otherwise, we try to start a
+ * new apply background worker.
+ */
+ if (list_length(ApplyWorkersFreeList) > 0)
+ {
+ wstate = (ApplyBgworkerState *) llast(ApplyWorkersFreeList);
+ ApplyWorkersFreeList = list_delete_last(ApplyWorkersFreeList);
+ }
+ else
+ {
+ wstate = apply_bgworker_setup();
+
+ if (wstate == NULL)
+ {
+ /*
+ * If the apply background worker cannot be launched, remove entry
+ * in hash table.
+ */
+ hash_search(ApplyWorkersHash, &xid, HASH_REMOVE, &found);
+ return NULL;
+ }
+ }
+
+ /* Fill up the hash entry */
+ wstate->pstate->status = APPLY_BGWORKER_BUSY;
+ wstate->pstate->server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
+ wstate->pstate->stream_xid = xid;
+ entry->wstate = wstate;
+ entry->xid = xid;
+
+ return wstate;
+}
+
+/*
+ * Add the worker to the free list and remove the entry from hash table.
+ */
+void
+apply_bgworker_free(ApplyBgworkerState *wstate)
+{
+ MemoryContext oldctx;
+ TransactionId xid = wstate->pstate->stream_xid;
+
+ Assert(wstate->pstate->status == APPLY_BGWORKER_FINISHED);
+
+ oldctx = MemoryContextSwitchTo(ApplyContext);
+
+ hash_search(ApplyWorkersHash, &xid, HASH_REMOVE, NULL);
+
+ elog(DEBUG1, "adding finished apply worker #%u for xid %u to the idle list",
+ wstate->pstate->n, wstate->pstate->stream_xid);
+
+ ApplyWorkersFreeList = lappend(ApplyWorkersFreeList, wstate);
+
+ MemoryContextSwitchTo(oldctx);
+}
+
+/* Apply Background Worker main loop */
+static void
+LogicalApplyBgwLoop(shm_mq_handle *mqh, volatile ApplyBgworkerShared *pst)
+{
+ shm_mq_result shmq_res;
+ PGPROC *registrant;
+ ErrorContextCallback errcallback;
+ XLogRecPtr last_received = InvalidXLogRecPtr;
+
+ registrant = BackendPidGetProc(MyBgworkerEntry->bgw_notify_pid);
+ SetLatch(®istrant->procLatch);
+
+ /*
+ * Push apply error context callback. Fields will be filled during
+ * applying a change.
+ */
+ errcallback.callback = apply_error_callback;
+ errcallback.previous = error_context_stack;
+ error_context_stack = &errcallback;
+
+ for (;;)
+ {
+ void *data;
+ Size len;
+ int c;
+ StringInfoData s;
+ MemoryContext oldctx;
+ XLogRecPtr start_lsn;
+ XLogRecPtr end_lsn;
+ TimestampTz send_time;
+
+ CHECK_FOR_INTERRUPTS();
+
+ /* Ensure we are reading the data into our memory context. */
+ oldctx = MemoryContextSwitchTo(ApplyMessageContext);
+
+ shmq_res = shm_mq_receive(mqh, &len, &data, false);
+
+ if (shmq_res != SHM_MQ_SUCCESS)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("lost connection to the main apply worker")));
+
+ if (len == 0)
+ break;
+
+ s.cursor = 0;
+ s.maxlen = -1;
+ s.data = (char *) data;
+ s.len = len;
+
+ /*
+ * We use first byte of message for additional communication between
+ * main Logical replication worker and apply bgworkers, so if it
+ * differs from 'w', then process it first.
+ */
+ c = pq_getmsgbyte(&s);
+ switch (c)
+ {
+ /* End message of streaming chunk */
+ case LOGICAL_REP_MSG_STREAM_STOP:
+ elog(DEBUG1, "[Apply BGW #%u] ended processing streaming chunk,"
+ "waiting on shm_mq_receive", pst->n);
+
+ apply_bgworker_set_status(APPLY_BGWORKER_READY);
+
+ SetLatch(®istrant->procLatch);
+
+ in_streamed_transaction = false;
+ pgstat_report_activity(STATE_IDLE, NULL);
+ continue;
+ case 'w':
+ break;
+ default:
+ elog(ERROR, "[Apply BGW #%u] unexpected message \"%c\"",
+ pst->n, c);
+ break;
+ }
+
+ start_lsn = pq_getmsgint64(&s);
+ end_lsn = pq_getmsgint64(&s);
+ send_time = pq_getmsgint64(&s);
+
+ if (last_received < start_lsn)
+ last_received = start_lsn;
+
+ if (last_received < end_lsn)
+ last_received = end_lsn;
+
+ /*
+ * TO IMPROVE: Do we need to display the apply background worker's
+ * information in pg_stat_replication ?
+ */
+ UpdateWorkerStats(last_received, send_time, false);
+
+ apply_dispatch(&s);
+
+ if (ConfigReloadPending)
+ {
+ ConfigReloadPending = false;
+ ProcessConfigFile(PGC_SIGHUP);
+ }
+
+ MemoryContextSwitchTo(oldctx);
+ MemoryContextReset(ApplyMessageContext);
+ }
+
+ MemoryContextSwitchTo(TopMemoryContext);
+ MemoryContextReset(ApplyContext);
+
+ /* Pop the error context stack */
+ error_context_stack = errcallback.previous;
+
+ elog(DEBUG1, "[Apply BGW #%u] exiting", pst->n);
+
+ /* Signal main process that we are done. */
+ SetLatch(®istrant->procLatch);
+}
+
+/*
+ * Set the exit status so that the main apply worker can realize we have
+ * shutdown.
+ */
+static void
+ApplyBgwShutdown(int code, Datum arg)
+{
+ SpinLockAcquire(&MyParallelState->mutex);
+ MyParallelState->status = APPLY_BGWORKER_EXIT;
+ SpinLockRelease(&MyParallelState->mutex);
+
+ dsm_detach((dsm_segment *) DatumGetPointer(arg));
+}
+
+/*
+ * Apply Background Worker entry point
+ */
+void
+ApplyBgworkerMain(Datum main_arg)
+{
+ volatile ApplyBgworkerShared *pst;
+
+ dsm_handle handle;
+ dsm_segment *seg;
+ shm_toc *toc;
+ shm_mq *mq;
+ shm_mq_handle *mqh;
+ MemoryContext oldcontext;
+ RepOriginId originid;
+ int worker_slot = DatumGetInt32(main_arg);
+ char originname[NAMEDATALEN];
+
+ /* Attach to slot */
+ logicalrep_worker_attach(worker_slot);
+
+ MemoryContextSwitchTo(TopMemoryContext);
+
+ /* Load the subscription into persistent memory context. */
+ ApplyContext = AllocSetContextCreate(TopMemoryContext,
+ "ApplyContext",
+ ALLOCSET_DEFAULT_SIZES);
+
+ /*
+ * Init the ApplyMessageContext which we clean up after each replication
+ * protocol message.
+ */
+ ApplyMessageContext = AllocSetContextCreate(ApplyContext,
+ "ApplyMessageContext",
+ ALLOCSET_DEFAULT_SIZES);
+
+ /* Setup signal handling */
+ pqsignal(SIGHUP, SignalHandlerForConfigReload);
+ pqsignal(SIGTERM, die);
+ BackgroundWorkerUnblockSignals();
+
+ /*
+ * Connect to the dynamic shared memory segment.
+ *
+ * The backend that registered this worker passed us the ID of a shared
+ * memory segment to which we must attach for further instructions. In
+ * order to attach to dynamic shared memory, we need a resource owner.
+ * Once we've mapped the segment in our address space, attach to the table
+ * of contents so we can locate the various data structures we'll need to
+ * find within the segment.
+ */
+ CurrentResourceOwner = ResourceOwnerCreate(NULL, "Logical apply worker");
+ memcpy(&handle, MyBgworkerEntry->bgw_extra, sizeof(dsm_handle));
+ seg = dsm_attach(handle);
+ if (seg == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("unable to map dynamic shared memory segment")));
+ toc = shm_toc_attach(PG_LOGICAL_APPLY_SHM_MAGIC, dsm_segment_address(seg));
+ if (toc == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("bad magic number in dynamic shared memory segment")));
+
+ before_shmem_exit(ApplyBgwShutdown, PointerGetDatum(seg));
+
+ /* Look up the parallel state. */
+ pst = shm_toc_lookup(toc, APPLY_BGWORKER_KEY_SHARED, false);
+ MyParallelState = pst;
+
+ /*
+ * Attach to the message queue.
+ */
+ mq = shm_toc_lookup(toc, APPLY_BGWORKER_KEY_MQ, false);
+ shm_mq_set_receiver(mq, MyProc);
+ mqh = shm_mq_attach(mq, seg, NULL);
+
+ /* Run as replica session replication role. */
+ SetConfigOption("session_replication_role", "replica",
+ PGC_SUSET, PGC_S_OVERRIDE);
+
+ /* Connect to our database. */
+ BackgroundWorkerInitializeConnectionByOid(MyLogicalRepWorker->dbid,
+ MyLogicalRepWorker->userid,
+ 0);
+
+ /*
+ * Set always-secure search path, so malicious users can't redirect user
+ * code (e.g. pg_index.indexprs).
+ */
+ SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
+
+ /*
+ * Set the client encoding to the database encoding, since that is what
+ * the leader will expect.
+ */
+ SetClientEncoding(GetDatabaseEncoding());
+
+ stream_xid = pst->stream_xid;
+
+ StartTransactionCommand();
+ oldcontext = MemoryContextSwitchTo(ApplyContext);
+
+ MySubscription = GetSubscription(MyLogicalRepWorker->subid, true);
+ if (!MySubscription)
+ {
+ ereport(LOG,
+ (errmsg("logical replication apply worker for subscription %u will not "
+ "start because the subscription was removed during startup",
+ MyLogicalRepWorker->subid)));
+ proc_exit(0);
+ }
+
+ MySubscriptionValid = true;
+ MemoryContextSwitchTo(oldcontext);
+
+ /* Setup synchronous commit according to the user's wishes */
+ SetConfigOption("synchronous_commit", MySubscription->synccommit,
+ PGC_BACKEND, PGC_S_OVERRIDE);
+
+ /* Keep us informed about subscription changes. */
+ CacheRegisterSyscacheCallback(SUBSCRIPTIONOID,
+ subscription_change_cb,
+ (Datum) 0);
+
+ CommitTransactionCommand();
+
+ /* Setup replication origin tracking. */
+ StartTransactionCommand();
+ snprintf(originname, sizeof(originname), "pg_%u", MySubscription->oid);
+ originid = replorigin_by_name(originname, true);
+ if (!OidIsValid(originid))
+ originid = replorigin_create(originname);
+
+ /*
+ * The apply background worker doesn't need to monopolize this replication
+ * origin which was already acquired by its leader process.
+ */
+ replorigin_session_setup(originid, false);
+ replorigin_session_origin = originid;
+ CommitTransactionCommand();
+
+ /*
+ * Allocate the origin name in long-lived context for error context
+ * message.
+ */
+ apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
+ originname);
+
+ /*
+ * Indicate that we're fully initialized and ready to begin the main part
+ * of the apply operation.
+ */
+ apply_bgworker_set_status(APPLY_BGWORKER_ATTACHED);
+
+ elog(DEBUG1, "[Apply BGW #%u] started", pst->n);
+
+ LogicalApplyBgwLoop(mqh, pst);
+
+ /*
+ * We're done. Explicitly detach the shared memory segment so that we
+ * don't get a resource leak warning at commit time. This will fire any
+ * on_dsm_detach callbacks we've registered, as well. Once that's done,
+ * we can go ahead and exit.
+ */
+ dsm_detach(seg);
+ proc_exit(0);
+}
+
+/*
+ * Set up a dynamic shared memory segment.
+ *
+ * We set up a control region that contains a ApplyBgworkerShared,
+ * plus one region per message queue. There are as many message queues as
+ * the number of workers.
+ */
+static void
+apply_bgworker_setup_dsm(ApplyBgworkerState *wstate)
+{
+ shm_toc_estimator e;
+ Size segsize;
+ dsm_segment *seg;
+ shm_toc *toc;
+ ApplyBgworkerShared *pst;
+ shm_mq *mq;
+ int64 queue_size = 160000000; /* 16 MB for now */
+
+ /*
+ * Estimate how much shared memory we need.
+ *
+ * Because the TOC machinery may choose to insert padding of oddly-sized
+ * requests, we must estimate each chunk separately.
+ *
+ * We need one key to register the location of the header, and we need
+ * another key to track the location of the message queue.
+ */
+ shm_toc_initialize_estimator(&e);
+ shm_toc_estimate_chunk(&e, sizeof(ApplyBgworkerShared));
+ shm_toc_estimate_chunk(&e, (Size) queue_size);
+
+ shm_toc_estimate_keys(&e, 1 + 1);
+ segsize = shm_toc_estimate(&e);
+
+ /* Create the shared memory segment and establish a table of contents. */
+ seg = dsm_create(shm_toc_estimate(&e), 0);
+ toc = shm_toc_create(PG_LOGICAL_APPLY_SHM_MAGIC, dsm_segment_address(seg),
+ segsize);
+
+ /* Set up the header region. */
+ pst = shm_toc_allocate(toc, sizeof(ApplyBgworkerShared));
+ SpinLockInit(&pst->mutex);
+ pst->status = APPLY_BGWORKER_BUSY;
+ pst->server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
+ pst->stream_xid = stream_xid;
+ pst->n = list_length(ApplyWorkersList) + 1;
+
+ shm_toc_insert(toc, APPLY_BGWORKER_KEY_SHARED, pst);
+
+ /* Set up one message queue per worker, plus one. */
+ mq = shm_mq_create(shm_toc_allocate(toc, (Size) queue_size),
+ (Size) queue_size);
+ shm_toc_insert(toc, APPLY_BGWORKER_KEY_MQ, mq);
+ shm_mq_set_sender(mq, MyProc);
+
+ /* Attach the queues. */
+ wstate->mq_handle = shm_mq_attach(mq, seg, NULL);
+
+ /* Return results to caller. */
+ wstate->dsm_seg = seg;
+ wstate->pstate = pst;
+}
+
+/*
+ * Start apply background worker process and allocate shared memory for it.
+ */
+static ApplyBgworkerState *
+apply_bgworker_setup(void)
+{
+ MemoryContext oldcontext;
+ bool launched;
+ ApplyBgworkerState *wstate;
+ int napplyworkers;
+
+ elog(DEBUG1, "setting up apply worker #%u", list_length(ApplyWorkersList) + 1);
+
+ /* Check If there are free worker slot(s) */
+ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+ napplyworkers = logicalrep_apply_background_worker_count(MyLogicalRepWorker->subid);
+ LWLockRelease(LogicalRepWorkerLock);
+ if (napplyworkers >= max_apply_bgworkers_per_subscription)
+ return NULL;
+
+ oldcontext = MemoryContextSwitchTo(ApplyContext);
+
+ wstate = (ApplyBgworkerState *) palloc0(sizeof(ApplyBgworkerState));
+
+ /* Setup shared memory */
+ apply_bgworker_setup_dsm(wstate);
+
+ launched = logicalrep_worker_launch(MyLogicalRepWorker->dbid,
+ MySubscription->oid,
+ MySubscription->name,
+ MyLogicalRepWorker->userid,
+ InvalidOid,
+ dsm_segment_handle(wstate->dsm_seg));
+
+ if (launched)
+ {
+ /* Wait for worker to attach. */
+ apply_bgworker_wait_for(wstate, APPLY_BGWORKER_ATTACHED);
+
+ ApplyWorkersList = lappend(ApplyWorkersList, wstate);
+ }
+ else
+ {
+ pfree(wstate);
+ wstate = NULL;
+ }
+
+ MemoryContextSwitchTo(oldcontext);
+
+ return wstate;
+}
+
+/*
+ * Send the data to the specified apply background worker via shared-memory queue.
+ */
+void
+apply_bgworker_send_data(ApplyBgworkerState *wstate, Size nbytes, const void *data)
+{
+ shm_mq_result result;
+
+ result = shm_mq_send(wstate->mq_handle, nbytes, data, false, true);
+
+ if (result != SHM_MQ_SUCCESS)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("could not send tuples to shared-memory queue")));
+}
+
+/*
+ * Wait until the status of apply background worker reaches the
+ * 'wait_for_status'
+ */
+void
+apply_bgworker_wait_for(ApplyBgworkerState *wstate,
+ ApplyBgworkerStatus wait_for_status)
+{
+ for (;;)
+ {
+ char status;
+
+ SpinLockAcquire(&wstate->pstate->mutex);
+ status = wstate->pstate->status;
+ SpinLockRelease(&wstate->pstate->mutex);
+
+ /* Done if already in correct status. */
+ if (status == wait_for_status)
+ break;
+
+ /* If any workers (or the postmaster) have died, we have failed. */
+ if (status == APPLY_BGWORKER_EXIT)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("background worker %u failed to apply transaction %u",
+ wstate->pstate->n, wstate->pstate->stream_xid)));
+
+ /* Wait to be signalled. */
+ WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
+ WAIT_EVENT_LOGICAL_APPLY_WORKER_STATE_CHANGE);
+
+ /* Reset the latch so we don't spin. */
+ ResetLatch(MyLatch);
+
+ /* An interrupt may have occurred while we were waiting. */
+ CHECK_FOR_INTERRUPTS();
+ }
+}
+
+/*
+ * Check the status of workers and report an error if any apply background
+ * worker has exited unexpectedly.
+ *
+ * Exit if any relation is not in the READY state and if any worker is handling
+ * the streaming transaction at the same time. Because for streaming
+ * transactions that is being applied in apply background worker, we cannot
+ * decide whether to apply the change for a relation that is not in the READY
+ * state (see should_apply_changes_for_rel) as we won't know remote_final_lsn
+ * by that time.
+ */
+void
+apply_bgworker_check_status(void)
+{
+ ListCell *lc;
+
+ if (am_apply_bgworker() || MySubscription->stream != SUBSTREAM_APPLY)
+ return;
+
+ foreach(lc, ApplyWorkersList)
+ {
+ ApplyBgworkerState *wstate = (ApplyBgworkerState *) lfirst(lc);
+
+ /*
+ * We don't lock here as in the worst case we will just detect the
+ * failure of worker a bit later.
+ */
+ if (wstate->pstate->status == APPLY_BGWORKER_EXIT)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("background worker %u exited unexpectedly",
+ wstate->pstate->n)));
+ }
+
+ if (list_length(ApplyWorkersFreeList) != list_length(ApplyWorkersList) &&
+ !AllTablesyncsReady())
+ {
+ ereport(LOG,
+ (errmsg("logical replication apply workers for subscription \"%s\" will restart",
+ MySubscription->name),
+ errdetail("Cannot handle streamed replication transaction by apply "
+ "bgworkers until all tables are synchronized")));
+
+ proc_exit(0);
+ }
+}
+
+/* Set the apply background worker status */
+void
+apply_bgworker_set_status(ApplyBgworkerStatus status)
+{
+ if (!am_apply_bgworker())
+ return;
+
+ elog(DEBUG1, "[Apply BGW #%u] set status to %d", MyParallelState->n, status);
+
+ SpinLockAcquire(&MyParallelState->mutex);
+ MyParallelState->status = status;
+ SpinLockRelease(&MyParallelState->mutex);
+}
+
+/*
+ * Define a savepoint for a subxact in apply background worker if needed.
+ *
+ * Inside apply background worker we can figure out that new subtransaction was
+ * started if new change arrived with different xid. In that case we can define
+ * named savepoint, so that we were able to commit/rollback it separately
+ * later.
+ * Special case is if the first change comes from subtransaction, then
+ * we check that current_xid differs from stream_xid.
+ */
+void
+apply_bgworker_subxact_info_add(TransactionId current_xid)
+{
+ if (current_xid != stream_xid &&
+ !list_member_int(subxactlist, (int) current_xid))
+ {
+ MemoryContext oldctx;
+ char spname[MAXPGPATH];
+
+ snprintf(spname, MAXPGPATH, "savepoint_for_xid_%u", current_xid);
+
+ elog(DEBUG1, "[Apply BGW #%u] defining savepoint %s",
+ MyParallelState->n, spname);
+
+ DefineSavepoint(spname);
+ CommitTransactionCommand();
+
+ oldctx = MemoryContextSwitchTo(ApplyContext);
+ subxactlist = lappend_int(subxactlist, (int) current_xid);
+ MemoryContextSwitchTo(oldctx);
+ }
+}
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
index aa2427ba73..00fdde0830 100644
--- a/src/backend/replication/logical/decode.c
+++ b/src/backend/replication/logical/decode.c
@@ -651,9 +651,10 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
{
for (i = 0; i < parsed->nsubxacts; i++)
{
- ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr);
+ ReorderBufferForget(ctx->reorder, parsed->subxacts[i], buf->origptr,
+ commit_time);
}
- ReorderBufferForget(ctx->reorder, xid, buf->origptr);
+ ReorderBufferForget(ctx->reorder, xid, buf->origptr, commit_time);
return;
}
@@ -821,10 +822,11 @@ DecodeAbort(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
for (i = 0; i < parsed->nsubxacts; i++)
{
ReorderBufferAbort(ctx->reorder, parsed->subxacts[i],
- buf->record->EndRecPtr);
+ buf->record->EndRecPtr, abort_time);
}
- ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr);
+ ReorderBufferAbort(ctx->reorder, xid, buf->record->EndRecPtr,
+ abort_time);
}
/* update the decoding stats */
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 2bdab53e19..dd68651314 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -54,6 +54,7 @@
int max_logical_replication_workers = 4;
int max_sync_workers_per_subscription = 2;
+int max_apply_bgworkers_per_subscription = 3;
LogicalRepWorker *MyLogicalRepWorker = NULL;
@@ -73,6 +74,7 @@ static void logicalrep_launcher_onexit(int code, Datum arg);
static void logicalrep_worker_onexit(int code, Datum arg);
static void logicalrep_worker_detach(void);
static void logicalrep_worker_cleanup(LogicalRepWorker *worker);
+static void logicalrep_worker_stop_internal(LogicalRepWorker *worker);
static bool on_commit_launcher_wakeup = false;
@@ -223,6 +225,10 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
{
LogicalRepWorker *w = &LogicalRepCtx->workers[i];
+ /* We only need main apply worker or table sync worker here */
+ if (w->subworker)
+ continue;
+
if (w->in_use && w->subid == subid && w->relid == relid &&
(!only_running || w->proc))
{
@@ -259,11 +265,11 @@ logicalrep_workers_find(Oid subid, bool only_running)
}
/*
- * Start new apply background worker, if possible.
+ * Start new background worker, if possible.
*/
-void
+bool
logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
- Oid relid)
+ Oid relid, dsm_handle subworker_dsm)
{
BackgroundWorker bgw;
BackgroundWorkerHandle *bgw_handle;
@@ -273,6 +279,10 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid,
LogicalRepWorker *worker = NULL;
int nsyncworkers;
TimestampTz now;
+ bool is_subworker = (subworker_dsm != DSM_HANDLE_INVALID);
+
+ /* We don't support table sync in subworker */
+ Assert(!(is_subworker && OidIsValid(relid)));
ereport(DEBUG1,
(errmsg_internal("starting logical replication worker for subscription \"%s\"",
@@ -350,7 +360,7 @@ retry:
if (OidIsValid(relid) && nsyncworkers >= max_sync_workers_per_subscription)
{
LWLockRelease(LogicalRepWorkerLock);
- return;
+ return false;
}
/*
@@ -364,7 +374,7 @@ retry:
(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
errmsg("out of logical replication worker slots"),
errhint("You might need to increase max_logical_replication_workers.")));
- return;
+ return false;
}
/* Prepare the worker slot. */
@@ -379,6 +389,7 @@ retry:
worker->relstate = SUBREL_STATE_UNKNOWN;
worker->relstate_lsn = InvalidXLogRecPtr;
worker->stream_fileset = NULL;
+ worker->subworker = is_subworker;
worker->last_lsn = InvalidXLogRecPtr;
TIMESTAMP_NOBEGIN(worker->last_send_time);
TIMESTAMP_NOBEGIN(worker->last_recv_time);
@@ -396,19 +407,30 @@ retry:
BGWORKER_BACKEND_DATABASE_CONNECTION;
bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres");
- snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
+
+ if (!is_subworker)
+ snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
+ else
+ snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyBgworkerMain");
+
if (OidIsValid(relid))
snprintf(bgw.bgw_name, BGW_MAXLEN,
"logical replication worker for subscription %u sync %u", subid, relid);
- else
+ else if (!is_subworker)
snprintf(bgw.bgw_name, BGW_MAXLEN,
"logical replication worker for subscription %u", subid);
+ else
+ snprintf(bgw.bgw_name, BGW_MAXLEN,
+ "logical replication background apply worker for subscription %u ", subid);
snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication worker");
bgw.bgw_restart_time = BGW_NEVER_RESTART;
bgw.bgw_notify_pid = MyProcPid;
bgw.bgw_main_arg = Int32GetDatum(slot);
+ if (is_subworker)
+ memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
+
if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
{
/* Failed to start worker, so clean up the worker slot. */
@@ -421,11 +443,13 @@ retry:
(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
errmsg("out of background worker slots"),
errhint("You might need to increase max_worker_processes.")));
- return;
+ return false;
}
/* Now wait until it attaches. */
WaitForReplicationWorkerAttach(worker, generation, bgw_handle);
+
+ return true;
}
/*
@@ -436,7 +460,6 @@ void
logicalrep_worker_stop(Oid subid, Oid relid)
{
LogicalRepWorker *worker;
- uint16 generation;
LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
@@ -449,6 +472,22 @@ logicalrep_worker_stop(Oid subid, Oid relid)
return;
}
+ logicalrep_worker_stop_internal(worker);
+
+ LWLockRelease(LogicalRepWorkerLock);
+}
+
+/*
+ * Workhorse for logicalrep_worker_stop() and logicalrep_worker_detach(). Stop
+ * the worker and wait for it to die.
+ */
+static void
+logicalrep_worker_stop_internal(LogicalRepWorker *worker)
+{
+ uint16 generation;
+
+ Assert(LWLockHeldByMe(LogicalRepWorkerLock));
+
/*
* Remember which generation was our worker so we can check if what we see
* is still the same one.
@@ -485,10 +524,7 @@ logicalrep_worker_stop(Oid subid, Oid relid)
* different, meaning that a different worker has taken the slot.
*/
if (!worker->in_use || worker->generation != generation)
- {
- LWLockRelease(LogicalRepWorkerLock);
return;
- }
/* Worker has assigned proc, so it has started. */
if (worker->proc)
@@ -522,8 +558,6 @@ logicalrep_worker_stop(Oid subid, Oid relid)
LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
}
-
- LWLockRelease(LogicalRepWorkerLock);
}
/*
@@ -599,6 +633,29 @@ logicalrep_worker_attach(int slot)
static void
logicalrep_worker_detach(void)
{
+ /*
+ * If we are the main apply worker, stop all the apply background workers
+ * we started before.
+ */
+ if (!MyLogicalRepWorker->subworker)
+ {
+ List *workers;
+ ListCell *lc;
+
+ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+
+ workers = logicalrep_workers_find(MyLogicalRepWorker->subid, true);
+ foreach(lc, workers)
+ {
+ LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
+
+ if (w->subworker)
+ logicalrep_worker_stop_internal(w);
+ }
+
+ LWLockRelease(LogicalRepWorkerLock);
+ }
+
/* Block concurrent access. */
LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
@@ -621,6 +678,7 @@ logicalrep_worker_cleanup(LogicalRepWorker *worker)
worker->userid = InvalidOid;
worker->subid = InvalidOid;
worker->relid = InvalidOid;
+ worker->subworker = false;
}
/*
@@ -679,6 +737,30 @@ logicalrep_sync_worker_count(Oid subid)
return res;
}
+/*
+ * Count the number of registered (not necessarily running) apply background
+ * worker for a subscription.
+ */
+int
+logicalrep_apply_background_worker_count(Oid subid)
+{
+ int i;
+ int res = 0;
+
+ Assert(LWLockHeldByMe(LogicalRepWorkerLock));
+
+ /* Search for attached worker for a given subscription id. */
+ for (i = 0; i < max_logical_replication_workers; i++)
+ {
+ LogicalRepWorker *w = &LogicalRepCtx->workers[i];
+
+ if (w->subid == subid && w->subworker)
+ res++;
+ }
+
+ return res;
+}
+
/*
* ApplyLauncherShmemSize
* Compute space needed for replication launcher shared memory
@@ -868,7 +950,7 @@ ApplyLauncherMain(Datum main_arg)
wait_time = wal_retrieve_retry_interval;
logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
- sub->owner, InvalidOid);
+ sub->owner, InvalidOid, DSM_HANDLE_INVALID);
}
}
diff --git a/src/backend/replication/logical/origin.c b/src/backend/replication/logical/origin.c
index 21937ab2d3..9273011b0a 100644
--- a/src/backend/replication/logical/origin.c
+++ b/src/backend/replication/logical/origin.c
@@ -1063,12 +1063,21 @@ ReplicationOriginExitCleanup(int code, Datum arg)
* array doesn't have to be searched when calling
* replorigin_session_advance().
*
- * Obviously only one such cached origin can exist per process and the current
+ * Normally only one such cached origin can exist per process and the current
* cached value can only be set again after the previous value is torn down
* with replorigin_session_reset().
+ *
+ * However, If must_acquire is false, we allow process to get the slot which is
+ * already acquired by other process. It's safe because 1) The only caller
+ * (apply background workers) will maintain the commit order by allowing only
+ * one process to commit at a time, so no two workers will be operating on the
+ * same origin at the same time (see comments in logical/worker.c). 2) Even
+ * though we try to advance the session's origin concurrently, it's safe to do
+ * so as we change/advance the session_origin LSNs under replicate_state
+ * LWLock.
*/
void
-replorigin_session_setup(RepOriginId node)
+replorigin_session_setup(RepOriginId node, bool must_acquire)
{
static bool registered_cleanup;
int i;
@@ -1110,7 +1119,7 @@ replorigin_session_setup(RepOriginId node)
if (curstate->roident != node)
continue;
- else if (curstate->acquired_by != 0)
+ else if (curstate->acquired_by != 0 && must_acquire)
{
ereport(ERROR,
(errcode(ERRCODE_OBJECT_IN_USE),
@@ -1141,7 +1150,14 @@ replorigin_session_setup(RepOriginId node)
Assert(session_replication_state->roident != InvalidRepOriginId);
- session_replication_state->acquired_by = MyProcPid;
+ if (must_acquire)
+ session_replication_state->acquired_by = MyProcPid;
+ else if (session_replication_state->acquired_by == 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
+ errmsg("could not find correct replication state slot for replication origin with OID %u for apply background worker",
+ node),
+ errhint("There is no replication state slot set by its main apply worker.")));
LWLockRelease(ReplicationOriginLock);
@@ -1321,7 +1337,7 @@ pg_replication_origin_session_setup(PG_FUNCTION_ARGS)
name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
origin = replorigin_by_name(name, false);
- replorigin_session_setup(origin);
+ replorigin_session_setup(origin, true);
replorigin_session_origin = origin;
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index ff8513e2d2..a888038eb4 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -1166,28 +1166,47 @@ logicalrep_read_stream_commit(StringInfo in, LogicalRepCommitData *commit_data)
*/
void
logicalrep_write_stream_abort(StringInfo out, TransactionId xid,
- TransactionId subxid)
+ ReorderBufferTXN *txn, XLogRecPtr abort_lsn)
{
pq_sendbyte(out, LOGICAL_REP_MSG_STREAM_ABORT);
- Assert(TransactionIdIsValid(xid) && TransactionIdIsValid(subxid));
+ Assert(TransactionIdIsValid(xid) && TransactionIdIsValid(txn->xid));
/* transaction ID */
pq_sendint32(out, xid);
- pq_sendint32(out, subxid);
+ pq_sendint32(out, txn->xid);
+ pq_sendint64(out, abort_lsn);
+ pq_sendint64(out, txn->xact_time.abort_time);
}
/*
* Read STREAM ABORT from the output stream.
*/
void
-logicalrep_read_stream_abort(StringInfo in, TransactionId *xid,
- TransactionId *subxid)
+logicalrep_read_stream_abort(StringInfo in,
+ LogicalRepStreamAbortData *abort_data,
+ bool include_abort_lsn)
{
- Assert(xid && subxid);
+ Assert(abort_data);
+
+ abort_data->xid = pq_getmsgint(in, 4);
+ abort_data->subxid = pq_getmsgint(in, 4);
- *xid = pq_getmsgint(in, 4);
- *subxid = pq_getmsgint(in, 4);
+ /*
+ * If the version of the publisher is lower than the version of the
+ * subscriber, it may not support sending these two fields, so only take
+ * these fields when include_abort_lsn is true.
+ */
+ if (include_abort_lsn)
+ {
+ abort_data->abort_lsn = pq_getmsgint64(in);
+ abort_data->abort_time = pq_getmsgint64(in);
+ }
+ else
+ {
+ abort_data->abort_lsn = InvalidXLogRecPtr;
+ abort_data->abort_time = 0;
+ }
}
/*
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 8da5f9089c..3d2bbdb38d 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -2826,7 +2826,8 @@ ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid,
* disk.
*/
void
-ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
+ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
+ TimestampTz abort_time)
{
ReorderBufferTXN *txn;
@@ -2837,6 +2838,8 @@ ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
if (txn == NULL)
return;
+ txn->xact_time.abort_time = abort_time;
+
/* For streamed transactions notify the remote node about the abort. */
if (rbtxn_is_streamed(txn))
{
@@ -2911,7 +2914,8 @@ ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid)
* to this xid might re-create the transaction incompletely.
*/
void
-ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
+ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
+ TimestampTz abort_time)
{
ReorderBufferTXN *txn;
@@ -2922,6 +2926,8 @@ ReorderBufferForget(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn)
if (txn == NULL)
return;
+ txn->xact_time.abort_time = abort_time;
+
/* For streamed transactions notify the remote node about the abort. */
if (rbtxn_is_streamed(txn))
rb->stream_abort(rb, txn, lsn);
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 670c6fcada..8ffba7e2e5 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -568,7 +568,8 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
MySubscription->oid,
MySubscription->name,
MyLogicalRepWorker->userid,
- rstate->relid);
+ rstate->relid,
+ DSM_HANDLE_INVALID);
hentry->last_start_time = now;
}
}
@@ -589,6 +590,9 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
void
process_syncing_tables(XLogRecPtr current_lsn)
{
+ if (MyLogicalRepWorker->subworker)
+ return;
+
if (am_tablesync_worker())
process_syncing_tables_for_sync(current_lsn);
else
@@ -1273,7 +1277,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
* time this tablesync was launched.
*/
originid = replorigin_by_name(originname, false);
- replorigin_session_setup(originid);
+ replorigin_session_setup(originid, true);
replorigin_session_origin = originid;
*origin_startpos = replorigin_session_get_progress(false);
@@ -1384,7 +1388,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
true /* go backward */ , true /* WAL log */ );
UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
- replorigin_session_setup(originid);
+ replorigin_session_setup(originid, true);
replorigin_session_origin = originid;
}
else
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index fc210a9e7b..3f8dea0fc1 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -22,8 +22,25 @@
* STREAMED TRANSACTIONS
* ---------------------
* Streamed transactions (large transactions exceeding a memory limit on the
- * upstream) are not applied immediately, but instead, the data is written
- * to temporary files and then applied at once when the final commit arrives.
+ * upstream) are applied via one of two approaches.
+ *
+ * 1) Separate background workers
+ *
+ * Assign a new apply background worker (if available) as soon as the xact's
+ * first stream is received and the main apply worker will send changes to this
+ * new worker via shared memory. We keep this worker assigned till the
+ * transaction commit is received and also wait for the worker to finish at
+ * commit. This preserves commit ordering and avoids writing to and reading
+ * from file in most cases. We still need to spill if there is no worker
+ * available. We also need to allow stream_stop to complete by the background
+ * worker to avoid deadlocks because T-1's current stream of changes can update
+ * rows in conflicting order with T-2's next stream of changes.
+ *
+ * 2) Write to temporary files and apply when the final commit arrives
+ *
+ * If no worker is available to handle streamed transaction, the data is
+ * written to temporary files and then applied at once when the final commit
+ * arrives.
*
* Unlike the regular (non-streamed) case, handling streamed transactions has
* to handle aborts of both the toplevel transaction and subtransactions. This
@@ -219,20 +236,7 @@ typedef struct ApplyExecutionData
PartitionTupleRouting *proute; /* partition routing info */
} ApplyExecutionData;
-/* Struct for saving and restoring apply errcontext information */
-typedef struct ApplyErrorCallbackArg
-{
- LogicalRepMsgType command; /* 0 if invalid */
- LogicalRepRelMapEntry *rel;
-
- /* Remote node information */
- int remote_attnum; /* -1 if invalid */
- TransactionId remote_xid;
- XLogRecPtr finish_lsn;
- char *origin_name;
-} ApplyErrorCallbackArg;
-
-static ApplyErrorCallbackArg apply_error_callback_arg =
+ApplyErrorCallbackArg apply_error_callback_arg =
{
.command = 0,
.rel = NULL,
@@ -242,7 +246,7 @@ static ApplyErrorCallbackArg apply_error_callback_arg =
.origin_name = NULL,
};
-static MemoryContext ApplyMessageContext = NULL;
+MemoryContext ApplyMessageContext = NULL;
MemoryContext ApplyContext = NULL;
/* per stream context for streaming transactions */
@@ -251,27 +255,38 @@ static MemoryContext LogicalStreamingContext = NULL;
WalReceiverConn *LogRepWorkerWalRcvConn = NULL;
Subscription *MySubscription = NULL;
-static bool MySubscriptionValid = false;
+bool MySubscriptionValid = false;
bool in_remote_transaction = false;
static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;
/* fields valid only when processing streamed transaction */
-static bool in_streamed_transaction = false;
+bool in_streamed_transaction = false;
+
+TransactionId stream_xid = InvalidTransactionId;
+static ApplyBgworkerState *stream_apply_worker = NULL;
+
+/* check if we are applying the transaction in apply background worker */
+#define apply_bgworker_active() (in_streamed_transaction && stream_apply_worker != NULL)
-static TransactionId stream_xid = InvalidTransactionId;
+/*
+ * The number of changes during one streaming block (only for apply background
+ * workers)
+ */
+static uint32 nchanges = 0;
/*
* We enable skipping all data modification changes (INSERT, UPDATE, etc.) for
* the subscription if the remote transaction's finish LSN matches the subskiplsn.
* Once we start skipping changes, we don't stop it until we skip all changes of
* the transaction even if pg_subscription is updated and MySubscription->skiplsn
- * gets changed or reset during that. Also, in streaming transaction cases, we
- * don't skip receiving and spooling the changes since we decide whether or not
+ * gets changed or reset during that. Also, in streaming transaction cases (streaming = on),
+ * we don't skip receiving and spooling the changes since we decide whether or not
* to skip applying the changes when starting to apply changes. The subskiplsn is
* cleared after successfully skipping the transaction or applying non-empty
* transaction. The latter prevents the mistakenly specified subskiplsn from
- * being left.
+ * being left. Note that we cannot skip the streaming transaction in apply mode,
+ * because we cannot get the finish LSN before applying the changes.
*/
static XLogRecPtr skip_xact_finish_lsn = InvalidXLogRecPtr;
#define is_skipping_changes() (unlikely(!XLogRecPtrIsInvalid(skip_xact_finish_lsn)))
@@ -324,9 +339,6 @@ static void maybe_reread_subscription(void);
static void DisableSubscriptionAndExit(void);
-/* prototype needed because of stream_commit */
-static void apply_dispatch(StringInfo s);
-
static void apply_handle_commit_internal(LogicalRepCommitData *commit_data);
static void apply_handle_insert_internal(ApplyExecutionData *edata,
ResultRelInfo *relinfo,
@@ -359,7 +371,6 @@ static void stop_skipping_changes(void);
static void clear_subscription_skip_lsn(XLogRecPtr finish_lsn);
/* Functions for apply error callback */
-static void apply_error_callback(void *arg);
static inline void set_apply_error_context_xact(TransactionId xid, XLogRecPtr lsn);
static inline void reset_apply_error_context_info(void);
@@ -426,41 +437,76 @@ end_replication_step(void)
}
/*
- * Handle streamed transactions.
+ * Handle streamed transactions for both main apply worker and apply background
+ * worker.
*
- * If in streaming mode (receiving a block of streamed transaction), we
- * simply redirect it to a file for the proper toplevel transaction.
+ * In streaming case (receiving a block of streamed transaction), for
+ * SUBSTREAM_ON mode, we simply redirect it to a file for the proper toplevel
+ * transaction, and for SUBSTREAM_APPLY mode, we send the changes to background
+ * apply worker (LOGICAL_REP_MSG_RELATION or LOGICAL_REP_MSG_TYPE changes will
+ * also be applied in main apply worker).
*
- * Returns true for streamed transactions, false otherwise (regular mode).
+ * For non-streamed transactions, returns false;
+ * For streamed transactions, returns true if in main apply worker (except we
+ * apply streamed transaction in "apply" mode and address
+ * LOGICAL_REP_MSG_RELATION or LOGICAL_REP_MSG_TYPE changes), false otherwise.
*/
static bool
handle_streamed_transaction(LogicalRepMsgType action, StringInfo s)
{
- TransactionId xid;
+ TransactionId current_xid = InvalidTransactionId;
/* not in streaming mode */
- if (!in_streamed_transaction)
+ if (!(in_streamed_transaction || am_apply_bgworker()))
return false;
- Assert(stream_fd != NULL);
Assert(TransactionIdIsValid(stream_xid));
/*
* We should have received XID of the subxact as the first part of the
* message, so extract it.
*/
- xid = pq_getmsgint(s, 4);
+ current_xid = pq_getmsgint(s, 4);
- if (!TransactionIdIsValid(xid))
+ if (!TransactionIdIsValid(current_xid))
ereport(ERROR,
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg_internal("invalid transaction ID in streamed replication transaction")));
- /* Add the new subxact to the array (unless already there). */
- subxact_info_add(xid);
+ if (am_apply_bgworker())
+ {
+ /* define a savepoint for a subxact if needed. */
+ apply_bgworker_subxact_info_add(current_xid);
- /* write the change to the current file */
- stream_write_change(action, s);
+ return false;
+ }
+ else if (apply_bgworker_active())
+ {
+ /*
+ * If we decided to apply the changes of this transaction in an apply
+ * background worker, pass the data to the worker.
+ */
+ apply_bgworker_send_data(stream_apply_worker, s->len, s->data);
+ nchanges += 1;
+
+ /*
+ * XXX The publisher side doesn't always send relation/type update
+ * messages after the streaming transaction, so also update the
+ * relation/type in main apply worker here. See function
+ * cleanup_rel_sync_cache.
+ */
+ if (action == LOGICAL_REP_MSG_RELATION ||
+ action == LOGICAL_REP_MSG_TYPE)
+ return false;
+ }
+ else
+ {
+ /* Add the new subxact to the array (unless already there). */
+ subxact_info_add(current_xid);
+
+ /* write the change to the current file */
+ stream_write_change(action, s);
+ }
return true;
}
@@ -844,6 +890,9 @@ apply_handle_commit(StringInfo s)
apply_handle_commit_internal(&commit_data);
+ /* Check the status of apply background worker if any. */
+ apply_bgworker_check_status();
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(commit_data.end_lsn);
@@ -898,7 +947,9 @@ apply_handle_prepare_internal(LogicalRepPreparedTxnData *prepare_data)
* BeginTransactionBlock is necessary to balance the EndTransactionBlock
* called within the PrepareTransactionBlock below.
*/
- BeginTransactionBlock();
+ if (!IsTransactionBlock())
+ BeginTransactionBlock();
+
CommitTransactionCommand(); /* Completes the preceding Begin command. */
/*
@@ -950,6 +1001,9 @@ apply_handle_prepare(StringInfo s)
in_remote_transaction = false;
+ /* Check the status of apply background worker if any. */
+ apply_bgworker_check_status();
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(prepare_data.end_lsn);
@@ -1064,10 +1118,6 @@ apply_handle_rollback_prepared(StringInfo s)
/*
* Handle STREAM PREPARE.
- *
- * Logic is in two parts:
- * 1. Replay all the spooled operations
- * 2. Mark the transaction as prepared
*/
static void
apply_handle_stream_prepare(StringInfo s)
@@ -1088,24 +1138,70 @@ apply_handle_stream_prepare(StringInfo s)
logicalrep_read_stream_prepare(s, &prepare_data);
set_apply_error_context_xact(prepare_data.xid, prepare_data.prepare_lsn);
- elog(DEBUG1, "received prepare for streamed transaction %u", prepare_data.xid);
+ if (am_apply_bgworker())
+ {
+ /* Mark the transaction as prepared. */
+ apply_handle_prepare_internal(&prepare_data);
- /* Replay all the spooled operations. */
- apply_spooled_messages(prepare_data.xid, prepare_data.prepare_lsn);
+ CommitTransactionCommand();
- /* Mark the transaction as prepared. */
- apply_handle_prepare_internal(&prepare_data);
+ pgstat_report_stat(false);
- CommitTransactionCommand();
+ list_free(subxactlist);
+ subxactlist = NIL;
- pgstat_report_stat(false);
+ apply_bgworker_set_status(APPLY_BGWORKER_FINISHED);
+ }
+ else
+ {
+ ApplyBgworkerState *wstate = apply_bgworker_find_or_start(prepare_data.xid, false);
- store_flush_position(prepare_data.end_lsn);
+ elog(DEBUG1, "received prepare for streamed transaction %u",
+ prepare_data.xid);
+
+ /*
+ * This is the main apply worker. Check if we are processing this
+ * transaction in a apply background worker.
+ */
+ if (wstate)
+ {
+ apply_bgworker_send_data(wstate, s->len, s->data);
+
+ apply_bgworker_wait_for(wstate, APPLY_BGWORKER_FINISHED);
+ apply_bgworker_free(wstate);
+
+ pgstat_report_stat(false);
+ store_flush_position(prepare_data.end_lsn);
+ }
+ else
+ {
+ /*
+ * This is the main apply worker and the transaction has been
+ * serialized to file, replay all the spooled operations.
+ */
+ apply_spooled_messages(prepare_data.xid, prepare_data.prepare_lsn);
+
+ /* Mark the transaction as prepared. */
+ apply_handle_prepare_internal(&prepare_data);
+
+ CommitTransactionCommand();
+
+ pgstat_report_stat(false);
+
+ store_flush_position(prepare_data.end_lsn);
+
+ in_remote_transaction = false;
+
+ /* unlink the files with serialized changes and subxact info. */
+ stream_cleanup_files(MyLogicalRepWorker->subid, prepare_data.xid);
+ }
+ }
in_remote_transaction = false;
+ stream_apply_worker = NULL;
- /* unlink the files with serialized changes and subxact info. */
- stream_cleanup_files(MyLogicalRepWorker->subid, prepare_data.xid);
+ /* Check the status of apply background worker if any. */
+ apply_bgworker_check_status();
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(prepare_data.end_lsn);
@@ -1155,15 +1251,6 @@ apply_handle_stream_start(StringInfo s)
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg_internal("duplicate STREAM START message")));
- /*
- * Start a transaction on stream start, this transaction will be committed
- * on the stream stop unless it is a tablesync worker in which case it
- * will be committed after processing all the messages. We need the
- * transaction for handling the buffile, used for serializing the
- * streaming data and subxact info.
- */
- begin_replication_step();
-
/* notify handle methods we're processing a remote transaction */
in_streamed_transaction = true;
@@ -1177,36 +1264,92 @@ apply_handle_stream_start(StringInfo s)
set_apply_error_context_xact(stream_xid, InvalidXLogRecPtr);
+ if (am_apply_bgworker())
+ {
+ /*
+ * Make sure the handle apply_dispatch methods are aware we're in a
+ * remote transaction.
+ */
+ in_remote_transaction = true;
+
+ /* If we are in a apply background worker, begin the transaction */
+ AcceptInvalidationMessages();
+ maybe_reread_subscription();
+
+ StartTransactionCommand();
+ BeginTransactionBlock();
+ CommitTransactionCommand();
+
+ return;
+ }
+
/*
- * Initialize the worker's stream_fileset if we haven't yet. This will be
- * used for the entire duration of the worker so create it in a permanent
- * context. We create this on the very first streaming message from any
- * transaction and then use it for this and other streaming transactions.
- * Now, we could create a fileset at the start of the worker as well but
- * then we won't be sure that it will ever be used.
+ * This is the main apply worker. Check if there is any free apply
+ * background worker we can use to process this transaction.
*/
- if (MyLogicalRepWorker->stream_fileset == NULL)
+ stream_apply_worker = apply_bgworker_find_or_start(stream_xid, first_segment);
+
+ if (stream_apply_worker)
{
- MemoryContext oldctx;
+ /*
+ * If we have found a free worker or if we are already applying this
+ * transaction in an apply background worker, then we pass the data to
+ * that worker.
+ */
+ if (first_segment)
+ apply_bgworker_send_data(stream_apply_worker, s->len, s->data);
- oldctx = MemoryContextSwitchTo(ApplyContext);
+ nchanges = 0;
+ elog(DEBUG1, "starting streaming of xid %u", stream_xid);
+ }
- MyLogicalRepWorker->stream_fileset = palloc(sizeof(FileSet));
- FileSetInit(MyLogicalRepWorker->stream_fileset);
+ /*
+ * If no worker is available for the first stream start, we start to
+ * serialize all the changes of the transaction.
+ */
+ else
+ {
+ /*
+ * Start a transaction on stream start, this transaction will be
+ * committed on the stream stop unless it is a tablesync worker in
+ * which case it will be committed after processing all the messages.
+ * We need the transaction for handling the buffile, used for
+ * serializing the streaming data and subxact info.
+ */
+ begin_replication_step();
- MemoryContextSwitchTo(oldctx);
- }
+ /*
+ * Initialize the worker's stream_fileset if we haven't yet. This will
+ * be used for the entire duration of the worker so create it in a
+ * permanent context. We create this on the very first streaming
+ * message from any transaction and then use it for this and other
+ * streaming transactions. Now, we could create a fileset at the start
+ * of the worker as well but then we won't be sure that it will ever
+ * be used.
+ */
+ if (MyLogicalRepWorker->stream_fileset == NULL)
+ {
+ MemoryContext oldctx;
- /* open the spool file for this transaction */
- stream_open_file(MyLogicalRepWorker->subid, stream_xid, first_segment);
+ oldctx = MemoryContextSwitchTo(ApplyContext);
- /* if this is not the first segment, open existing subxact file */
- if (!first_segment)
- subxact_info_read(MyLogicalRepWorker->subid, stream_xid);
+ MyLogicalRepWorker->stream_fileset = palloc(sizeof(FileSet));
+ FileSetInit(MyLogicalRepWorker->stream_fileset);
- pgstat_report_activity(STATE_RUNNING, NULL);
+ MemoryContextSwitchTo(oldctx);
+ }
- end_replication_step();
+ /* open the spool file for this transaction */
+ stream_open_file(MyLogicalRepWorker->subid, stream_xid, first_segment);
+
+ /* if this is not the first segment, open existing subxact file */
+ if (!first_segment)
+ subxact_info_read(MyLogicalRepWorker->subid, stream_xid);
+
+ end_replication_step();
+ }
+
+ pgstat_report_activity(STATE_RUNNING, NULL);
}
/*
@@ -1220,44 +1363,47 @@ apply_handle_stream_stop(StringInfo s)
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg_internal("STREAM STOP message without STREAM START")));
- /*
- * Close the file with serialized changes, and serialize information about
- * subxacts for the toplevel transaction.
- */
- subxact_info_write(MyLogicalRepWorker->subid, stream_xid);
- stream_close_file();
+ if (apply_bgworker_active())
+ {
+ char action = LOGICAL_REP_MSG_STREAM_STOP;
- /* We must be in a valid transaction state */
- Assert(IsTransactionState());
+ apply_bgworker_send_data(stream_apply_worker, 1, &action);
+ apply_bgworker_wait_for(stream_apply_worker, APPLY_BGWORKER_READY);
- /* Commit the per-stream transaction */
- CommitTransactionCommand();
+ elog(DEBUG1, "stopped streaming of xid %u, %u changes streamed", stream_xid, nchanges);
+ }
+ else
+ {
+ /*
+ * Close the file with serialized changes, and serialize information
+ * about subxacts for the toplevel transaction.
+ */
+ subxact_info_write(MyLogicalRepWorker->subid, stream_xid);
+ stream_close_file();
- in_streamed_transaction = false;
+ /* We must be in a valid transaction state */
+ Assert(IsTransactionState());
+
+ /* Commit the per-stream transaction */
+ CommitTransactionCommand();
+
+ /* Reset per-stream context */
+ MemoryContextReset(LogicalStreamingContext);
+ }
- /* Reset per-stream context */
- MemoryContextReset(LogicalStreamingContext);
+ in_streamed_transaction = false;
+ stream_apply_worker = NULL;
pgstat_report_activity(STATE_IDLE, NULL);
reset_apply_error_context_info();
}
/*
- * Handle STREAM abort message.
+ * Handle STREAM ABORT message when the transaction was spilled to disk.
*/
static void
-apply_handle_stream_abort(StringInfo s)
+serialize_stream_abort(TransactionId xid, TransactionId subxid)
{
- TransactionId xid;
- TransactionId subxid;
-
- if (in_streamed_transaction)
- ereport(ERROR,
- (errcode(ERRCODE_PROTOCOL_VIOLATION),
- errmsg_internal("STREAM ABORT message without STREAM STOP")));
-
- logicalrep_read_stream_abort(s, &xid, &subxid);
-
/*
* If the two XIDs are the same, it's in fact abort of toplevel xact, so
* just delete the files with serialized info.
@@ -1339,8 +1485,137 @@ apply_handle_stream_abort(StringInfo s)
end_replication_step();
CommitTransactionCommand();
}
+}
- reset_apply_error_context_info();
+/*
+ * Handle STREAM ABORT message.
+ */
+static void
+apply_handle_stream_abort(StringInfo s)
+{
+ TransactionId xid;
+ TransactionId subxid;
+ LogicalRepStreamAbortData abort_data;
+ bool include_abort_lsn = false;
+
+ if (in_streamed_transaction)
+ ereport(ERROR,
+ (errcode(ERRCODE_PROTOCOL_VIOLATION),
+ errmsg_internal("STREAM ABORT message without STREAM STOP")));
+
+ /* Check whether the publisher sends abort_lsn and abort_time. */
+ if (am_apply_bgworker())
+ include_abort_lsn = MyParallelState->server_version >= 150000;
+
+ logicalrep_read_stream_abort(s, &abort_data, include_abort_lsn);
+
+ xid = abort_data.xid;
+ subxid = abort_data.subxid;
+
+ if (am_apply_bgworker())
+ {
+ elog(DEBUG1, "[Apply BGW #%u] aborting current transaction xid=%u, subxid=%u",
+ MyParallelState->n, GetCurrentTransactionIdIfAny(),
+ GetCurrentSubTransactionId());
+
+ /*
+ * Update origin state so we can restart streaming from correct
+ * position in case of crash.
+ */
+ if (include_abort_lsn)
+ {
+ replorigin_session_origin_lsn = abort_data.abort_lsn;
+ replorigin_session_origin_timestamp = abort_data.abort_time;
+ }
+
+ /*
+ * If the two XIDs are the same, it's in fact abort of toplevel xact,
+ * so just free the subxactlist.
+ */
+ if (subxid == xid)
+ {
+ set_apply_error_context_xact(subxid, InvalidXLogRecPtr);
+
+ AbortCurrentTransaction();
+
+ EndTransactionBlock(false);
+ CommitTransactionCommand();
+
+ in_remote_transaction = false;
+
+ list_free(subxactlist);
+ subxactlist = NIL;
+
+ apply_bgworker_set_status(APPLY_BGWORKER_FINISHED);
+ }
+ else
+ {
+ /*
+ * OK, so it's a subxact. Rollback to the savepoint.
+ *
+ * We also need to read the subxactlist, determine the offset
+ * tracked for the subxact, and truncate the list.
+ */
+ int i;
+ bool found = false;
+ char spname[MAXPGPATH];
+
+ set_apply_error_context_xact(subxid, InvalidXLogRecPtr);
+ snprintf(spname, MAXPGPATH, "savepoint_for_xid_%u", subxid);
+
+ elog(DEBUG1, "[Apply BGW #%u] rolling back to savepoint %s",
+ MyParallelState->n, spname);
+
+ for (i = list_length(subxactlist) - 1; i >= 0; i--)
+ {
+ xid = (TransactionId) list_nth_int(subxactlist, i);
+ if (xid == subxid)
+ {
+ found = true;
+ break;
+ }
+ }
+
+ if (found)
+ {
+ RollbackToSavepoint(spname);
+ CommitTransactionCommand();
+ subxactlist = list_truncate(subxactlist, i + 1);
+ }
+
+ apply_bgworker_set_status(APPLY_BGWORKER_READY);
+ }
+
+ reset_apply_error_context_info();
+ }
+ else
+ {
+ ApplyBgworkerState *wstate = apply_bgworker_find_or_start(xid, false);
+
+ /*
+ * This is the main apply worker. Check if we are processing this
+ * transaction in a apply background worker.
+ */
+ if (wstate)
+ {
+ apply_bgworker_send_data(wstate, s->len, s->data);
+
+ if (subxid == xid)
+ {
+ apply_bgworker_wait_for(wstate, APPLY_BGWORKER_FINISHED);
+ apply_bgworker_free(wstate);
+ }
+ else
+ apply_bgworker_wait_for(wstate, APPLY_BGWORKER_READY);
+ }
+
+ /*
+ * We are in main apply worker and the transaction has been serialized
+ * to file.
+ */
+ else
+ serialize_stream_abort(xid, subxid);
+ }
}
/*
@@ -1462,40 +1737,6 @@ apply_spooled_messages(TransactionId xid, XLogRecPtr lsn)
return;
}
-/*
- * Handle STREAM COMMIT message.
- */
-static void
-apply_handle_stream_commit(StringInfo s)
-{
- TransactionId xid;
- LogicalRepCommitData commit_data;
-
- if (in_streamed_transaction)
- ereport(ERROR,
- (errcode(ERRCODE_PROTOCOL_VIOLATION),
- errmsg_internal("STREAM COMMIT message without STREAM STOP")));
-
- xid = logicalrep_read_stream_commit(s, &commit_data);
- set_apply_error_context_xact(xid, commit_data.commit_lsn);
-
- elog(DEBUG1, "received commit for streamed transaction %u", xid);
-
- apply_spooled_messages(xid, commit_data.commit_lsn);
-
- apply_handle_commit_internal(&commit_data);
-
- /* unlink the files with serialized changes and subxact info */
- stream_cleanup_files(MyLogicalRepWorker->subid, xid);
-
- /* Process any tables that are being synchronized in parallel. */
- process_syncing_tables(commit_data.end_lsn);
-
- pgstat_report_activity(STATE_IDLE, NULL);
-
- reset_apply_error_context_info();
-}
-
/*
* Helper function for apply_handle_commit and apply_handle_stream_commit.
*/
@@ -2445,11 +2686,105 @@ apply_handle_truncate(StringInfo s)
end_replication_step();
}
+/*
+ * Handle STREAM COMMIT message.
+ */
+static void
+apply_handle_stream_commit(StringInfo s)
+{
+ LogicalRepCommitData commit_data;
+ TransactionId xid;
+
+ if (in_streamed_transaction)
+ ereport(ERROR,
+ (errcode(ERRCODE_PROTOCOL_VIOLATION),
+ errmsg_internal("STREAM COMMIT message without STREAM STOP")));
+
+ xid = logicalrep_read_stream_commit(s, &commit_data);
+ set_apply_error_context_xact(xid, commit_data.commit_lsn);
+
+ if (am_apply_bgworker())
+ {
+ /*
+ * Update origin state so we can restart streaming from correct
+ * position in case of crash.
+ */
+ replorigin_session_origin_lsn = commit_data.end_lsn;
+ replorigin_session_origin_timestamp = commit_data.committime;
+
+ CommitTransactionCommand();
+ EndTransactionBlock(false);
+ CommitTransactionCommand();
+
+ in_remote_transaction = false;
+
+ pgstat_report_stat(false);
+
+ list_free(subxactlist);
+ subxactlist = NIL;
+
+ apply_bgworker_set_status(APPLY_BGWORKER_FINISHED);
+ }
+ else
+ {
+ /*
+ * This is the main apply worker. Check if we are processing this
+ * transaction in an apply background worker.
+ */
+ ApplyBgworkerState *wstate = apply_bgworker_find_or_start(xid, false);
+
+ elog(DEBUG1, "received commit for streamed transaction %u", xid);
+
+ if (wstate)
+ {
+ /* Send commit message */
+ apply_bgworker_send_data(wstate, s->len, s->data);
+
+ /* Wait for apply background worker to finish */
+ apply_bgworker_wait_for(wstate, APPLY_BGWORKER_FINISHED);
+
+ pgstat_report_stat(false);
+ store_flush_position(commit_data.end_lsn);
+ stop_skipping_changes();
+
+ apply_bgworker_free(wstate);
+
+ /*
+ * The transaction is either non-empty or skipped, so we clear the
+ * subskiplsn.
+ */
+ clear_subscription_skip_lsn(commit_data.commit_lsn);
+ }
+ else
+ {
+ /*
+ * This is the main apply worker and the transaction has been
+ * serialized to file, replay all the spooled operations.
+ */
+ apply_spooled_messages(xid, commit_data.commit_lsn);
+
+ apply_handle_commit_internal(&commit_data);
+
+ /* unlink the files with serialized changes and subxact info */
+ stream_cleanup_files(MyLogicalRepWorker->subid, xid);
+ }
+ }
+
+ /* Check the status of apply background worker if any. */
+ apply_bgworker_check_status();
+
+ /* Process any tables that are being synchronized in parallel. */
+ process_syncing_tables(commit_data.end_lsn);
+
+ pgstat_report_activity(STATE_IDLE, NULL);
+
+ reset_apply_error_context_info();
+}
/*
* Logical replication protocol message dispatcher.
*/
-static void
+void
apply_dispatch(StringInfo s)
{
LogicalRepMsgType action = pq_getmsgbyte(s);
@@ -2618,6 +2953,10 @@ store_flush_position(XLogRecPtr remote_lsn)
{
FlushPosition *flushpos;
+ /* We only need to collect the LSN in main apply worker */
+ if (am_apply_bgworker())
+ return;
+
/* Need to do this in permanent context */
MemoryContextSwitchTo(ApplyContext);
@@ -2632,7 +2971,7 @@ store_flush_position(XLogRecPtr remote_lsn)
/* Update statistics of the worker. */
-static void
+void
UpdateWorkerStats(XLogRecPtr last_lsn, TimestampTz send_time, bool reply)
{
MyLogicalRepWorker->last_lsn = last_lsn;
@@ -2794,6 +3133,9 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
AcceptInvalidationMessages();
maybe_reread_subscription();
+ /* Check the status of apply background worker if any. */
+ apply_bgworker_check_status();
+
/* Process any table synchronization changes. */
process_syncing_tables(last_received);
}
@@ -3095,7 +3437,7 @@ maybe_reread_subscription(void)
/*
* Callback from subscription syscache invalidation.
*/
-static void
+void
subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue)
{
MySubscriptionValid = false;
@@ -3691,7 +4033,7 @@ ApplyWorkerMain(Datum main_arg)
originid = replorigin_by_name(originname, true);
if (!OidIsValid(originid))
originid = replorigin_create(originname);
- replorigin_session_setup(originid);
+ replorigin_session_setup(originid, true);
replorigin_session_origin = originid;
origin_startpos = replorigin_session_get_progress(false);
CommitTransactionCommand();
@@ -3738,7 +4080,7 @@ ApplyWorkerMain(Datum main_arg)
options.proto.logical.publication_names = MySubscription->publications;
options.proto.logical.binary = MySubscription->binary;
- options.proto.logical.streaming = MySubscription->stream;
+ options.proto.logical.streaming = (MySubscription->stream != SUBSTREAM_OFF);
options.proto.logical.twophase = false;
if (!am_tablesync_worker())
@@ -3896,7 +4238,8 @@ clear_subscription_skip_lsn(XLogRecPtr finish_lsn)
XLogRecPtr myskiplsn = MySubscription->skiplsn;
bool started_tx = false;
- if (likely(XLogRecPtrIsInvalid(myskiplsn)))
+ if (likely(XLogRecPtrIsInvalid(myskiplsn)) ||
+ am_apply_bgworker())
return;
if (!IsTransactionState())
@@ -3968,7 +4311,7 @@ clear_subscription_skip_lsn(XLogRecPtr finish_lsn)
}
/* Error callback to give more context info about the change being applied */
-static void
+void
apply_error_callback(void *arg)
{
ApplyErrorCallbackArg *errarg = &apply_error_callback_arg;
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index 8deae57143..6ebfa24baf 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -1833,7 +1833,7 @@ pgoutput_stream_abort(struct LogicalDecodingContext *ctx,
Assert(rbtxn_is_streamed(toptxn));
OutputPluginPrepareWrite(ctx, true);
- logicalrep_write_stream_abort(ctx->out, toptxn->xid, txn->xid);
+ logicalrep_write_stream_abort(ctx->out, toptxn->xid, txn, abort_lsn);
OutputPluginWrite(ctx, true);
cleanup_rel_sync_cache(toptxn->xid, false);
diff --git a/src/backend/utils/activity/wait_event.c b/src/backend/utils/activity/wait_event.c
index 87c15b9c6f..ba781e6f08 100644
--- a/src/backend/utils/activity/wait_event.c
+++ b/src/backend/utils/activity/wait_event.c
@@ -388,6 +388,9 @@ pgstat_get_wait_ipc(WaitEventIPC w)
case WAIT_EVENT_HASH_GROW_BUCKETS_REINSERT:
event_name = "HashGrowBucketsReinsert";
break;
+ case WAIT_EVENT_LOGICAL_APPLY_WORKER_STATE_CHANGE:
+ event_name = "LogicalApplyWorkerStateChange";
+ break;
case WAIT_EVENT_LOGICAL_SYNC_DATA:
event_name = "LogicalSyncData";
break;
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index a7cc49898b..519efff1d2 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -3220,6 +3220,18 @@ static struct config_int ConfigureNamesInt[] =
NULL, NULL, NULL
},
+ {
+ {"max_apply_bgworkers_per_subscription",
+ PGC_SIGHUP,
+ REPLICATION_SUBSCRIBERS,
+ gettext_noop("Maximum number of apply backgrand workers per subscription."),
+ NULL,
+ },
+ &max_apply_bgworkers_per_subscription,
+ 3, 0, MAX_BACKENDS,
+ NULL, NULL, NULL
+ },
+
{
{"log_rotation_age", PGC_SIGHUP, LOGGING_WHERE,
gettext_noop("Sets the amount of time to wait before forcing "
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 48ad80cf2e..b71340549d 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -360,6 +360,7 @@
#max_logical_replication_workers = 4 # taken from max_worker_processes
# (change requires restart)
#max_sync_workers_per_subscription = 2 # taken from max_logical_replication_workers
+#max_apply_bgworkers_per_subscription = 3 # taken from max_logical_replication_workers
#------------------------------------------------------------------------------
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 7cc9c72e49..6f8b30abb0 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4450,7 +4450,7 @@ getSubscriptions(Archive *fout)
if (fout->remoteVersion >= 140000)
appendPQExpBufferStr(query, " s.substream,\n");
else
- appendPQExpBufferStr(query, " false AS substream,\n");
+ appendPQExpBufferStr(query, " 'f' AS substream,\n");
if (fout->remoteVersion >= 150000)
appendPQExpBufferStr(query,
@@ -4580,8 +4580,10 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
if (strcmp(subinfo->subbinary, "t") == 0)
appendPQExpBufferStr(query, ", binary = true");
- if (strcmp(subinfo->substream, "f") != 0)
+ if (strcmp(subinfo->substream, "t") == 0)
appendPQExpBufferStr(query, ", streaming = on");
+ else if (strcmp(subinfo->substream, "a") == 0)
+ appendPQExpBufferStr(query, ", streaming = apply");
if (strcmp(subinfo->subtwophasestate, two_phase_disabled) != 0)
appendPQExpBufferStr(query, ", two_phase = on");
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index d1260f590c..9b394a45fe 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -68,7 +68,8 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
bool subbinary; /* True if the subscription wants the
* publisher to send data in binary */
- bool substream; /* Stream in-progress transactions. */
+ char substream; /* Stream in-progress transactions.
+ * See SUBSTREAM_xxx constants. */
char subtwophasestate; /* Stream two-phase transactions */
@@ -109,7 +110,7 @@ typedef struct Subscription
bool enabled; /* Indicates if the subscription is enabled */
bool binary; /* Indicates if the subscription wants data in
* binary format */
- bool stream; /* Allow streaming in-progress transactions. */
+ char stream; /* Allow streaming in-progress transactions. */
char twophasestate; /* Allow streaming two-phase transactions */
bool disableonerr; /* Indicates if the subscription should be
* automatically disabled if a worker error
@@ -120,6 +121,18 @@ typedef struct Subscription
List *publications; /* List of publication names to subscribe to */
} Subscription;
+/* Disallow streaming in-progress transactions */
+#define SUBSTREAM_OFF 'f'
+
+/*
+ * Streaming transactions are written to a temporary file and applied only
+ * after the transaction is committed on upstream.
+ */
+#define SUBSTREAM_ON 't'
+
+/* Streaming transactions are applied immediately via a background worker */
+#define SUBSTREAM_APPLY 'a'
+
extern Subscription *GetSubscription(Oid subid, bool missing_ok);
extern void FreeSubscription(Subscription *sub);
extern void DisableSubscription(Oid subid);
diff --git a/src/include/replication/logicallauncher.h b/src/include/replication/logicallauncher.h
index f1e2821e25..ac8ef94381 100644
--- a/src/include/replication/logicallauncher.h
+++ b/src/include/replication/logicallauncher.h
@@ -14,6 +14,7 @@
extern PGDLLIMPORT int max_logical_replication_workers;
extern PGDLLIMPORT int max_sync_workers_per_subscription;
+extern PGDLLIMPORT int max_apply_bgworkers_per_subscription;
extern void ApplyLauncherRegister(void);
extern void ApplyLauncherMain(Datum main_arg);
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index a771ab8ff3..7bba77c9e7 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -175,6 +175,17 @@ typedef struct LogicalRepRollbackPreparedTxnData
char gid[GIDSIZE];
} LogicalRepRollbackPreparedTxnData;
+/*
+ * Transaction protocol information for stream abort.
+ */
+typedef struct LogicalRepStreamAbortData
+{
+ TransactionId xid;
+ TransactionId subxid;
+ XLogRecPtr abort_lsn;
+ TimestampTz abort_time;
+} LogicalRepStreamAbortData;
+
extern void logicalrep_write_begin(StringInfo out, ReorderBufferTXN *txn);
extern void logicalrep_read_begin(StringInfo in,
LogicalRepBeginData *begin_data);
@@ -246,9 +257,11 @@ extern void logicalrep_write_stream_commit(StringInfo out, ReorderBufferTXN *txn
extern TransactionId logicalrep_read_stream_commit(StringInfo out,
LogicalRepCommitData *commit_data);
extern void logicalrep_write_stream_abort(StringInfo out, TransactionId xid,
- TransactionId subxid);
-extern void logicalrep_read_stream_abort(StringInfo in, TransactionId *xid,
- TransactionId *subxid);
+ ReorderBufferTXN *txn,
+ XLogRecPtr abort_lsn);
+extern void logicalrep_read_stream_abort(StringInfo in,
+ LogicalRepStreamAbortData *abort_data,
+ bool include_abort_lsn);
extern char *logicalrep_message_type(LogicalRepMsgType action);
#endif /* LOGICAL_PROTO_H */
diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h
index cd1b6e8afc..6a1af7f13c 100644
--- a/src/include/replication/logicalworker.h
+++ b/src/include/replication/logicalworker.h
@@ -13,6 +13,7 @@
#define LOGICALWORKER_H
extern void ApplyWorkerMain(Datum main_arg);
+extern void ApplyBgworkerMain(Datum main_arg);
extern bool IsLogicalWorker(void);
diff --git a/src/include/replication/origin.h b/src/include/replication/origin.h
index 2d1b5e5c28..c7389b40a7 100644
--- a/src/include/replication/origin.h
+++ b/src/include/replication/origin.h
@@ -53,7 +53,7 @@ extern XLogRecPtr replorigin_get_progress(RepOriginId node, bool flush);
extern void replorigin_session_advance(XLogRecPtr remote_commit,
XLogRecPtr local_commit);
-extern void replorigin_session_setup(RepOriginId node);
+extern void replorigin_session_setup(RepOriginId node, bool must_acquire);
extern void replorigin_session_reset(void);
extern XLogRecPtr replorigin_session_get_progress(bool flush);
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index 4a01f877e5..aba1640f4d 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -301,6 +301,7 @@ typedef struct ReorderBufferTXN
{
TimestampTz commit_time;
TimestampTz prepare_time;
+ TimestampTz abort_time;
} xact_time;
/*
@@ -647,9 +648,11 @@ extern void ReorderBufferFinishPrepared(ReorderBuffer *rb, TransactionId xid,
extern void ReorderBufferAssignChild(ReorderBuffer *, TransactionId, TransactionId, XLogRecPtr commit_lsn);
extern void ReorderBufferCommitChild(ReorderBuffer *, TransactionId, TransactionId,
XLogRecPtr commit_lsn, XLogRecPtr end_lsn);
-extern void ReorderBufferAbort(ReorderBuffer *, TransactionId, XLogRecPtr lsn);
+extern void ReorderBufferAbort(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
+ TimestampTz abort_time);
extern void ReorderBufferAbortOld(ReorderBuffer *, TransactionId xid);
-extern void ReorderBufferForget(ReorderBuffer *, TransactionId, XLogRecPtr lsn);
+extern void ReorderBufferForget(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
+ TimestampTz abort_time);
extern void ReorderBufferInvalidate(ReorderBuffer *, TransactionId, XLogRecPtr lsn);
extern void ReorderBufferSetBaseSnapshot(ReorderBuffer *, TransactionId, XLogRecPtr lsn, struct SnapshotData *snap);
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 901845abc2..08e63de013 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -17,8 +17,11 @@
#include "access/xlogdefs.h"
#include "catalog/pg_subscription.h"
#include "datatype/timestamp.h"
+#include "replication/logicalrelation.h"
#include "storage/fileset.h"
#include "storage/lock.h"
+#include "storage/shm_mq.h"
+#include "storage/shm_toc.h"
#include "storage/spin.h"
@@ -60,6 +63,9 @@ typedef struct LogicalRepWorker
*/
FileSet *stream_fileset;
+ /* Indicates if this slot is used for an apply background worker. */
+ bool subworker;
+
/* Stats. */
XLogRecPtr last_lsn;
TimestampTz last_send_time;
@@ -68,9 +74,70 @@ typedef struct LogicalRepWorker
TimestampTz reply_time;
} LogicalRepWorker;
+/* Struct for saving and restoring apply errcontext information */
+typedef struct ApplyErrorCallbackArg
+{
+ LogicalRepMsgType command; /* 0 if invalid */
+ LogicalRepRelMapEntry *rel;
+
+ /* Remote node information */
+ int remote_attnum; /* -1 if invalid */
+ TransactionId remote_xid;
+ XLogRecPtr finish_lsn;
+ char *origin_name;
+} ApplyErrorCallbackArg;
+
+/*
+ * Status for apply background worker.
+ */
+typedef enum ApplyBgworkerStatus
+{
+ APPLY_BGWORKER_ATTACHED = 0,
+ APPLY_BGWORKER_READY,
+ APPLY_BGWORKER_BUSY,
+ APPLY_BGWORKER_FINISHED,
+ APPLY_BGWORKER_EXIT
+} ApplyBgworkerStatus;
+
+/*
+ * Shared information among apply workers.
+ */
+typedef struct ApplyBgworkerShared
+{
+ slock_t mutex;
+
+ /* Status for apply background worker. */
+ ApplyBgworkerStatus status;
+
+ /* server version of publisher. */
+ int server_version;
+
+ TransactionId stream_xid;
+ uint32 n; /* id of apply background worker */
+} ApplyBgworkerShared;
+
+/*
+ * Struct for maintaining an apply background worker.
+ */
+typedef struct ApplyBgworkerState
+{
+ shm_mq_handle *mq_handle;
+ dsm_segment *dsm_seg;
+ ApplyBgworkerShared volatile *pstate;
+} ApplyBgworkerState;
+
/* Main memory context for apply worker. Permanent during worker lifetime. */
extern PGDLLIMPORT MemoryContext ApplyContext;
+extern PGDLLIMPORT MemoryContext ApplyMessageContext;
+
+extern PGDLLIMPORT ApplyErrorCallbackArg apply_error_callback_arg;
+
+extern PGDLLIMPORT bool MySubscriptionValid;
+
+extern PGDLLIMPORT volatile ApplyBgworkerShared *MyParallelState;
+extern PGDLLIMPORT List *subxactlist;
+
/* libpqreceiver connection */
extern PGDLLIMPORT struct WalReceiverConn *LogRepWorkerWalRcvConn;
@@ -79,18 +146,22 @@ extern PGDLLIMPORT Subscription *MySubscription;
extern PGDLLIMPORT LogicalRepWorker *MyLogicalRepWorker;
extern PGDLLIMPORT bool in_remote_transaction;
+extern PGDLLIMPORT bool in_streamed_transaction;
+extern PGDLLIMPORT TransactionId stream_xid;
extern void logicalrep_worker_attach(int slot);
extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
bool only_running);
extern List *logicalrep_workers_find(Oid subid, bool only_running);
-extern void logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
- Oid userid, Oid relid);
+extern bool logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
+ Oid userid, Oid relid,
+ dsm_handle subworker_dsm);
extern void logicalrep_worker_stop(Oid subid, Oid relid);
extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
extern int logicalrep_sync_worker_count(Oid subid);
+extern int logicalrep_apply_background_worker_count(Oid subid);
extern void ReplicationOriginNameForTablesync(Oid suboid, Oid relid,
char *originname, int szorgname);
@@ -103,10 +174,39 @@ extern void process_syncing_tables(XLogRecPtr current_lsn);
extern void invalidate_syncing_table_states(Datum arg, int cacheid,
uint32 hashvalue);
+extern void UpdateWorkerStats(XLogRecPtr last_lsn, TimestampTz send_time,
+ bool reply);
+
+/* prototype needed because of stream_commit */
+extern void apply_dispatch(StringInfo s);
+
+/* Function for apply error callback */
+extern void apply_error_callback(void *arg);
+
+extern void subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue);
+
+/* apply background worker setup and interactions */
+extern ApplyBgworkerState *apply_bgworker_find_or_start(TransactionId xid,
+ bool start);
+extern void apply_bgworker_wait_for(ApplyBgworkerState *wstate,
+ ApplyBgworkerStatus wait_for_status);
+extern void apply_bgworker_send_data(ApplyBgworkerState *wstate, Size nbytes,
+ const void *data);
+extern void apply_bgworker_free(ApplyBgworkerState *wstate);
+extern void apply_bgworker_check_status(void);
+extern void apply_bgworker_set_status(ApplyBgworkerStatus status);
+extern void apply_bgworker_subxact_info_add(TransactionId current_xid);
+
static inline bool
am_tablesync_worker(void)
{
return OidIsValid(MyLogicalRepWorker->relid);
}
+static inline bool
+am_apply_bgworker(void)
+{
+ return MyLogicalRepWorker->subworker;
+}
+
#endif /* WORKER_INTERNAL_H */
diff --git a/src/include/utils/wait_event.h b/src/include/utils/wait_event.h
index b578e2ec75..c2d2a114d7 100644
--- a/src/include/utils/wait_event.h
+++ b/src/include/utils/wait_event.h
@@ -105,6 +105,7 @@ typedef enum
WAIT_EVENT_HASH_GROW_BUCKETS_ALLOCATE,
WAIT_EVENT_HASH_GROW_BUCKETS_ELECT,
WAIT_EVENT_HASH_GROW_BUCKETS_REINSERT,
+ WAIT_EVENT_LOGICAL_APPLY_WORKER_STATE_CHANGE,
WAIT_EVENT_LOGICAL_SYNC_DATA,
WAIT_EVENT_LOGICAL_SYNC_STATE_CHANGE,
WAIT_EVENT_MQ_INTERNAL,
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 7fcfad1591..f769835af0 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -197,7 +197,7 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
-- fail - streaming must be boolean
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, streaming = foo);
-ERROR: streaming requires a Boolean value
+ERROR: streaming requires a Boolean value or "apply"
-- now it works
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, streaming = true);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 4fb746930a..664d4b8b1a 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -116,6 +116,10 @@ Append
AppendPath
AppendRelInfo
AppendState
+ApplyBgworkerEntry
+ApplyBgworkerShared
+ApplyBgworkerState
+ApplyBgworkerStatus
ApplyErrorCallbackArg
ApplyExecutionData
ApplySubXactData
@@ -1483,6 +1487,7 @@ LogicalRepRelId
LogicalRepRelMapEntry
LogicalRepRelation
LogicalRepRollbackPreparedTxnData
+LogicalRepStreamAbortData
LogicalRepTupleData
LogicalRepTyp
LogicalRepWorker
--
2.23.0.windows.1
[application/octet-stream] v10-0002-Test-streaming-apply-option-in-tap-test.patch (68.9K, ../OS3PR01MB62756BD9482EB6BB1CA4CD4D9EAA9@OS3PR01MB6275.jpnprd01.prod.outlook.com/3-v10-0002-Test-streaming-apply-option-in-tap-test.patch)
download | inline diff:
From feac9609b26da1deaef14898ac364dd91d18a99e Mon Sep 17 00:00:00 2001
From: "shiy.fnst" <[email protected]>
Date: Fri, 13 May 2022 14:50:30 +0800
Subject: [PATCH v10 2/5] Test streaming apply option in tap test
Change all TAP tests using the SUBSCRIPTION "streaming" option, so they
now test both 'on' and 'apply' values.
---
src/test/subscription/t/015_stream.pl | 199 ++++---
src/test/subscription/t/016_stream_subxact.pl | 119 +++--
src/test/subscription/t/017_stream_ddl.pl | 188 ++++---
.../t/018_stream_subxact_abort.pl | 195 ++++---
.../t/019_stream_subxact_ddl_abort.pl | 110 +++-
.../subscription/t/022_twophase_cascade.pl | 363 +++++++------
.../subscription/t/023_twophase_stream.pl | 498 ++++++++++--------
7 files changed, 1035 insertions(+), 637 deletions(-)
diff --git a/src/test/subscription/t/015_stream.pl b/src/test/subscription/t/015_stream.pl
index 6561b189de..b3d84b1c7c 100644
--- a/src/test/subscription/t/015_stream.pl
+++ b/src/test/subscription/t/015_stream.pl
@@ -8,6 +8,116 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+# Encapsulate all the common test steps which are related to "streaming" option
+# so the same code can be run both for the streaming=on and streaming=apply
+# cases.
+sub test_streaming
+{
+ my ($node_publisher, $node_subscriber, $appname, $is_apply) = @_;
+
+ # Interleave a pair of transactions, each exceeding the 64kB limit.
+ my $in = '';
+ my $out = '';
+
+ my $timer = IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default);
+
+ my $h = $node_publisher->background_psql('postgres', \$in, \$out, $timer,
+ on_error_stop => 0);
+
+ # Check that a background worker starts if "streaming" option is
+ # specified as "apply". We have to look for the DEBUG1 log messages
+ # about that, so temporarily bump up the log verbosity.
+ if ($is_apply)
+ {
+ $node_subscriber->append_conf('postgresql.conf',
+ "log_min_messages = debug1");
+ $node_subscriber->reload;
+ }
+
+ $in .= q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ };
+ $h->pump_nb;
+
+ if ($is_apply)
+ {
+ $node_subscriber->wait_for_log(qr/\[Apply BGW #\d+\] started/, 0);
+ $node_subscriber->append_conf('postgresql.conf',
+ "log_min_messages = warning");
+ $node_subscriber->reload;
+ }
+
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(5001, 9999) s(i);
+ DELETE FROM test_tab WHERE a > 5000;
+ COMMIT;
+ });
+
+ $in .= q{
+ COMMIT;
+ \q
+ };
+ $h->finish; # errors make the next test fail, so ignore them here
+
+ $node_publisher->wait_for_catchup($appname);
+
+ my $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(3334|3334|3334),
+ 'check extra columns contain local defaults');
+
+ # Test the streaming in binary mode
+ $node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub SET (binary = on)");
+
+ # Insert, update and delete enough rows to exceed the 64kB limit.
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(5001, 10000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ COMMIT;
+ });
+
+ $node_publisher->wait_for_catchup($appname);
+
+ $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(6667|6667|6667),
+ 'check extra columns contain local defaults');
+
+ # Change the local values of the extra columns on the subscriber,
+ # update publisher, and check that subscriber retains the expected
+ # values. This is to ensure that non-streaming transactions behave
+ # properly after a streaming transaction.
+ $node_subscriber->safe_psql('postgres',
+ "UPDATE test_tab SET c = 'epoch'::timestamptz + 987654321 * interval '1s'"
+ );
+ $node_publisher->safe_psql('postgres',
+ "UPDATE test_tab SET b = md5(a::text)");
+
+ $node_publisher->wait_for_catchup($appname);
+
+ $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(extract(epoch from c) = 987654321), count(d = 999) FROM test_tab"
+ );
+ is($result, qq(6667|6667|6667),
+ 'check extra columns contain locally changed data');
+
+ # Cleanup the test data
+ $node_publisher->safe_psql('postgres',
+ "DELETE FROM test_tab WHERE (a > 2)");
+ $node_publisher->wait_for_catchup($appname);
+}
+
# Create publisher node
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
@@ -37,6 +147,10 @@ $node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tap_pub FOR TABLE test_tab");
my $appname = 'tap_sub';
+
+################################
+# Test using streaming mode 'on'
+################################
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
@@ -54,82 +168,25 @@ my $result =
"SELECT count(*), count(c), count(d = 999) FROM test_tab");
is($result, qq(2|2|2), 'check initial data was copied to subscriber');
-# Interleave a pair of transactions, each exceeding the 64kB limit.
-my $in = '';
-my $out = '';
-
-my $timer = IPC::Run::timeout($PostgreSQL::Test::Utils::timeout_default);
-
-my $h = $node_publisher->background_psql('postgres', \$in, \$out, $timer,
- on_error_stop => 0);
-
-$in .= q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
-UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
-DELETE FROM test_tab WHERE mod(a,3) = 0;
-};
-$h->pump_nb;
-
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(5001, 9999) s(i);
-DELETE FROM test_tab WHERE a > 5000;
-COMMIT;
-});
-
-$in .= q{
-COMMIT;
-\q
-};
-$h->finish; # errors make the next test fail, so ignore them here
-
-$node_publisher->wait_for_catchup($appname);
-
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(3334|3334|3334), 'check extra columns contain local defaults');
-
-# Test the streaming in binary mode
-$node_subscriber->safe_psql('postgres',
- "ALTER SUBSCRIPTION tap_sub SET (binary = on)");
-
-# Insert, update and delete enough rows to exceed the 64kB limit.
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(5001, 10000) s(i);
-UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
-DELETE FROM test_tab WHERE mod(a,3) = 0;
-COMMIT;
-});
-
-$node_publisher->wait_for_catchup($appname);
+test_streaming($node_publisher, $node_subscriber, $appname, 0);
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(6667|6667|6667), 'check extra columns contain local defaults');
+###################################
+# Test using streaming mode 'apply'
+###################################
+my $oldpid = $node_publisher->safe_psql('postgres',
+ "SELECT pid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
+);
-# Change the local values of the extra columns on the subscriber,
-# update publisher, and check that subscriber retains the expected
-# values. This is to ensure that non-streaming transactions behave
-# properly after a streaming transaction.
$node_subscriber->safe_psql('postgres',
- "UPDATE test_tab SET c = 'epoch'::timestamptz + 987654321 * interval '1s'"
-);
-$node_publisher->safe_psql('postgres',
- "UPDATE test_tab SET b = md5(a::text)");
+ "ALTER SUBSCRIPTION tap_sub SET(streaming = apply, binary = off)");
-$node_publisher->wait_for_catchup($appname);
+$node_publisher->poll_query_until('postgres',
+ "SELECT pid != $oldpid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
+ )
+ or die
+ "Timed out while waiting for apply to restart after changing SUBSCRIPTION";
-$result = $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(extract(epoch from c) = 987654321), count(d = 999) FROM test_tab"
-);
-is($result, qq(6667|6667|6667),
- 'check extra columns contain locally changed data');
+test_streaming($node_publisher, $node_subscriber, $appname, 1);
$node_subscriber->stop;
$node_publisher->stop;
diff --git a/src/test/subscription/t/016_stream_subxact.pl b/src/test/subscription/t/016_stream_subxact.pl
index f27f1694f2..b5b6dc379f 100644
--- a/src/test/subscription/t/016_stream_subxact.pl
+++ b/src/test/subscription/t/016_stream_subxact.pl
@@ -8,6 +8,72 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+# Encapsulate all the common test steps which are related to "streaming" option
+# so the same code can be run both for the streaming=on and streaming=apply
+# cases.
+sub test_streaming
+{
+ my ($node_publisher, $node_subscriber, $appname, $is_apply) = @_;
+
+ # Check that a background worker starts if "streaming" option is
+ # specified as "apply". We have to look for the DEBUG1 log messages
+ # about that, so temporarily bump up the log verbosity.
+ if ($is_apply)
+ {
+ $node_subscriber->append_conf('postgresql.conf',
+ "log_min_messages = debug1");
+ $node_subscriber->reload;
+ }
+
+ # Insert, update and delete enough rows to exceed 64kB limit.
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series( 3, 500) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ SAVEPOINT s1;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(501, 1000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ SAVEPOINT s2;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1001, 1500) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ SAVEPOINT s3;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1501, 2000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ SAVEPOINT s4;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(2001, 2500) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ COMMIT;
+ });
+
+ if ($is_apply)
+ {
+ $node_subscriber->wait_for_log(qr/\[Apply BGW #\d+\] started/, 0);
+ $node_subscriber->append_conf('postgresql.conf',
+ "log_min_messages = warning");
+ $node_subscriber->reload;
+ }
+
+ $node_publisher->wait_for_catchup($appname);
+
+ my $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(1667|1667|1667),
+ 'check data was copied to subscriber in streaming mode and extra columns contain local defaults'
+ );
+
+ # Cleanup the test data
+ $node_publisher->safe_psql('postgres',
+ "DELETE FROM test_tab WHERE (a > 2)");
+ $node_publisher->wait_for_catchup($appname);
+}
+
# Create publisher node
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
@@ -37,6 +103,10 @@ $node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tap_pub FOR TABLE test_tab");
my $appname = 'tap_sub';
+
+################################
+# Test using streaming mode 'on'
+################################
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
@@ -54,41 +124,26 @@ my $result =
"SELECT count(*), count(c), count(d = 999) FROM test_tab");
is($result, qq(2|2|2), 'check initial data was copied to subscriber');
-# Insert, update and delete enough rows to exceed 64kB limit.
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series( 3, 500) s(i);
-UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
-DELETE FROM test_tab WHERE mod(a,3) = 0;
-SAVEPOINT s1;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(501, 1000) s(i);
-UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
-DELETE FROM test_tab WHERE mod(a,3) = 0;
-SAVEPOINT s2;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1001, 1500) s(i);
-UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
-DELETE FROM test_tab WHERE mod(a,3) = 0;
-SAVEPOINT s3;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1501, 2000) s(i);
-UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
-DELETE FROM test_tab WHERE mod(a,3) = 0;
-SAVEPOINT s4;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(2001, 2500) s(i);
-UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
-DELETE FROM test_tab WHERE mod(a,3) = 0;
-COMMIT;
-});
+test_streaming($node_publisher, $node_subscriber, $appname, 0);
-$node_publisher->wait_for_catchup($appname);
-
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(1667|1667|1667),
- 'check data was copied to subscriber in streaming mode and extra columns contain local defaults'
+###################################
+# Test using streaming mode 'apply'
+###################################
+my $oldpid = $node_publisher->safe_psql('postgres',
+ "SELECT pid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
);
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub SET(streaming = apply)");
+
+$node_publisher->poll_query_until('postgres',
+ "SELECT pid != $oldpid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
+ )
+ or die
+ "Timed out while waiting for apply to restart after changing SUBSCRIPTION";
+
+test_streaming($node_publisher, $node_subscriber, $appname, 1);
+
$node_subscriber->stop;
$node_publisher->stop;
diff --git a/src/test/subscription/t/017_stream_ddl.pl b/src/test/subscription/t/017_stream_ddl.pl
index 0bce63b716..7e5dc18cd2 100644
--- a/src/test/subscription/t/017_stream_ddl.pl
+++ b/src/test/subscription/t/017_stream_ddl.pl
@@ -8,6 +8,111 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+# Encapsulate all the common test steps which are related to "streaming" option
+# so the same code can be run both for the streaming=on and streaming=apply
+# cases.
+sub test_streaming
+{
+ my ($node_publisher, $node_subscriber, $appname, $is_apply) = @_;
+
+ # a small (non-streamed) transaction with DDL and DML
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab VALUES (3, md5(3::text));
+ ALTER TABLE test_tab ADD COLUMN c INT;
+ SAVEPOINT s1;
+ INSERT INTO test_tab VALUES (4, md5(4::text), -4);
+ COMMIT;
+ });
+
+ # Check that a background worker starts if "streaming" option is
+ # specified as "apply". We have to look for the DEBUG1 log messages
+ # about that, so temporarily bump up the log verbosity.
+ if ($is_apply)
+ {
+ $node_subscriber->append_conf('postgresql.conf',
+ "log_min_messages = debug1");
+ $node_subscriber->reload;
+ }
+
+ # large (streamed) transaction with DDL and DML
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text), -i FROM generate_series(5, 1000) s(i);
+ ALTER TABLE test_tab ADD COLUMN d INT;
+ SAVEPOINT s1;
+ INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i FROM generate_series(1001, 2000) s(i);
+ COMMIT;
+ });
+
+ if ($is_apply)
+ {
+ $node_subscriber->wait_for_log(qr/\[Apply BGW #\d+\] started/, 0);
+ $node_subscriber->append_conf('postgresql.conf',
+ "log_min_messages = warning");
+ $node_subscriber->reload;
+ }
+
+ # a small (non-streamed) transaction with DDL and DML
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab VALUES (2001, md5(2001::text), -2001, 2*2001);
+ ALTER TABLE test_tab ADD COLUMN e INT;
+ SAVEPOINT s1;
+ INSERT INTO test_tab VALUES (2002, md5(2002::text), -2002, 2*2002, -3*2002);
+ COMMIT;
+ });
+
+ $node_publisher->wait_for_catchup($appname);
+
+ my $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d), count(e) FROM test_tab");
+ is($result, qq(2002|1999|1002|1),
+ 'check data was copied to subscriber in streaming mode and extra columns contain local defaults'
+ );
+
+ # A large (streamed) transaction with DDL and DML. One of the DDL is performed
+ # after DML to ensure that we invalidate the schema sent for test_tab so that
+ # the next transaction has to send the schema again.
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i, -3*i FROM generate_series(2003,5000) s(i);
+ ALTER TABLE test_tab ADD COLUMN f INT;
+ COMMIT;
+ });
+
+ # A small transaction that won't get streamed. This is just to ensure that we
+ # send the schema again to reflect the last column added in the previous test.
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i, -3*i, 4*i FROM generate_series(5001,5005) s(i);
+ COMMIT;
+ });
+
+ $node_publisher->wait_for_catchup($appname);
+
+ $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d), count(e), count(f) FROM test_tab"
+ );
+ is($result, qq(5005|5002|4005|3004|5),
+ 'check data was copied to subscriber for both streaming and non-streaming transactions'
+ );
+
+ # Cleanup the test data
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ DELETE FROM test_tab WHERE (a > 2);
+ ALTER TABLE test_tab DROP COLUMN c, DROP COLUMN d, DROP COLUMN e, DROP COLUMN f;
+ });
+ $node_publisher->wait_for_catchup($appname);
+}
+
# Create publisher node
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
@@ -37,6 +142,10 @@ $node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tap_pub FOR TABLE test_tab");
my $appname = 'tap_sub';
+
+################################
+# Test using streaming mode 'on'
+################################
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
@@ -54,76 +163,25 @@ my $result =
"SELECT count(*), count(c), count(d = 999) FROM test_tab");
is($result, qq(2|0|0), 'check initial data was copied to subscriber');
-# a small (non-streamed) transaction with DDL and DML
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab VALUES (3, md5(3::text));
-ALTER TABLE test_tab ADD COLUMN c INT;
-SAVEPOINT s1;
-INSERT INTO test_tab VALUES (4, md5(4::text), -4);
-COMMIT;
-});
-
-# large (streamed) transaction with DDL and DML
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text), -i FROM generate_series(5, 1000) s(i);
-ALTER TABLE test_tab ADD COLUMN d INT;
-SAVEPOINT s1;
-INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i FROM generate_series(1001, 2000) s(i);
-COMMIT;
-});
-
-# a small (non-streamed) transaction with DDL and DML
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab VALUES (2001, md5(2001::text), -2001, 2*2001);
-ALTER TABLE test_tab ADD COLUMN e INT;
-SAVEPOINT s1;
-INSERT INTO test_tab VALUES (2002, md5(2002::text), -2002, 2*2002, -3*2002);
-COMMIT;
-});
+test_streaming($node_publisher, $node_subscriber, $appname, 0);
-$node_publisher->wait_for_catchup($appname);
-
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d), count(e) FROM test_tab");
-is($result, qq(2002|1999|1002|1),
- 'check data was copied to subscriber in streaming mode and extra columns contain local defaults'
+###################################
+# Test using streaming mode 'apply'
+###################################
+my $oldpid = $node_publisher->safe_psql('postgres',
+ "SELECT pid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
);
-# A large (streamed) transaction with DDL and DML. One of the DDL is performed
-# after DML to ensure that we invalidate the schema sent for test_tab so that
-# the next transaction has to send the schema again.
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i, -3*i FROM generate_series(2003,5000) s(i);
-ALTER TABLE test_tab ADD COLUMN f INT;
-COMMIT;
-});
-
-# A small transaction that won't get streamed. This is just to ensure that we
-# send the schema again to reflect the last column added in the previous test.
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i, -3*i, 4*i FROM generate_series(5001,5005) s(i);
-COMMIT;
-});
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub SET(streaming = apply)");
-$node_publisher->wait_for_catchup($appname);
+$node_publisher->poll_query_until('postgres',
+ "SELECT pid != $oldpid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
+ )
+ or die
+ "Timed out while waiting for apply to restart after changing SUBSCRIPTION";
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d), count(e), count(f) FROM test_tab");
-is($result, qq(5005|5002|4005|3004|5),
- 'check data was copied to subscriber for both streaming and non-streaming transactions'
-);
+test_streaming($node_publisher, $node_subscriber, $appname, 1);
$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 7155442e76..71f8c1dd05 100644
--- a/src/test/subscription/t/018_stream_subxact_abort.pl
+++ b/src/test/subscription/t/018_stream_subxact_abort.pl
@@ -8,6 +8,113 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+# Encapsulate all the common test steps which are related to "streaming" option
+# so the same code can be run both for the streaming=on and streaming=apply
+# cases.
+sub test_streaming
+{
+ my ($node_publisher, $node_subscriber, $appname, $is_apply) = @_;
+
+ # Check that a background worker starts if "streaming" option is
+ # specified as "apply". We have to look for the DEBUG1 log messages
+ # about that, so temporarily bump up the log verbosity.
+ if ($is_apply)
+ {
+ $node_subscriber->append_conf('postgresql.conf',
+ "log_min_messages = debug1");
+ $node_subscriber->reload;
+ }
+
+ # large (streamed) transaction with DDL, DML and ROLLBACKs
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3,500) s(i);
+ SAVEPOINT s1;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(501,1000) s(i);
+ SAVEPOINT s2;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1001,1500) s(i);
+ SAVEPOINT s3;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1501,2000) s(i);
+ ROLLBACK TO s2;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(2001,2500) s(i);
+ ROLLBACK TO s1;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(2501,3000) s(i);
+ SAVEPOINT s4;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3001,3500) s(i);
+ SAVEPOINT s5;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3501,4000) s(i);
+ COMMIT;
+ });
+
+ if ($is_apply)
+ {
+ $node_subscriber->wait_for_log(qr/\[Apply BGW #\d+\] started/, 0);
+ $node_subscriber->append_conf('postgresql.conf',
+ "log_min_messages = warning");
+ $node_subscriber->reload;
+ }
+
+ $node_publisher->wait_for_catchup($appname);
+
+ my $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c) FROM test_tab");
+ is($result, qq(2000|0),
+ 'check rollback to savepoint was reflected on subscriber and extra columns contain local defaults'
+ );
+
+ # large (streamed) transaction with subscriber receiving out of order
+ # subtransaction ROLLBACKs
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(4001,4500) s(i);
+ SAVEPOINT s1;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(5001,5500) s(i);
+ SAVEPOINT s2;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(6001,6500) s(i);
+ SAVEPOINT s3;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(7001,7500) s(i);
+ RELEASE s2;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(8001,8500) s(i);
+ ROLLBACK TO s1;
+ COMMIT;
+ });
+
+ $node_publisher->wait_for_catchup($appname);
+
+ $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c) FROM test_tab");
+ is($result, qq(2500|0),
+ 'check rollback to savepoint was reflected on subscriber');
+
+ # large (streamed) transaction with subscriber receiving rollback
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(8501,9000) s(i);
+ SAVEPOINT s1;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(9001,9500) s(i);
+ SAVEPOINT s2;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(9501,10000) s(i);
+ ROLLBACK;
+ });
+
+ $node_publisher->wait_for_catchup($appname);
+
+ $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c) FROM test_tab");
+ is($result, qq(2500|0), 'check rollback was reflected on subscriber');
+
+ # Cleanup the test data
+ $node_publisher->safe_psql('postgres',
+ "DELETE FROM test_tab WHERE (a > 2)");
+ $node_publisher->wait_for_catchup($appname);
+}
+
# Create publisher node
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
@@ -36,6 +143,10 @@ $node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tap_pub FOR TABLE test_tab");
my $appname = 'tap_sub';
+
+################################
+# Test using streaming mode 'on'
+################################
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
@@ -53,81 +164,25 @@ my $result =
"SELECT count(*), count(c) FROM test_tab");
is($result, qq(2|0), 'check initial data was copied to subscriber');
-# large (streamed) transaction with DDL, DML and ROLLBACKs
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3,500) s(i);
-SAVEPOINT s1;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(501,1000) s(i);
-SAVEPOINT s2;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1001,1500) s(i);
-SAVEPOINT s3;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1501,2000) s(i);
-ROLLBACK TO s2;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(2001,2500) s(i);
-ROLLBACK TO s1;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(2501,3000) s(i);
-SAVEPOINT s4;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3001,3500) s(i);
-SAVEPOINT s5;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3501,4000) s(i);
-COMMIT;
-});
-
-$node_publisher->wait_for_catchup($appname);
+test_streaming($node_publisher, $node_subscriber, $appname, 0);
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c) FROM test_tab");
-is($result, qq(2000|0),
- 'check rollback to savepoint was reflected on subscriber and extra columns contain local defaults'
+###################################
+# Test using streaming mode 'apply'
+###################################
+my $oldpid = $node_publisher->safe_psql('postgres',
+ "SELECT pid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
);
-# large (streamed) transaction with subscriber receiving out of order
-# subtransaction ROLLBACKs
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(4001,4500) s(i);
-SAVEPOINT s1;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(5001,5500) s(i);
-SAVEPOINT s2;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(6001,6500) s(i);
-SAVEPOINT s3;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(7001,7500) s(i);
-RELEASE s2;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(8001,8500) s(i);
-ROLLBACK TO s1;
-COMMIT;
-});
-
-$node_publisher->wait_for_catchup($appname);
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub SET(streaming = apply)");
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c) FROM test_tab");
-is($result, qq(2500|0),
- 'check rollback to savepoint was reflected on subscriber');
-
-# large (streamed) transaction with subscriber receiving rollback
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(8501,9000) s(i);
-SAVEPOINT s1;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(9001,9500) s(i);
-SAVEPOINT s2;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(9501,10000) s(i);
-ROLLBACK;
-});
+$node_publisher->poll_query_until('postgres',
+ "SELECT pid != $oldpid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
+ )
+ or die
+ "Timed out while waiting for apply to restart after changing SUBSCRIPTION";
-$node_publisher->wait_for_catchup($appname);
-
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c) FROM test_tab");
-is($result, qq(2500|0), 'check rollback was reflected on subscriber');
+test_streaming($node_publisher, $node_subscriber, $appname, 1);
$node_subscriber->stop;
$node_publisher->stop;
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 dbd0fca4d1..d0cd869430 100644
--- a/src/test/subscription/t/019_stream_subxact_ddl_abort.pl
+++ b/src/test/subscription/t/019_stream_subxact_ddl_abort.pl
@@ -9,6 +9,69 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+# Encapsulate all the common test steps which are related to "streaming" option
+# so the same code can be run both for the streaming=on and streaming=apply
+# cases.
+sub test_streaming
+{
+ my ($node_publisher, $node_subscriber, $appname, $is_apply) = @_;
+
+ # Check that a background worker starts if "streaming" option is
+ # specified as "apply". We have to look for the DEBUG1 log messages
+ # about that, so temporarily bump up the log verbosity.
+ if ($is_apply)
+ {
+ $node_subscriber->append_conf('postgresql.conf',
+ "log_min_messages = debug1");
+ $node_subscriber->reload;
+ }
+
+ # large (streamed) transaction with DDL, DML and ROLLBACKs
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3,500) s(i);
+ ALTER TABLE test_tab ADD COLUMN c INT;
+ SAVEPOINT s1;
+ INSERT INTO test_tab SELECT i, md5(i::text), -i FROM generate_series(501,1000) s(i);
+ ALTER TABLE test_tab ADD COLUMN d INT;
+ SAVEPOINT s2;
+ INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i FROM generate_series(1001,1500) s(i);
+ ALTER TABLE test_tab ADD COLUMN e INT;
+ SAVEPOINT s3;
+ INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i, -3*i FROM generate_series(1501,2000) s(i);
+ ALTER TABLE test_tab DROP COLUMN c;
+ ROLLBACK TO s1;
+ INSERT INTO test_tab SELECT i, md5(i::text), i FROM generate_series(501,1000) s(i);
+ COMMIT;
+ });
+
+ if ($is_apply)
+ {
+ $node_subscriber->wait_for_log(qr/\[Apply BGW #\d+\] started/, 0);
+ $node_subscriber->append_conf('postgresql.conf',
+ "log_min_messages = warning");
+ $node_subscriber->reload;
+ }
+
+ $node_publisher->wait_for_catchup($appname);
+
+ my $result =
+ $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c) FROM test_tab");
+ is($result, qq(1000|500),
+ 'check rollback to savepoint was reflected on subscriber and extra columns contain local defaults'
+ );
+
+ # Cleanup the test data
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ DELETE FROM test_tab WHERE (a > 2);
+ ALTER TABLE test_tab DROP COLUMN c;
+ });
+ $node_publisher->wait_for_catchup($appname);
+}
+
# Create publisher node
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
@@ -37,6 +100,10 @@ $node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tap_pub FOR TABLE test_tab");
my $appname = 'tap_sub';
+
+################################
+# Test using streaming mode 'on'
+################################
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub WITH (streaming = on)"
);
@@ -54,35 +121,26 @@ my $result =
"SELECT count(*), count(c) FROM test_tab");
is($result, qq(2|0), 'check initial data was copied to subscriber');
-# large (streamed) transaction with DDL, DML and ROLLBACKs
-$node_publisher->safe_psql(
- 'postgres', q{
-BEGIN;
-INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3,500) s(i);
-ALTER TABLE test_tab ADD COLUMN c INT;
-SAVEPOINT s1;
-INSERT INTO test_tab SELECT i, md5(i::text), -i FROM generate_series(501,1000) s(i);
-ALTER TABLE test_tab ADD COLUMN d INT;
-SAVEPOINT s2;
-INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i FROM generate_series(1001,1500) s(i);
-ALTER TABLE test_tab ADD COLUMN e INT;
-SAVEPOINT s3;
-INSERT INTO test_tab SELECT i, md5(i::text), -i, 2*i, -3*i FROM generate_series(1501,2000) s(i);
-ALTER TABLE test_tab DROP COLUMN c;
-ROLLBACK TO s1;
-INSERT INTO test_tab SELECT i, md5(i::text), i FROM generate_series(501,1000) s(i);
-COMMIT;
-});
+test_streaming($node_publisher, $node_subscriber, $appname, 0);
-$node_publisher->wait_for_catchup($appname);
-
-$result =
- $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c) FROM test_tab");
-is($result, qq(1000|500),
- 'check rollback to savepoint was reflected on subscriber and extra columns contain local defaults'
+###################################
+# Test using streaming mode 'apply'
+###################################
+my $oldpid = $node_publisher->safe_psql('postgres',
+ "SELECT pid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
);
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub SET(streaming = apply)");
+
+$node_publisher->poll_query_until('postgres',
+ "SELECT pid != $oldpid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
+ )
+ or die
+ "Timed out while waiting for apply to restart after changing SUBSCRIPTION";
+
+test_streaming($node_publisher, $node_subscriber, $appname, 1);
+
$node_subscriber->stop;
$node_publisher->stop;
diff --git a/src/test/subscription/t/022_twophase_cascade.pl b/src/test/subscription/t/022_twophase_cascade.pl
index 7a797f37ba..b52af74e35 100644
--- a/src/test/subscription/t/022_twophase_cascade.pl
+++ b/src/test/subscription/t/022_twophase_cascade.pl
@@ -11,6 +11,208 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+# Encapsulate all the common test steps which are related to "streaming" option
+# so the same code can be run both for the streaming=on and streaming=apply
+# cases.
+sub test_streaming
+{
+ my ($node_A, $node_B, $node_C, $appname_B, $appname_C, $streaming_mode) =
+ @_;
+
+ my $oldpid_B = $node_A->safe_psql(
+ 'postgres', "
+ SELECT pid FROM pg_stat_replication
+ WHERE application_name = '$appname_B' AND state = 'streaming';");
+ my $oldpid_C = $node_B->safe_psql(
+ 'postgres', "
+ SELECT pid FROM pg_stat_replication
+ WHERE application_name = '$appname_C' AND state = 'streaming';");
+
+ # Setup logical replication streaming mode
+
+ $node_B->safe_psql(
+ 'postgres', "
+ ALTER SUBSCRIPTION tap_sub_B
+ SET (streaming = $streaming_mode);");
+ $node_C->safe_psql(
+ 'postgres', "
+ ALTER SUBSCRIPTION tap_sub_C
+ SET (streaming = $streaming_mode)");
+
+ # Wait for subscribers to finish initialization
+
+ $node_A->poll_query_until(
+ 'postgres', "
+ SELECT pid != $oldpid_B FROM pg_stat_replication
+ WHERE application_name = '$appname_B' AND state = 'streaming';"
+ ) or die "Timed out while waiting for apply to restart";
+ $node_B->poll_query_until(
+ 'postgres', "
+ SELECT pid != $oldpid_C FROM pg_stat_replication
+ WHERE application_name = '$appname_C' AND state = 'streaming';"
+ ) or die "Timed out while waiting for apply to restart";
+
+ ###############################
+ # Test 2PC PREPARE / COMMIT PREPARED.
+ # 1. Data is streamed as a 2PC transaction.
+ # 2. Then do commit prepared.
+ #
+ # Expect all data is replicated on subscriber(s) after the commit.
+ ###############################
+
+ # Check that a background worker starts if "streaming" option is
+ # specified as "apply". We have to look for the DEBUG1 log messages
+ # about that, so temporarily bump up the log verbosity.
+ if ($streaming_mode eq 'apply')
+ {
+ $node_B->append_conf('postgresql.conf', "log_min_messages = debug1");
+ $node_B->reload;
+
+ $node_C->append_conf('postgresql.conf', "log_min_messages = debug1");
+ $node_C->reload;
+ }
+
+ # Insert, update and delete enough rows to exceed the 64kB limit.
+ # Then 2PC PREPARE
+ $node_A->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ PREPARE TRANSACTION 'test_prepared_tab';});
+
+ if ($streaming_mode eq 'apply')
+ {
+ $node_B->wait_for_log(qr/\[Apply BGW #\d+\] started/, 0);
+ $node_B->append_conf('postgresql.conf', "log_min_messages = warning");
+ $node_B->reload;
+
+ $node_C->wait_for_log(qr/\[Apply BGW #\d+\] started/, 0);
+ $node_C->append_conf('postgresql.conf', "log_min_messages = warning");
+ $node_C->reload;
+ }
+
+ $node_A->wait_for_catchup($appname_B);
+ $node_B->wait_for_catchup($appname_C);
+
+ # check the transaction state is prepared on subscriber(s)
+ my $result =
+ $node_B->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(1), 'transaction is prepared on subscriber B');
+ $result =
+ $node_C->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(1), 'transaction is prepared on subscriber C');
+
+ # 2PC COMMIT
+ $node_A->safe_psql('postgres', "COMMIT PREPARED 'test_prepared_tab';");
+
+ $node_A->wait_for_catchup($appname_B);
+ $node_B->wait_for_catchup($appname_C);
+
+ # check that transaction was committed on subscriber(s)
+ $result = $node_B->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(3334|3334|3334),
+ 'Rows inserted by 2PC have committed on subscriber B, and extra columns have local defaults'
+ );
+ $result = $node_C->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(3334|3334|3334),
+ 'Rows inserted by 2PC have committed on subscriber C, and extra columns have local defaults'
+ );
+
+ # check the transaction state is ended on subscriber(s)
+ $result =
+ $node_B->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is committed on subscriber B');
+ $result =
+ $node_C->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is committed on subscriber C');
+
+ ###############################
+ # Test 2PC PREPARE with a nested ROLLBACK TO SAVEPOINT.
+ # 0. Cleanup from previous test leaving only 2 rows.
+ # 1. Insert one more row.
+ # 2. Record a SAVEPOINT.
+ # 3. Data is streamed using 2PC.
+ # 4. Do rollback to SAVEPOINT prior to the streamed inserts.
+ # 5. Then COMMIT PREPARED.
+ #
+ # Expect data after the SAVEPOINT is aborted leaving only 3 rows (= 2 original + 1 from step 1).
+ ###############################
+
+ # First, delete the data except for 2 rows (delete will be replicated)
+ $node_A->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
+
+ # 2PC PREPARE with a nested ROLLBACK TO SAVEPOINT
+ $node_A->safe_psql(
+ 'postgres', "
+ BEGIN;
+ INSERT INTO test_tab VALUES (9999, 'foobar');
+ SAVEPOINT sp_inner;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ ROLLBACK TO SAVEPOINT sp_inner;
+ PREPARE TRANSACTION 'outer';
+ ");
+
+ $node_A->wait_for_catchup($appname_B);
+ $node_B->wait_for_catchup($appname_C);
+
+ # check the transaction state prepared on subscriber(s)
+ $result =
+ $node_B->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(1), 'transaction is prepared on subscriber B');
+ $result =
+ $node_C->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(1), 'transaction is prepared on subscriber C');
+
+ # 2PC COMMIT
+ $node_A->safe_psql('postgres', "COMMIT PREPARED 'outer';");
+
+ $node_A->wait_for_catchup($appname_B);
+ $node_B->wait_for_catchup($appname_C);
+
+ # check the transaction state is ended on subscriber
+ $result =
+ $node_B->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is ended on subscriber B');
+ $result =
+ $node_C->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is ended on subscriber C');
+
+ # check inserts are visible at subscriber(s).
+ # All the streamed data (prior to the SAVEPOINT) should be rolled back.
+ # (9999, 'foobar') should be committed.
+ $result = $node_B->safe_psql('postgres',
+ "SELECT count(*) FROM test_tab where b = 'foobar';");
+ is($result, qq(1), 'Rows committed are present on subscriber B');
+ $result =
+ $node_B->safe_psql('postgres', "SELECT count(*) FROM test_tab;");
+ is($result, qq(3), 'Rows committed are present on subscriber B');
+ $result = $node_C->safe_psql('postgres',
+ "SELECT count(*) FROM test_tab where b = 'foobar';");
+ is($result, qq(1), 'Rows committed are present on subscriber C');
+ $result =
+ $node_C->safe_psql('postgres', "SELECT count(*) FROM test_tab;");
+ is($result, qq(3), 'Rows committed are present on subscriber C');
+
+ # Cleanup the test data
+ $node_A->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
+ $node_A->wait_for_catchup($appname_B);
+ $node_B->wait_for_catchup($appname_C);
+}
+
###############################
# Setup a cascade of pub/sub nodes.
# node_A -> node_B -> node_C
@@ -260,160 +462,15 @@ is($result, qq(21), 'Rows committed are present on subscriber C');
# 2PC + STREAMING TESTS
# ---------------------
-my $oldpid_B = $node_A->safe_psql(
- 'postgres', "
- SELECT pid FROM pg_stat_replication
- WHERE application_name = '$appname_B' AND state = 'streaming';");
-my $oldpid_C = $node_B->safe_psql(
- 'postgres', "
- SELECT pid FROM pg_stat_replication
- WHERE application_name = '$appname_C' AND state = 'streaming';");
-
-# Setup logical replication (streaming = on)
-
-$node_B->safe_psql(
- 'postgres', "
- ALTER SUBSCRIPTION tap_sub_B
- SET (streaming = on);");
-$node_C->safe_psql(
- 'postgres', "
- ALTER SUBSCRIPTION tap_sub_C
- SET (streaming = on)");
-
-# Wait for subscribers to finish initialization
-
-$node_A->poll_query_until(
- 'postgres', "
- SELECT pid != $oldpid_B FROM pg_stat_replication
- WHERE application_name = '$appname_B' AND state = 'streaming';"
-) or die "Timed out while waiting for apply to restart";
-$node_B->poll_query_until(
- 'postgres', "
- SELECT pid != $oldpid_C FROM pg_stat_replication
- WHERE application_name = '$appname_C' AND state = 'streaming';"
-) or die "Timed out while waiting for apply to restart";
-
-###############################
-# Test 2PC PREPARE / COMMIT PREPARED.
-# 1. Data is streamed as a 2PC transaction.
-# 2. Then do commit prepared.
-#
-# Expect all data is replicated on subscriber(s) after the commit.
-###############################
-
-# Insert, update and delete enough rows to exceed the 64kB limit.
-# Then 2PC PREPARE
-$node_A->safe_psql(
- 'postgres', q{
- BEGIN;
- INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- PREPARE TRANSACTION 'test_prepared_tab';});
-
-$node_A->wait_for_catchup($appname_B);
-$node_B->wait_for_catchup($appname_C);
+################################
+# Test using streaming mode 'on'
+################################
+test_streaming($node_A, $node_B, $node_C, $appname_B, $appname_C, 'on');
-# check the transaction state is prepared on subscriber(s)
-$result =
- $node_B->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(1), 'transaction is prepared on subscriber B');
-$result =
- $node_C->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(1), 'transaction is prepared on subscriber C');
-
-# 2PC COMMIT
-$node_A->safe_psql('postgres', "COMMIT PREPARED 'test_prepared_tab';");
-
-$node_A->wait_for_catchup($appname_B);
-$node_B->wait_for_catchup($appname_C);
-
-# check that transaction was committed on subscriber(s)
-$result = $node_B->safe_psql('postgres',
- "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(3334|3334|3334),
- 'Rows inserted by 2PC have committed on subscriber B, and extra columns have local defaults'
-);
-$result = $node_C->safe_psql('postgres',
- "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(3334|3334|3334),
- 'Rows inserted by 2PC have committed on subscriber C, and extra columns have local defaults'
-);
-
-# check the transaction state is ended on subscriber(s)
-$result =
- $node_B->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is committed on subscriber B');
-$result =
- $node_C->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is committed on subscriber C');
-
-###############################
-# Test 2PC PREPARE with a nested ROLLBACK TO SAVEPOINT.
-# 0. Cleanup from previous test leaving only 2 rows.
-# 1. Insert one more row.
-# 2. Record a SAVEPOINT.
-# 3. Data is streamed using 2PC.
-# 4. Do rollback to SAVEPOINT prior to the streamed inserts.
-# 5. Then COMMIT PREPARED.
-#
-# Expect data after the SAVEPOINT is aborted leaving only 3 rows (= 2 original + 1 from step 1).
-###############################
-
-# First, delete the data except for 2 rows (delete will be replicated)
-$node_A->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
-
-# 2PC PREPARE with a nested ROLLBACK TO SAVEPOINT
-$node_A->safe_psql(
- 'postgres', "
- BEGIN;
- INSERT INTO test_tab VALUES (9999, 'foobar');
- SAVEPOINT sp_inner;
- INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- ROLLBACK TO SAVEPOINT sp_inner;
- PREPARE TRANSACTION 'outer';
- ");
-
-$node_A->wait_for_catchup($appname_B);
-$node_B->wait_for_catchup($appname_C);
-
-# check the transaction state prepared on subscriber(s)
-$result =
- $node_B->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(1), 'transaction is prepared on subscriber B');
-$result =
- $node_C->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(1), 'transaction is prepared on subscriber C');
-
-# 2PC COMMIT
-$node_A->safe_psql('postgres', "COMMIT PREPARED 'outer';");
-
-$node_A->wait_for_catchup($appname_B);
-$node_B->wait_for_catchup($appname_C);
-
-# check the transaction state is ended on subscriber
-$result =
- $node_B->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is ended on subscriber B');
-$result =
- $node_C->safe_psql('postgres', "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is ended on subscriber C');
-
-# check inserts are visible at subscriber(s).
-# All the streamed data (prior to the SAVEPOINT) should be rolled back.
-# (9999, 'foobar') should be committed.
-$result = $node_B->safe_psql('postgres',
- "SELECT count(*) FROM test_tab where b = 'foobar';");
-is($result, qq(1), 'Rows committed are present on subscriber B');
-$result = $node_B->safe_psql('postgres', "SELECT count(*) FROM test_tab;");
-is($result, qq(3), 'Rows committed are present on subscriber B');
-$result = $node_C->safe_psql('postgres',
- "SELECT count(*) FROM test_tab where b = 'foobar';");
-is($result, qq(1), 'Rows committed are present on subscriber C');
-$result = $node_C->safe_psql('postgres', "SELECT count(*) FROM test_tab;");
-is($result, qq(3), 'Rows committed are present on subscriber C');
+###################################
+# Test using streaming mode 'apply'
+###################################
+test_streaming($node_A, $node_B, $node_C, $appname_B, $appname_C, 'apply');
###############################
# check all the cleanup
diff --git a/src/test/subscription/t/023_twophase_stream.pl b/src/test/subscription/t/023_twophase_stream.pl
index d8475d25a4..f55c5c95e7 100644
--- a/src/test/subscription/t/023_twophase_stream.pl
+++ b/src/test/subscription/t/023_twophase_stream.pl
@@ -8,6 +8,266 @@ use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
+# Encapsulate all the common test steps which are related to "streaming" option
+# so the same code can be run both for the streaming=on and streaming=apply
+# cases.
+sub test_streaming
+{
+ my ($node_publisher, $node_subscriber, $appname, $is_apply) = @_;
+
+ ###############################
+ # Test 2PC PREPARE / COMMIT PREPARED.
+ # 1. Data is streamed as a 2PC transaction.
+ # 2. Then do commit prepared.
+ #
+ # Expect all data is replicated on subscriber side after the commit.
+ ###############################
+
+ # Check that a background worker starts if "streaming" option is
+ # specified as "apply". We have to look for the DEBUG1 log messages
+ # about that, so temporarily bump up the log verbosity.
+ if ($is_apply)
+ {
+ $node_subscriber->append_conf('postgresql.conf',
+ "log_min_messages = debug1");
+ $node_subscriber->reload;
+ }
+
+ # check that 2PC gets replicated to subscriber
+ # Insert, update and delete enough rows to exceed the 64kB limit.
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ PREPARE TRANSACTION 'test_prepared_tab';});
+
+ if ($is_apply)
+ {
+ $node_subscriber->wait_for_log(qr/\[Apply BGW #\d+\] started/, 0);
+ $node_subscriber->append_conf('postgresql.conf',
+ "log_min_messages = warning");
+ $node_subscriber->reload;
+ }
+
+ $node_publisher->wait_for_catchup($appname);
+
+ # check that transaction is in prepared state on subscriber
+ my $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(1), 'transaction is prepared on subscriber');
+
+ # 2PC transaction gets committed
+ $node_publisher->safe_psql('postgres',
+ "COMMIT PREPARED 'test_prepared_tab';");
+
+ $node_publisher->wait_for_catchup($appname);
+
+ # check that transaction is committed on subscriber
+ $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(3334|3334|3334),
+ 'Rows inserted by 2PC have committed on subscriber, and extra columns contain local defaults'
+ );
+ $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is committed on subscriber');
+
+ ###############################
+ # Test 2PC PREPARE / ROLLBACK PREPARED.
+ # 1. Table is deleted back to 2 rows which are replicated on subscriber.
+ # 2. Data is streamed using 2PC.
+ # 3. Do rollback prepared.
+ #
+ # Expect data rolls back leaving only the original 2 rows.
+ ###############################
+
+ # First, delete the data except for 2 rows (will be replicated)
+ $node_publisher->safe_psql('postgres',
+ "DELETE FROM test_tab WHERE a > 2;");
+
+ # Then insert, update and delete enough rows to exceed the 64kB limit.
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ PREPARE TRANSACTION 'test_prepared_tab';});
+
+ $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');
+
+ # 2PC transaction gets aborted
+ $node_publisher->safe_psql('postgres',
+ "ROLLBACK PREPARED 'test_prepared_tab';");
+
+ $node_publisher->wait_for_catchup($appname);
+
+ # check that transaction is aborted on subscriber
+ $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(2|2|2),
+ 'Rows inserted by 2PC are rolled back, leaving only the original 2 rows'
+ );
+
+ $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is aborted on subscriber');
+
+ ###############################
+ # Check that 2PC COMMIT PREPARED is decoded properly on crash restart.
+ # 1. insert, update and delete enough rows to exceed the 64kB limit.
+ # 2. Then server crashes before the 2PC transaction is committed.
+ # 3. After servers are restarted the pending transaction is committed.
+ #
+ # Expect all data is replicated on subscriber side after the commit.
+ # Note: both publisher and subscriber do crash/restart.
+ ###############################
+
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ PREPARE TRANSACTION 'test_prepared_tab';});
+
+ $node_subscriber->stop('immediate');
+ $node_publisher->stop('immediate');
+
+ $node_publisher->start;
+ $node_subscriber->start;
+
+ # commit post the restart
+ $node_publisher->safe_psql('postgres',
+ "COMMIT PREPARED 'test_prepared_tab';");
+ $node_publisher->wait_for_catchup($appname);
+
+ # check inserts are visible
+ $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(3334|3334|3334),
+ 'Rows inserted by 2PC have committed on subscriber, and extra columns contain local defaults'
+ );
+
+ ###############################
+ # Do INSERT after the PREPARE but before ROLLBACK PREPARED.
+ # 1. Table is deleted back to 2 rows which are replicated on subscriber.
+ # 2. Data is streamed using 2PC.
+ # 3. A single row INSERT is done which is after the PREPARE.
+ # 4. Then do a ROLLBACK PREPARED.
+ #
+ # Expect the 2PC data rolls back leaving only 3 rows on the subscriber
+ # (the original 2 + inserted 1).
+ ###############################
+
+ # First, delete the data except for 2 rows (will be replicated)
+ $node_publisher->safe_psql('postgres',
+ "DELETE FROM test_tab WHERE a > 2;");
+
+ # Then insert, update and delete enough rows to exceed the 64kB limit.
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ PREPARE TRANSACTION 'test_prepared_tab';});
+
+ $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');
+
+ # Insert a different record (now we are outside of the 2PC transaction)
+ # Note: the 2PC transaction still holds row locks so make sure this insert is for a separate primary key
+ $node_publisher->safe_psql('postgres',
+ "INSERT INTO test_tab VALUES (99999, 'foobar')");
+
+ # 2PC transaction gets aborted
+ $node_publisher->safe_psql('postgres',
+ "ROLLBACK PREPARED 'test_prepared_tab';");
+
+ $node_publisher->wait_for_catchup($appname);
+
+ # check that transaction is aborted on subscriber,
+ # but the extra INSERT outside of the 2PC still was replicated
+ $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(3|3|3),
+ 'check the outside insert was copied to subscriber');
+
+ $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is aborted on subscriber');
+
+ ###############################
+ # Do INSERT after the PREPARE but before COMMIT PREPARED.
+ # 1. Table is deleted back to 2 rows which are replicated on subscriber.
+ # 2. Data is streamed using 2PC.
+ # 3. A single row INSERT is done which is after the PREPARE.
+ # 4. Then do a COMMIT PREPARED.
+ #
+ # Expect 2PC data + the extra row are on the subscriber
+ # (the 3334 + inserted 1 = 3335).
+ ###############################
+
+ # First, delete the data except for 2 rows (will be replicated)
+ $node_publisher->safe_psql('postgres',
+ "DELETE FROM test_tab WHERE a > 2;");
+
+ # Then insert, update and delete enough rows to exceed the 64kB limit.
+ $node_publisher->safe_psql(
+ 'postgres', q{
+ BEGIN;
+ INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
+ UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
+ DELETE FROM test_tab WHERE mod(a,3) = 0;
+ PREPARE TRANSACTION 'test_prepared_tab';});
+
+ $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');
+
+ # Insert a different record (now we are outside of the 2PC transaction)
+ # Note: the 2PC transaction still holds row locks so make sure this insert is for a separare primary key
+ $node_publisher->safe_psql('postgres',
+ "INSERT INTO test_tab VALUES (99999, 'foobar')");
+
+ # 2PC transaction gets committed
+ $node_publisher->safe_psql('postgres',
+ "COMMIT PREPARED 'test_prepared_tab';");
+
+ $node_publisher->wait_for_catchup($appname);
+
+ # check that transaction is committed on subscriber
+ $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*), count(c), count(d = 999) FROM test_tab");
+ is($result, qq(3335|3335|3335),
+ 'Rows inserted by 2PC (as well as outside insert) have committed on subscriber, and extra columns contain local defaults'
+ );
+
+ $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_prepared_xacts;");
+ is($result, qq(0), 'transaction is committed on subscriber');
+
+ # Cleanup the test data
+ $node_publisher->safe_psql('postgres',
+ "DELETE FROM test_tab WHERE a > 2;");
+ $node_publisher->wait_for_catchup($appname);
+}
+
###############################
# Setup
###############################
@@ -48,6 +308,10 @@ $node_publisher->safe_psql('postgres',
"CREATE PUBLICATION tap_pub FOR TABLE test_tab");
my $appname = 'tap_sub';
+
+################################
+# Test using streaming mode 'on'
+################################
$node_subscriber->safe_psql(
'postgres', "
CREATE SUBSCRIPTION tap_sub
@@ -70,236 +334,30 @@ my $twophase_query =
$node_subscriber->poll_query_until('postgres', $twophase_query)
or die "Timed out while waiting for subscriber to enable twophase";
-###############################
# Check initial data was copied to subscriber
-###############################
my $result = $node_subscriber->safe_psql('postgres',
"SELECT count(*), count(c), count(d = 999) FROM test_tab");
is($result, qq(2|2|2), 'check initial data was copied to subscriber');
-###############################
-# Test 2PC PREPARE / COMMIT PREPARED.
-# 1. Data is streamed as a 2PC transaction.
-# 2. Then do commit prepared.
-#
-# Expect all data is replicated on subscriber side after the commit.
-###############################
-
-# check that 2PC gets replicated to subscriber
-# Insert, update and delete enough rows to exceed the 64kB limit.
-$node_publisher->safe_psql(
- 'postgres', q{
- BEGIN;
- INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- PREPARE TRANSACTION 'test_prepared_tab';});
-
-$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');
-
-# 2PC transaction gets committed
-$node_publisher->safe_psql('postgres',
- "COMMIT PREPARED 'test_prepared_tab';");
-
-$node_publisher->wait_for_catchup($appname);
-
-# check that transaction is committed on subscriber
-$result = $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(3334|3334|3334),
- 'Rows inserted by 2PC have committed on subscriber, and extra columns contain local defaults'
-);
-$result = $node_subscriber->safe_psql('postgres',
- "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is committed on subscriber');
-
-###############################
-# Test 2PC PREPARE / ROLLBACK PREPARED.
-# 1. Table is deleted back to 2 rows which are replicated on subscriber.
-# 2. Data is streamed using 2PC.
-# 3. Do rollback prepared.
-#
-# Expect data rolls back leaving only the original 2 rows.
-###############################
-
-# First, delete the data except for 2 rows (will be replicated)
-$node_publisher->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
-
-# Then insert, update and delete enough rows to exceed the 64kB limit.
-$node_publisher->safe_psql(
- 'postgres', q{
- BEGIN;
- INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- PREPARE TRANSACTION 'test_prepared_tab';});
-
-$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');
-
-# 2PC transaction gets aborted
-$node_publisher->safe_psql('postgres',
- "ROLLBACK PREPARED 'test_prepared_tab';");
-
-$node_publisher->wait_for_catchup($appname);
-
-# check that transaction is aborted on subscriber
-$result = $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(2|2|2),
- 'Rows inserted by 2PC are rolled back, leaving only the original 2 rows');
-
-$result = $node_subscriber->safe_psql('postgres',
- "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is aborted on subscriber');
-
-###############################
-# Check that 2PC COMMIT PREPARED is decoded properly on crash restart.
-# 1. insert, update and delete enough rows to exceed the 64kB limit.
-# 2. Then server crashes before the 2PC transaction is committed.
-# 3. After servers are restarted the pending transaction is committed.
-#
-# Expect all data is replicated on subscriber side after the commit.
-# Note: both publisher and subscriber do crash/restart.
-###############################
-
-$node_publisher->safe_psql(
- 'postgres', q{
- BEGIN;
- INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- PREPARE TRANSACTION 'test_prepared_tab';});
-
-$node_subscriber->stop('immediate');
-$node_publisher->stop('immediate');
-
-$node_publisher->start;
-$node_subscriber->start;
-
-# commit post the restart
-$node_publisher->safe_psql('postgres',
- "COMMIT PREPARED 'test_prepared_tab';");
-$node_publisher->wait_for_catchup($appname);
+test_streaming($node_publisher, $node_subscriber, $appname, 0);
-# check inserts are visible
-$result = $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(3334|3334|3334),
- 'Rows inserted by 2PC have committed on subscriber, and extra columns contain local defaults'
+###################################
+# Test using streaming mode 'apply'
+###################################
+my $oldpid = $node_publisher->safe_psql('postgres',
+ "SELECT pid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
);
-###############################
-# Do INSERT after the PREPARE but before ROLLBACK PREPARED.
-# 1. Table is deleted back to 2 rows which are replicated on subscriber.
-# 2. Data is streamed using 2PC.
-# 3. A single row INSERT is done which is after the PREPARE.
-# 4. Then do a ROLLBACK PREPARED.
-#
-# Expect the 2PC data rolls back leaving only 3 rows on the subscriber
-# (the original 2 + inserted 1).
-###############################
-
-# First, delete the data except for 2 rows (will be replicated)
-$node_publisher->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
-
-# Then insert, update and delete enough rows to exceed the 64kB limit.
-$node_publisher->safe_psql(
- 'postgres', q{
- BEGIN;
- INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- PREPARE TRANSACTION 'test_prepared_tab';});
-
-$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');
-
-# Insert a different record (now we are outside of the 2PC transaction)
-# Note: the 2PC transaction still holds row locks so make sure this insert is for a separate primary key
-$node_publisher->safe_psql('postgres',
- "INSERT INTO test_tab VALUES (99999, 'foobar')");
-
-# 2PC transaction gets aborted
-$node_publisher->safe_psql('postgres',
- "ROLLBACK PREPARED 'test_prepared_tab';");
-
-$node_publisher->wait_for_catchup($appname);
-
-# check that transaction is aborted on subscriber,
-# but the extra INSERT outside of the 2PC still was replicated
-$result = $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(3|3|3), 'check the outside insert was copied to subscriber');
-
-$result = $node_subscriber->safe_psql('postgres',
- "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is aborted on subscriber');
-
-###############################
-# Do INSERT after the PREPARE but before COMMIT PREPARED.
-# 1. Table is deleted back to 2 rows which are replicated on subscriber.
-# 2. Data is streamed using 2PC.
-# 3. A single row INSERT is done which is after the PREPARE.
-# 4. Then do a COMMIT PREPARED.
-#
-# Expect 2PC data + the extra row are on the subscriber
-# (the 3334 + inserted 1 = 3335).
-###############################
-
-# First, delete the data except for 2 rows (will be replicated)
-$node_publisher->safe_psql('postgres', "DELETE FROM test_tab WHERE a > 2;");
-
-# Then insert, update and delete enough rows to exceed the 64kB limit.
-$node_publisher->safe_psql(
- 'postgres', q{
- BEGIN;
- INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);
- UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;
- DELETE FROM test_tab WHERE mod(a,3) = 0;
- PREPARE TRANSACTION 'test_prepared_tab';});
-
-$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');
-
-# Insert a different record (now we are outside of the 2PC transaction)
-# Note: the 2PC transaction still holds row locks so make sure this insert is for a separare primary key
-$node_publisher->safe_psql('postgres',
- "INSERT INTO test_tab VALUES (99999, 'foobar')");
-
-# 2PC transaction gets committed
-$node_publisher->safe_psql('postgres',
- "COMMIT PREPARED 'test_prepared_tab';");
-
-$node_publisher->wait_for_catchup($appname);
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub SET(streaming = apply)");
-# check that transaction is committed on subscriber
-$result = $node_subscriber->safe_psql('postgres',
- "SELECT count(*), count(c), count(d = 999) FROM test_tab");
-is($result, qq(3335|3335|3335),
- 'Rows inserted by 2PC (as well as outside insert) have committed on subscriber, and extra columns contain local defaults'
-);
+$node_publisher->poll_query_until('postgres',
+ "SELECT pid != $oldpid FROM pg_stat_replication WHERE application_name = '$appname' AND state = 'streaming';"
+ )
+ or die
+ "Timed out while waiting for apply to restart after changing SUBSCRIPTION";
-$result = $node_subscriber->safe_psql('postgres',
- "SELECT count(*) FROM pg_prepared_xacts;");
-is($result, qq(0), 'transaction is committed on subscriber');
+test_streaming($node_publisher, $node_subscriber, $appname, 1);
###############################
# check all the cleanup
--
2.23.0.windows.1
[application/octet-stream] v10-0003-The-v4-patch-0001-0003-in-thread-1.patch (17.2K, ../OS3PR01MB62756BD9482EB6BB1CA4CD4D9EAA9@OS3PR01MB6275.jpnprd01.prod.outlook.com/4-v10-0003-The-v4-patch-0001-0003-in-thread-1.patch)
download | inline diff:
From d5fcde5bcd6cf093376bdc670c39f74936d1919a Mon Sep 17 00:00:00 2001
From: wangw <[email protected]>
Date: Tue, 14 Jun 2022 11:21:25 +0800
Subject: [PATCH v10 3/5] The v4 patch(0001~0003) in thread [1]
[1] -
https://www.postgresql.org/message-id/OS0PR01MB5716377C85D4A164E6A7D45F94AB9%40OS0PR01MB5716.jpnprd01.prod.outlook.com
---
src/backend/replication/logical/relation.c | 226 +++++++++++++--------
src/backend/replication/logical/worker.c | 25 ++-
src/include/replication/logicalrelation.h | 1 +
src/test/subscription/t/013_partition.pl | 85 +++++++-
4 files changed, 242 insertions(+), 95 deletions(-)
diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index 80fb561a9a..f342396310 100644
--- a/src/backend/replication/logical/relation.c
+++ b/src/backend/replication/logical/relation.c
@@ -249,6 +249,72 @@ logicalrep_report_missing_attrs(LogicalRepRelation *remoterel,
}
}
+/*
+ * Check if replica identity matches and mark the updatable flag.
+ *
+ * We allow for stricter replica identity (fewer columns) on subscriber as
+ * that will not stop us from finding unique tuple. IE, if publisher has
+ * identity (id,timestamp) and subscriber just (id) this will not be a
+ * problem, but in the opposite scenario it will.
+ *
+ * Don't throw any error here just mark the relation entry as not updatable,
+ * as replica identity is only for updates and deletes but inserts can be
+ * replicated even without it.
+ */
+static void
+logicalrep_rel_mark_updatable(LogicalRepRelMapEntry *entry)
+{
+ Bitmapset *idkey;
+ LogicalRepRelation *remoterel = &entry->remoterel;
+ int i;
+
+ entry->updatable = true;
+
+ /*
+ * If it is a partitioned table, we don't check it, we will check its
+ * partition later.
+ */
+ if (entry->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+ return;
+
+ idkey = RelationGetIndexAttrBitmap(entry->localrel,
+ INDEX_ATTR_BITMAP_IDENTITY_KEY);
+ /* fallback to PK if no replica identity */
+ if (idkey == NULL)
+ {
+ idkey = RelationGetIndexAttrBitmap(entry->localrel,
+ INDEX_ATTR_BITMAP_PRIMARY_KEY);
+ /*
+ * If no replica identity index and no PK, the published table
+ * must have replica identity FULL.
+ */
+ if (idkey == NULL && remoterel->replident != REPLICA_IDENTITY_FULL)
+ entry->updatable = false;
+ }
+
+ i = -1;
+ while ((i = bms_next_member(idkey, i)) >= 0)
+ {
+ int attnum = i + FirstLowInvalidHeapAttributeNumber;
+
+ if (!AttrNumberIsForUserDefinedAttr(attnum))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("logical replication target relation \"%s.%s\" uses "
+ "system columns in REPLICA IDENTITY index",
+ remoterel->nspname, remoterel->relname)));
+
+ attnum = AttrNumberGetAttrOffset(attnum);
+
+ if (entry->attrmap->attnums[attnum] < 0 ||
+ !bms_is_member(entry->attrmap->attnums[attnum], remoterel->attkeys))
+ {
+ entry->updatable = false;
+ break;
+ }
+ }
+}
+
/*
* Open the local relation associated with the remote one.
*
@@ -307,7 +373,6 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
if (!entry->localrelvalid)
{
Oid relid;
- Bitmapset *idkey;
TupleDesc desc;
MemoryContext oldctx;
int i;
@@ -365,55 +430,8 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
/* be tidy */
bms_free(missingatts);
- /*
- * Check that replica identity matches. We allow for stricter replica
- * identity (fewer columns) on subscriber as that will not stop us
- * from finding unique tuple. IE, if publisher has identity
- * (id,timestamp) and subscriber just (id) this will not be a problem,
- * but in the opposite scenario it will.
- *
- * Don't throw any error here just mark the relation entry as not
- * updatable, as replica identity is only for updates and deletes but
- * inserts can be replicated even without it.
- */
- entry->updatable = true;
- idkey = RelationGetIndexAttrBitmap(entry->localrel,
- INDEX_ATTR_BITMAP_IDENTITY_KEY);
- /* fallback to PK if no replica identity */
- if (idkey == NULL)
- {
- idkey = RelationGetIndexAttrBitmap(entry->localrel,
- INDEX_ATTR_BITMAP_PRIMARY_KEY);
-
- /*
- * If no replica identity index and no PK, the published table
- * must have replica identity FULL.
- */
- if (idkey == NULL && remoterel->replident != REPLICA_IDENTITY_FULL)
- entry->updatable = false;
- }
-
- i = -1;
- while ((i = bms_next_member(idkey, i)) >= 0)
- {
- int attnum = i + FirstLowInvalidHeapAttributeNumber;
-
- if (!AttrNumberIsForUserDefinedAttr(attnum))
- ereport(ERROR,
- (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
- errmsg("logical replication target relation \"%s.%s\" uses "
- "system columns in REPLICA IDENTITY index",
- remoterel->nspname, remoterel->relname)));
-
- attnum = AttrNumberGetAttrOffset(attnum);
-
- if (entry->attrmap->attnums[attnum] < 0 ||
- !bms_is_member(entry->attrmap->attnums[attnum], remoterel->attkeys))
- {
- entry->updatable = false;
- break;
- }
- }
+ /* Check that replica identity matches. */
+ logicalrep_rel_mark_updatable(entry);
entry->localrelvalid = true;
}
@@ -436,22 +454,13 @@ logicalrep_rel_close(LogicalRepRelMapEntry *rel, LOCKMODE lockmode)
rel->localrel = NULL;
}
-/*
- * Partition cache: look up partition LogicalRepRelMapEntry's
- *
- * Unlike relation map cache, this is keyed by partition OID, not remote
- * relation OID, because we only have to use this cache in the case where
- * partitions are not directly mapped to any remote relation, such as when
- * replication is occurring with one of their ancestors as target.
- */
-
/*
* Relcache invalidation callback
*/
static void
logicalrep_partmap_invalidate_cb(Datum arg, Oid reloid)
{
- LogicalRepRelMapEntry *entry;
+ LogicalRepPartMapEntry *entry;
/* Just to be sure. */
if (LogicalRepPartMap == NULL)
@@ -464,11 +473,11 @@ logicalrep_partmap_invalidate_cb(Datum arg, Oid reloid)
hash_seq_init(&status, LogicalRepPartMap);
/* TODO, use inverse lookup hashtable? */
- while ((entry = (LogicalRepRelMapEntry *) hash_seq_search(&status)) != NULL)
+ while ((entry = (LogicalRepPartMapEntry *) hash_seq_search(&status)) != NULL)
{
- if (entry->localreloid == reloid)
+ if (entry->relmapentry.localreloid == reloid)
{
- entry->localrelvalid = false;
+ entry->relmapentry.localrelvalid = false;
hash_seq_term(&status);
break;
}
@@ -481,8 +490,42 @@ logicalrep_partmap_invalidate_cb(Datum arg, Oid reloid)
hash_seq_init(&status, LogicalRepPartMap);
- while ((entry = (LogicalRepRelMapEntry *) hash_seq_search(&status)) != NULL)
- entry->localrelvalid = false;
+ while ((entry = (LogicalRepPartMapEntry *) hash_seq_search(&status)) != NULL)
+ entry->relmapentry.localrelvalid = false;
+ }
+}
+
+/*
+ * Reset the entries in the partition map that refer to remoterel
+ *
+ * Called when new relation mapping is sent by the publisher to update our
+ * expected view of incoming data from said publisher.
+ *
+ * Note that we don't update the remoterel information in the entry here,
+ * we will update the information in logicalrep_partition_open to avoid
+ * unnecessary work.
+ */
+void
+logicalrep_partmap_reset_relmap(LogicalRepRelation *remoterel)
+{
+ HASH_SEQ_STATUS status;
+ LogicalRepPartMapEntry *part_entry;
+ LogicalRepRelMapEntry *entry;
+
+ if (LogicalRepPartMap == NULL)
+ return;
+
+ hash_seq_init(&status, LogicalRepPartMap);
+ while ((part_entry = (LogicalRepPartMapEntry *) hash_seq_search(&status)) != NULL)
+ {
+ entry = &part_entry->relmapentry;
+
+ if (entry->remoterel.remoteid != remoterel->remoteid)
+ continue;
+
+ logicalrep_relmap_free_entry(entry);
+
+ memset(entry, 0, sizeof(LogicalRepRelMapEntry));
}
}
@@ -534,7 +577,6 @@ logicalrep_partition_open(LogicalRepRelMapEntry *root,
Oid partOid = RelationGetRelid(partrel);
AttrMap *attrmap = root->attrmap;
bool found;
- int i;
MemoryContext oldctx;
if (LogicalRepPartMap == NULL)
@@ -545,31 +587,40 @@ logicalrep_partition_open(LogicalRepRelMapEntry *root,
(void *) &partOid,
HASH_ENTER, &found);
- if (found)
- return &part_entry->relmapentry;
+ entry = &part_entry->relmapentry;
- memset(part_entry, 0, sizeof(LogicalRepPartMapEntry));
+ if (found && entry->localrelvalid)
+ return entry;
/* Switch to longer-lived context. */
oldctx = MemoryContextSwitchTo(LogicalRepPartMapContext);
- part_entry->partoid = partOid;
+ if (!found)
+ {
+ memset(part_entry, 0, sizeof(LogicalRepPartMapEntry));
+ part_entry->partoid = partOid;
+ }
- /* Remote relation is copied as-is from the root entry. */
- entry = &part_entry->relmapentry;
- entry->remoterel.remoteid = remoterel->remoteid;
- entry->remoterel.nspname = pstrdup(remoterel->nspname);
- entry->remoterel.relname = pstrdup(remoterel->relname);
- entry->remoterel.natts = remoterel->natts;
- entry->remoterel.attnames = palloc(remoterel->natts * sizeof(char *));
- entry->remoterel.atttyps = palloc(remoterel->natts * sizeof(Oid));
- for (i = 0; i < remoterel->natts; i++)
+ if (!entry->remoterel.remoteid)
{
- entry->remoterel.attnames[i] = pstrdup(remoterel->attnames[i]);
- entry->remoterel.atttyps[i] = remoterel->atttyps[i];
+ int i;
+
+ /* Remote relation is copied as-is from the root entry. */
+ entry = &part_entry->relmapentry;
+ entry->remoterel.remoteid = remoterel->remoteid;
+ entry->remoterel.nspname = pstrdup(remoterel->nspname);
+ entry->remoterel.relname = pstrdup(remoterel->relname);
+ entry->remoterel.natts = remoterel->natts;
+ entry->remoterel.attnames = palloc(remoterel->natts * sizeof(char *));
+ entry->remoterel.atttyps = palloc(remoterel->natts * sizeof(Oid));
+ for (i = 0; i < remoterel->natts; i++)
+ {
+ entry->remoterel.attnames[i] = pstrdup(remoterel->attnames[i]);
+ entry->remoterel.atttyps[i] = remoterel->atttyps[i];
+ }
+ entry->remoterel.replident = remoterel->replident;
+ entry->remoterel.attkeys = bms_copy(remoterel->attkeys);
}
- entry->remoterel.replident = remoterel->replident;
- entry->remoterel.attkeys = bms_copy(remoterel->attkeys);
entry->localrel = partrel;
entry->localreloid = partOid;
@@ -594,7 +645,11 @@ logicalrep_partition_open(LogicalRepRelMapEntry *root,
{
AttrNumber root_attno = map->attnums[attno];
- entry->attrmap->attnums[attno] = attrmap->attnums[root_attno - 1];
+ /* 0 means it's a dropped attribute */
+ if (root_attno == 0)
+ entry->attrmap->attnums[attno] = -1;
+ else
+ entry->attrmap->attnums[attno] = attrmap->attnums[root_attno - 1];
}
}
else
@@ -605,7 +660,8 @@ logicalrep_partition_open(LogicalRepRelMapEntry *root,
attrmap->maplen * sizeof(AttrNumber));
}
- entry->updatable = root->updatable;
+ /* Check that replica identity matches. */
+ logicalrep_rel_mark_updatable(entry);
entry->localrelvalid = true;
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 3f8dea0fc1..4e3bcf7c28 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1803,6 +1803,11 @@ apply_handle_relation(StringInfo s)
rel = logicalrep_read_rel(s);
logicalrep_relmap_update(rel);
+
+ /*
+ * Also reset all entries in the partition map that refer to remoterel.
+ */
+ logicalrep_partmap_reset_relmap(rel);
}
/*
@@ -2359,6 +2364,8 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
TupleTableSlot *remoteslot_part;
TupleConversionMap *map;
MemoryContext oldctx;
+ LogicalRepRelMapEntry *part_entry;
+ AttrMap *attrmap = NULL;
/* ModifyTableState is needed for ExecFindPartition(). */
edata->mtstate = mtstate = makeNode(ModifyTableState);
@@ -2390,8 +2397,11 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
remoteslot_part = table_slot_create(partrel, &estate->es_tupleTable);
map = partrelinfo->ri_RootToPartitionMap;
if (map != NULL)
- remoteslot_part = execute_attr_map_slot(map->attrMap, remoteslot,
+ {
+ attrmap = map->attrMap;
+ remoteslot_part = execute_attr_map_slot(attrmap, remoteslot,
remoteslot_part);
+ }
else
{
remoteslot_part = ExecCopySlot(remoteslot_part, remoteslot);
@@ -2399,6 +2409,14 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
}
MemoryContextSwitchTo(oldctx);
+ /* Check if we can do the update or delete on the leaf partition. */
+ if(operation == CMD_UPDATE || operation == CMD_DELETE)
+ {
+ part_entry = logicalrep_partition_open(relmapentry, partrel,
+ attrmap);
+ check_relation_updatable(part_entry);
+ }
+
switch (operation)
{
case CMD_INSERT:
@@ -2420,15 +2438,10 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
* suitable partition.
*/
{
- AttrMap *attrmap = map ? map->attrMap : NULL;
- LogicalRepRelMapEntry *part_entry;
TupleTableSlot *localslot;
ResultRelInfo *partrelinfo_new;
bool found;
- part_entry = logicalrep_partition_open(relmapentry, partrel,
- attrmap);
-
/* Get the matching local tuple from the partition. */
found = FindReplTupleInLocalRel(estate, partrel,
&part_entry->remoterel,
diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h
index 7bf8cd22bd..78cd7e77f5 100644
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -38,6 +38,7 @@ typedef struct LogicalRepRelMapEntry
} LogicalRepRelMapEntry;
extern void logicalrep_relmap_update(LogicalRepRelation *remoterel);
+extern void logicalrep_partmap_reset_relmap(LogicalRepRelation *remoterel);
extern LogicalRepRelMapEntry *logicalrep_rel_open(LogicalRepRelId remoteid,
LOCKMODE lockmode);
diff --git a/src/test/subscription/t/013_partition.pl b/src/test/subscription/t/013_partition.pl
index e7f4a94f19..df897c86db 100644
--- a/src/test/subscription/t/013_partition.pl
+++ b/src/test/subscription/t/013_partition.pl
@@ -800,9 +800,86 @@ ok( $logfile =~
qr/logical replication did not find row to be deleted in replication target relation "tab2_1"/,
'delete target row is missing in tab2_1');
-# No need for this until more tests are added.
-# $node_subscriber1->append_conf('postgresql.conf',
-# "log_min_messages = warning");
-# $node_subscriber1->reload;
+$node_subscriber1->append_conf('postgresql.conf',
+ "log_min_messages = warning");
+$node_subscriber1->reload;
+
+# Test the case that target table on subscriber is a partitioned table and
+# check that the changes are replicated correctly after changing the schema of
+# table on subcriber.
+
+$node_publisher->safe_psql(
+ 'postgres', q{
+ CREATE TABLE tab5 (a int NOT NULL, b int);
+ CREATE UNIQUE INDEX tab5_a_idx ON tab5 (a);
+ ALTER TABLE tab5 REPLICA IDENTITY USING INDEX tab5_a_idx;});
+
+$node_subscriber2->safe_psql(
+ 'postgres', q{
+ CREATE TABLE tab5 (a int NOT NULL, b int, c int) PARTITION BY LIST (a);
+ CREATE TABLE tab5_1 PARTITION OF tab5 DEFAULT;
+ CREATE UNIQUE INDEX tab5_a_idx ON tab5 (a);
+ ALTER TABLE tab5 REPLICA IDENTITY USING INDEX tab5_a_idx;
+ ALTER TABLE tab5_1 REPLICA IDENTITY USING INDEX tab5_1_a_idx;});
+
+$node_subscriber2->safe_psql('postgres',
+ "ALTER SUBSCRIPTION sub2 REFRESH PUBLICATION");
+
+$node_subscriber2->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+# Make partition map cache
+$node_publisher->safe_psql('postgres', "INSERT INTO tab5 VALUES (1, 1)");
+$node_publisher->safe_psql('postgres', "UPDATE tab5 SET a = 2 WHERE a = 1");
+
+$node_publisher->wait_for_catchup('sub2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b FROM tab5 ORDER BY 1");
+is($result, qq(2|1), 'updates of tab5 replicated correctly');
+
+# Change the column order of partition on subscriber
+$node_subscriber2->safe_psql(
+ 'postgres', q{
+ ALTER TABLE tab5 DETACH PARTITION tab5_1;
+ ALTER TABLE tab5_1 DROP COLUMN b;
+ ALTER TABLE tab5_1 ADD COLUMN b int;
+ ALTER TABLE tab5 ATTACH PARTITION tab5_1 DEFAULT});
+
+$node_publisher->safe_psql('postgres', "UPDATE tab5 SET a = 3 WHERE a = 2");
+
+$node_publisher->wait_for_catchup('sub2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b, c FROM tab5 ORDER BY 1");
+is($result, qq(3|1|), 'updates of tab5 replicated correctly after altering table on subscriber');
+
+# Change the column order of table on publisher
+$node_publisher->safe_psql(
+ 'postgres', q{
+ ALTER TABLE tab5 DROP COLUMN b, ADD COLUMN c INT;
+ ALTER TABLE tab5 ADD COLUMN b INT;});
+
+$node_publisher->safe_psql('postgres', "UPDATE tab5 SET c = 1 WHERE a = 3");
+
+$node_publisher->wait_for_catchup('sub2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b, c FROM tab5 ORDER BY 1");
+is($result, qq(3||1), 'updates of tab5 replicated correctly after altering table on publisher');
+
+# Alter REPLICA IDENTITY on subscriber.
+# No REPLICA IDENTITY in the partitioned table on subscriber, but what we check
+# is the partition, so it works fine.
+$node_subscriber2->safe_psql('postgres',
+ "ALTER TABLE tab5 REPLICA IDENTITY NOTHING");
+
+$node_publisher->safe_psql('postgres', "UPDATE tab5 SET a = 4 WHERE a = 3");
+
+$node_publisher->wait_for_catchup('sub2');
+
+$result = $node_subscriber2->safe_psql('postgres',
+ "SELECT a, b, c FROM tab5_1 ORDER BY 1");
+is($result, qq(4||1), 'updates of tab5 replicated correctly');
done_testing();
--
2.23.0.windows.1
[application/octet-stream] v10-0004-Add-some-checks-before-using-apply-background-wo.patch (29.1K, ../OS3PR01MB62756BD9482EB6BB1CA4CD4D9EAA9@OS3PR01MB6275.jpnprd01.prod.outlook.com/5-v10-0004-Add-some-checks-before-using-apply-background-wo.patch)
download | inline diff:
From 1a55a4e0bb1789a91732a872015f17493032a33f Mon Sep 17 00:00:00 2001
From: wangw <[email protected]>
Date: Tue, 14 Jun 2022 11:23:52 +0800
Subject: [PATCH v10 4/5] Add some checks before using apply background worker
to apply changes.
If any of the following checks are violated, an error will be reported.
1. The unique columns between publisher and subscriber are difference.
2. There is any non-immutable function present in expression in
subscriber's relation. Check from the following 3 items:
a. The function in triggers;
b. Column default value expressions and domain constraints;
c. Constraint expressions.
---
doc/src/sgml/ref/create_subscription.sgml | 4 +
.../replication/logical/applybgwroker.c | 45 ++++
src/backend/replication/logical/proto.c | 62 ++++-
src/backend/replication/logical/relation.c | 198 +++++++++++++++
src/backend/replication/logical/tablesync.c | 1 +
src/backend/replication/logical/worker.c | 15 +-
src/backend/utils/cache/typcache.c | 17 ++
src/include/replication/logicalproto.h | 1 +
src/include/replication/logicalrelation.h | 15 ++
src/include/replication/worker_internal.h | 1 +
src/include/utils/typcache.h | 2 +
.../subscription/t/022_twophase_cascade.pl | 7 +
.../subscription/t/032_streaming_apply.pl | 238 ++++++++++++++++++
13 files changed, 598 insertions(+), 8 deletions(-)
create mode 100644 src/test/subscription/t/032_streaming_apply.pl
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index d4caae0222..8de1a23ce4 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -240,6 +240,10 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
the transaction is committed. Note that if an error happens when
applying changes in a background worker, it might not report the
finish LSN of the remote transaction in the server log.
+ To run in this mode, there are following two requirements. The first
+ is that the unique column should be the same between publisher and
+ subscriber; the second is that there should not be any non-immutable
+ function in subscriber-side replicated table.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/replication/logical/applybgwroker.c b/src/backend/replication/logical/applybgwroker.c
index 1f52b155d7..7d3c61c2e9 100644
--- a/src/backend/replication/logical/applybgwroker.c
+++ b/src/backend/replication/logical/applybgwroker.c
@@ -763,3 +763,48 @@ apply_bgworker_subxact_info_add(TransactionId current_xid)
MemoryContextSwitchTo(oldctx);
}
}
+
+/*
+ * Check if changes on this logical replication relation can be applied by
+ * apply background worker.
+ *
+ * Although we maintains the commit order by allowing only one process to
+ * commit at a time, our access order to the relation has changed.
+ * This could cause unexpected problems if the unique column on the replicated
+ * table is inconsistent with the publisher-side or contains non-immutable
+ * functions when applying transactions in the apply background worker.
+ */
+void
+apply_bgworker_relation_check(LogicalRepRelMapEntry *rel)
+{
+ /* Check only we are in apply bgworker. */
+ if (!am_apply_bgworker())
+ return;
+
+ /*
+ * If it is a partitioned table, we do not check it, we will check its
+ * partition later.
+ */
+ if (rel->localrel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+ return;
+
+ /*
+ * If any unique index exist, check that they are same as remoterel.
+ */
+ if (!rel->sameunique)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot replicate relation with different unique index"),
+ errhint("Please change the streaming option to 'on' instead of 'apply'.")));
+
+ /*
+ * Check if there is any non-immutable function present in expression in
+ * this relation.
+ */
+ if (rel->volatility == FUNCTION_NONIMMUTABLE)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot replicate relation. There is at least one non-immutable function"),
+ errhint("Please change the streaming option to 'on' instead of 'apply'.")));
+
+}
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index a888038eb4..6af3302270 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -23,7 +23,8 @@
/*
* Protocol message flags.
*/
-#define LOGICALREP_IS_REPLICA_IDENTITY 1
+#define LOGICALREP_IS_REPLICA_IDENTITY 0x0001
+#define LOGICALREP_IS_UNIQUE 0x0002
#define MESSAGE_TRANSACTIONAL (1<<0)
#define TRUNCATE_CASCADE (1<<0)
@@ -933,11 +934,55 @@ logicalrep_write_attrs(StringInfo out, Relation rel, Bitmapset *columns)
TupleDesc desc;
int i;
uint16 nliveatts = 0;
- Bitmapset *idattrs = NULL;
+ Bitmapset *idattrs = NULL,
+ *attunique = NULL;
bool replidentfull;
desc = RelationGetDescr(rel);
+ if (rel->rd_rel->relhasindex)
+ {
+ List *indexoidlist = RelationGetIndexList(rel);
+ ListCell *indexoidscan;
+
+ foreach(indexoidscan, indexoidlist)
+ {
+ Oid indexoid = lfirst_oid(indexoidscan);
+ Relation indexRel;
+
+ /* Look up the description for index */
+ indexRel = RelationIdGetRelation(indexoid);
+
+ if (!RelationIsValid(indexRel))
+ elog(ERROR, "could not open relation with OID %u", indexoid);
+
+ if (indexRel->rd_index->indisunique)
+ {
+ int i;
+
+ /* Add referenced attributes to idindexattrs */
+ for (i = 0; i < indexRel->rd_index->indnatts; i++)
+ {
+ int attrnum = indexRel->rd_index->indkey.values[i];
+
+ /*
+ * We don't include non-key columns into idindexattrs
+ * bitmaps. See RelationGetIndexAttrBitmap.
+ */
+ if (attrnum != 0)
+ {
+ if (i < indexRel->rd_index->indnkeyatts &&
+ !bms_is_member(attrnum - FirstLowInvalidHeapAttributeNumber, attunique))
+ attunique = bms_add_member(attunique,
+ attrnum - FirstLowInvalidHeapAttributeNumber);
+ }
+ }
+ }
+ RelationClose(indexRel);
+ }
+ list_free(indexoidlist);
+ }
+
/* send number of live attributes */
for (i = 0; i < desc->natts; i++)
{
@@ -976,6 +1021,10 @@ logicalrep_write_attrs(StringInfo out, Relation rel, Bitmapset *columns)
idattrs))
flags |= LOGICALREP_IS_REPLICA_IDENTITY;
+ if (bms_is_member(att->attnum - FirstLowInvalidHeapAttributeNumber,
+ attunique))
+ flags |= LOGICALREP_IS_UNIQUE;
+
pq_sendbyte(out, flags);
/* attribute name */
@@ -1001,7 +1050,8 @@ logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel)
int natts;
char **attnames;
Oid *atttyps;
- Bitmapset *attkeys = NULL;
+ Bitmapset *attkeys = NULL,
+ *attunique = NULL;
natts = pq_getmsgint(in, 2);
attnames = palloc(natts * sizeof(char *));
@@ -1012,11 +1062,14 @@ logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel)
{
uint8 flags;
- /* Check for replica identity column */
+ /* Check for replica identity and unique column */
flags = pq_getmsgbyte(in);
if (flags & LOGICALREP_IS_REPLICA_IDENTITY)
attkeys = bms_add_member(attkeys, i);
+ if (flags & LOGICALREP_IS_UNIQUE)
+ attunique = bms_add_member(attunique, i);
+
/* attribute name */
attnames[i] = pstrdup(pq_getmsgstring(in));
@@ -1030,6 +1083,7 @@ logicalrep_read_attrs(StringInfo in, LogicalRepRelation *rel)
rel->attnames = attnames;
rel->atttyps = atttyps;
rel->attkeys = attkeys;
+ rel->attunique = attunique;
rel->natts = natts;
}
diff --git a/src/backend/replication/logical/relation.c b/src/backend/replication/logical/relation.c
index f342396310..03fbc61209 100644
--- a/src/backend/replication/logical/relation.c
+++ b/src/backend/replication/logical/relation.c
@@ -19,12 +19,19 @@
#include "access/table.h"
#include "catalog/namespace.h"
+#include "catalog/pg_proc.h"
#include "catalog/pg_subscription_rel.h"
+#include "commands/trigger.h"
#include "executor/executor.h"
#include "nodes/makefuncs.h"
+#include "optimizer/optimizer.h"
#include "replication/logicalrelation.h"
#include "replication/worker_internal.h"
+#include "rewrite/rewriteHandler.h"
#include "utils/inval.h"
+#include "utils/lsyscache.h"
+#include "utils/syscache.h"
+#include "utils/typcache.h"
static MemoryContext LogicalRepRelMapContext = NULL;
@@ -91,6 +98,23 @@ logicalrep_relmap_invalidate_cb(Datum arg, Oid reloid)
}
}
+/*
+ * Reset the flag volatility of all existing entry in the relation map cache.
+ */
+static void
+logicalrep_relmap_reset_volatility_cb(Datum arg, int cacheid, uint32 hashvalue)
+{
+ HASH_SEQ_STATUS hash_seq;
+ LogicalRepRelMapEntry *entry;
+
+ if (LogicalRepRelMap == NULL)
+ return;
+
+ hash_seq_init(&hash_seq, LogicalRepRelMap);
+ while ((entry = hash_seq_search(&hash_seq)) != NULL)
+ entry->volatility = FUNCTION_UNKNOWN;
+}
+
/*
* Initialize the relation map cache.
*/
@@ -116,6 +140,9 @@ logicalrep_relmap_init(void)
/* Watch for invalidation events. */
CacheRegisterRelcacheCallback(logicalrep_relmap_invalidate_cb,
(Datum) 0);
+ CacheRegisterSyscacheCallback(PROCOID,
+ logicalrep_relmap_reset_volatility_cb,
+ (Datum) 0);
}
/*
@@ -142,6 +169,7 @@ logicalrep_relmap_free_entry(LogicalRepRelMapEntry *entry)
pfree(remoterel->atttyps);
}
bms_free(remoterel->attkeys);
+ bms_free(remoterel->attunique);
if (entry->attrmap)
pfree(entry->attrmap);
@@ -190,6 +218,7 @@ logicalrep_relmap_update(LogicalRepRelation *remoterel)
}
entry->remoterel.replident = remoterel->replident;
entry->remoterel.attkeys = bms_copy(remoterel->attkeys);
+ entry->remoterel.attunique = bms_copy(remoterel->attunique);
MemoryContextSwitchTo(oldctx);
}
@@ -315,6 +344,162 @@ logicalrep_rel_mark_updatable(LogicalRepRelMapEntry *entry)
}
}
+/*
+ * Helper function to mark flags for apply background worker when opening the
+ * local table.
+ */
+static void
+logicalrep_rel_mark_apply_bgworker(LogicalRepRelMapEntry *entry)
+{
+ Bitmapset *ukey;
+ int i;
+ TupleDesc tupdesc;
+ int attnum;
+ List *fkeys = NIL;
+
+ /*
+ * Check whether the unique index of publisher and subscriber are
+ * consistent.
+ */
+ entry->sameunique = true;
+ ukey = RelationGetIndexAttrBitmap(entry->localrel,
+ INDEX_ATTR_BITMAP_KEY);
+
+ if (ukey)
+ {
+ i = -1;
+ while ((i = bms_next_member(ukey, i)) >= 0)
+ {
+ int attnum = i + FirstLowInvalidHeapAttributeNumber;
+
+ attnum = AttrNumberGetAttrOffset(attnum);
+
+ if (entry->attrmap->attnums[attnum] < 0 ||
+ !bms_is_member(entry->attrmap->attnums[attnum], entry->remoterel.attunique))
+ {
+ entry->sameunique = false;
+ break;
+ }
+ }
+ }
+
+ /*
+ * Check whether there is any non-immutable function in the local table.
+ *
+ * a. The function in triggers;
+ * b. Column default value expressions and domain constraints;
+ * c. Constraint expressions;
+ * d. Foreign keys.
+ */
+ if (entry->volatility != FUNCTION_UNKNOWN)
+ return;
+
+ /* Initialize the flag. */
+ entry->volatility = FUNCTION_IMMUTABLE;
+
+ /* Check the trigger functions. */
+ if (entry->localrel->trigdesc != NULL)
+ {
+ for (i = 0; i < entry->localrel->trigdesc->numtriggers; i++)
+ {
+ Trigger *trig = entry->localrel->trigdesc->triggers + i;
+
+ if (trig->tgenabled != TRIGGER_FIRES_ALWAYS &&
+ trig->tgenabled != TRIGGER_FIRES_ON_REPLICA)
+ continue;
+
+ if (func_volatile(trig->tgfoid) != PROVOLATILE_IMMUTABLE)
+ {
+ entry->volatility = FUNCTION_NONIMMUTABLE;
+ return;
+ }
+ }
+ }
+
+ /* Check the columns. */
+ tupdesc = RelationGetDescr(entry->localrel);
+ for (attnum = 0; attnum < tupdesc->natts; attnum++)
+ {
+ Form_pg_attribute att = TupleDescAttr(tupdesc, attnum);
+
+ /* We don't need info for dropped or generated attributes */
+ if (att->attisdropped || att->attgenerated)
+ continue;
+
+ /*
+ * We don't need to check columns that only exist on the
+ * subscriber
+ */
+ if (entry->attrmap->attnums[attnum] < 0)
+ continue;
+
+ if (att->atthasdef)
+ {
+ Node *defaultexpr;
+
+ defaultexpr = build_column_default(entry->localrel, attnum + 1);
+ if (contain_mutable_functions(defaultexpr))
+ {
+ entry->volatility = FUNCTION_NONIMMUTABLE;
+ return;
+ }
+ }
+
+ /*
+ * If the column is of a DOMAIN type, determine whether
+ * that domain has any CHECK expressions that are not
+ * immutable.
+ */
+ if (get_typtype(att->atttypid) == TYPTYPE_DOMAIN)
+ {
+ List *domain_constraints;
+ ListCell *lc;
+
+ domain_constraints = GetDomainConstraints(att->atttypid);
+
+ foreach(lc, domain_constraints)
+ {
+ DomainConstraintState *con = (DomainConstraintState *) lfirst(lc);
+
+ if (con->check_expr && contain_mutable_functions((Node *) con->check_expr))
+ {
+ entry->volatility = FUNCTION_NONIMMUTABLE;
+ return;
+ }
+ }
+ }
+ }
+
+
+ /* Check the constraints. */
+ tupdesc = RelationGetDescr(entry->localrel);
+
+ if (tupdesc->constr)
+ {
+ ConstrCheck *check = tupdesc->constr->check;
+
+ /*
+ * Determine if there are any CHECK constraints which
+ * contains non-immutable function.
+ */
+ for (i = 0; i < tupdesc->constr->num_check; i++)
+ {
+ Expr *check_expr = stringToNode(check[i].ccbin);
+
+ if (contain_mutable_functions((Node *) check_expr))
+ {
+ entry->volatility = FUNCTION_NONIMMUTABLE;
+ return;
+ }
+ }
+ }
+
+ /* Check the foreign keys. */
+ fkeys = RelationGetFKeyList(entry->localrel);
+ if (fkeys)
+ entry->volatility = FUNCTION_NONIMMUTABLE;
+}
+
/*
* Open the local relation associated with the remote one.
*
@@ -433,6 +618,12 @@ logicalrep_rel_open(LogicalRepRelId remoteid, LOCKMODE lockmode)
/* Check that replica identity matches. */
logicalrep_rel_mark_updatable(entry);
+ /*
+ * Set flags to check later whether changes could be applied in the
+ * apply background worker.
+ */
+ logicalrep_rel_mark_apply_bgworker(entry);
+
entry->localrelvalid = true;
}
@@ -620,6 +811,7 @@ logicalrep_partition_open(LogicalRepRelMapEntry *root,
}
entry->remoterel.replident = remoterel->replident;
entry->remoterel.attkeys = bms_copy(remoterel->attkeys);
+ entry->remoterel.attunique = bms_copy(remoterel->attunique);
}
entry->localrel = partrel;
@@ -663,6 +855,12 @@ logicalrep_partition_open(LogicalRepRelMapEntry *root,
/* Check that replica identity matches. */
logicalrep_rel_mark_updatable(entry);
+ /*
+ * Set flags to check later whether changes could be applied in the apply
+ * background worker.
+ */
+ logicalrep_rel_mark_apply_bgworker(entry);
+
entry->localrelvalid = true;
/* state and statelsn are left set to 0. */
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 8ffba7e2e5..3cdbf8b457 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -884,6 +884,7 @@ fetch_remote_table_info(char *nspname, char *relname,
lrel->attnames = palloc0(MaxTupleAttributeNumber * sizeof(char *));
lrel->atttyps = palloc0(MaxTupleAttributeNumber * sizeof(Oid));
lrel->attkeys = NULL;
+ lrel->attunique = NULL;
/*
* Store the columns as a list of names. Ignore those that are not
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 4e3bcf7c28..1b83a74685 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1920,6 +1920,8 @@ apply_handle_insert(StringInfo s)
/* Set relation for error callback */
apply_error_callback_arg.rel = rel;
+ apply_bgworker_relation_check(rel);
+
/* Initialize the executor state. */
edata = create_edata_for_relation(rel);
estate = edata->estate;
@@ -2056,6 +2058,8 @@ apply_handle_update(StringInfo s)
/* Check if we can do the update. */
check_relation_updatable(rel);
+ apply_bgworker_relation_check(rel);
+
/* Initialize the executor state. */
edata = create_edata_for_relation(rel);
estate = edata->estate;
@@ -2224,6 +2228,8 @@ apply_handle_delete(StringInfo s)
/* Check if we can do the delete. */
check_relation_updatable(rel);
+ apply_bgworker_relation_check(rel);
+
/* Initialize the executor state. */
edata = create_edata_for_relation(rel);
estate = edata->estate;
@@ -2409,13 +2415,14 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
}
MemoryContextSwitchTo(oldctx);
+ part_entry = logicalrep_partition_open(relmapentry, partrel,
+ attrmap);
+
+ apply_bgworker_relation_check(part_entry);
+
/* Check if we can do the update or delete on the leaf partition. */
if(operation == CMD_UPDATE || operation == CMD_DELETE)
- {
- part_entry = logicalrep_partition_open(relmapentry, partrel,
- attrmap);
check_relation_updatable(part_entry);
- }
switch (operation)
{
diff --git a/src/backend/utils/cache/typcache.c b/src/backend/utils/cache/typcache.c
index 808f9ebd0d..b248899d82 100644
--- a/src/backend/utils/cache/typcache.c
+++ b/src/backend/utils/cache/typcache.c
@@ -2540,6 +2540,23 @@ compare_values_of_enum(TypeCacheEntry *tcache, Oid arg1, Oid arg2)
return 0;
}
+/*
+ * GetDomainConstraints --- get DomainConstraintState list of specified domain type
+ */
+List *
+GetDomainConstraints(Oid type_id)
+{
+ TypeCacheEntry *typentry;
+ List *constraints = NIL;
+
+ typentry = lookup_type_cache(type_id, TYPECACHE_DOMAIN_CONSTR_INFO);
+
+ if(typentry->domainData != NULL)
+ constraints = typentry->domainData->constraints;
+
+ return constraints;
+}
+
/*
* Load (or re-load) the enumData member of the typcache entry.
*/
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index 7bba77c9e7..a503eb62c2 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -108,6 +108,7 @@ typedef struct LogicalRepRelation
char replident; /* replica identity */
char relkind; /* remote relation kind */
Bitmapset *attkeys; /* Bitmap of key columns */
+ Bitmapset *attunique; /* Bitmap of unique columns */
} LogicalRepRelation;
/* Type mapping info */
diff --git a/src/include/replication/logicalrelation.h b/src/include/replication/logicalrelation.h
index 78cd7e77f5..75b3a86c7d 100644
--- a/src/include/replication/logicalrelation.h
+++ b/src/include/replication/logicalrelation.h
@@ -15,6 +15,17 @@
#include "access/attmap.h"
#include "replication/logicalproto.h"
+/*
+ * States to determine volatility of the function in expressions in one
+ * relation.
+ */
+typedef enum RelFuncVolatility
+{
+ FUNCTION_UNKNOWN = 0, /* initializing */
+ FUNCTION_IMMUTABLE, /* all functions are immutable function */
+ FUNCTION_NONIMMUTABLE /* at least one non-immutable function */
+} RelFuncVolatility;
+
typedef struct LogicalRepRelMapEntry
{
LogicalRepRelation remoterel; /* key is remoterel.remoteid */
@@ -31,6 +42,10 @@ typedef struct LogicalRepRelMapEntry
Relation localrel; /* relcache entry (NULL when closed) */
AttrMap *attrmap; /* map of local attributes to remote ones */
bool updatable; /* Can apply updates/deletes? */
+ bool sameunique; /* Is the unique column of local and remote
+ consistent? */
+ RelFuncVolatility volatility; /* all functions in localrel are
+ immutable function? */
/* Sync state. */
char state;
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 08e63de013..dd49d8e2ec 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -196,6 +196,7 @@ extern void apply_bgworker_free(ApplyBgworkerState *wstate);
extern void apply_bgworker_check_status(void);
extern void apply_bgworker_set_status(ApplyBgworkerStatus status);
extern void apply_bgworker_subxact_info_add(TransactionId current_xid);
+extern void apply_bgworker_relation_check(LogicalRepRelMapEntry *rel);
static inline bool
am_tablesync_worker(void)
diff --git a/src/include/utils/typcache.h b/src/include/utils/typcache.h
index 431ad7f1b3..ed7c2e7f48 100644
--- a/src/include/utils/typcache.h
+++ b/src/include/utils/typcache.h
@@ -199,6 +199,8 @@ extern uint64 assign_record_type_identifier(Oid type_id, int32 typmod);
extern int compare_values_of_enum(TypeCacheEntry *tcache, Oid arg1, Oid arg2);
+extern List *GetDomainConstraints(Oid type_id);
+
extern size_t SharedRecordTypmodRegistryEstimate(void);
extern void SharedRecordTypmodRegistryInit(SharedRecordTypmodRegistry *,
diff --git a/src/test/subscription/t/022_twophase_cascade.pl b/src/test/subscription/t/022_twophase_cascade.pl
index b52af74e35..81791c8d3a 100644
--- a/src/test/subscription/t/022_twophase_cascade.pl
+++ b/src/test/subscription/t/022_twophase_cascade.pl
@@ -39,6 +39,13 @@ sub test_streaming
ALTER SUBSCRIPTION tap_sub_C
SET (streaming = $streaming_mode)");
+ if ($streaming_mode eq 'apply')
+ {
+ $node_C->safe_psql(
+ 'postgres', "
+ ALTER TABLE test_tab ALTER c DROP DEFAULT");
+ }
+
# Wait for subscribers to finish initialization
$node_A->poll_query_until(
diff --git a/src/test/subscription/t/032_streaming_apply.pl b/src/test/subscription/t/032_streaming_apply.pl
new file mode 100644
index 0000000000..776f1390fa
--- /dev/null
+++ b/src/test/subscription/t/032_streaming_apply.pl
@@ -0,0 +1,238 @@
+# Copyright (c) 2022, PostgreSQL Global Development Group
+
+# Test the restrictions of streaming mode "apply" in logical replication
+
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+my $offset = 0;
+
+# Create publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf',
+ 'logical_decoding_work_mem = 64kB');
+$node_publisher->start;
+
+# Create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->start;
+
+# Setup structure on publisher
+$node_publisher->safe_psql('postgres',
+ "CREATE TABLE test_tab (a int primary key, b varchar)");
+
+# Setup structure on subscriber
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE test_tab (a int primary key, b varchar)");
+
+# Setup logical replication
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub FOR TABLE test_tab");
+
+my $appname = 'tap_sub';
+$node_subscriber->safe_psql(
+ 'postgres', "
+ CREATE SUBSCRIPTION tap_sub
+ CONNECTION '$publisher_connstr application_name=$appname'
+ PUBLICATION tap_pub
+ WITH (streaming = apply, copy_data = false)");
+
+$node_publisher->wait_for_catchup($appname);
+
+# It is not allowed that the unique index on the publisher and the subscriber
+# is different. Check the error reported by background worker in this case.
+$node_subscriber->safe_psql('postgres',
+ "CREATE UNIQUE INDEX test_tab_b_idx ON test_tab (b)");
+
+# Check that a background worker starts if "streaming" option is
+# specified as "apply". We have to look for the DEBUG1 log messages
+# about that, so temporarily bump up the log verbosity.
+$node_subscriber->append_conf('postgresql.conf', "log_min_messages = debug1");
+$node_subscriber->reload;
+
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1, 5000) s(i)"
+);
+
+$node_subscriber->wait_for_log(qr/\[Apply BGW #\d+\] started/, 0);
+$node_subscriber->append_conf('postgresql.conf',
+ "log_min_messages = warning");
+$node_subscriber->reload;
+
+$node_subscriber->wait_for_log(
+ qr/ERROR: cannot replicate relation with different unique index/,
+ $offset);
+
+# Drop the unique index on the subscriber, now it works.
+$node_subscriber->safe_psql('postgres', "DROP INDEX test_tab_b_idx");
+
+$node_publisher->wait_for_catchup($appname);
+
+my $result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab");
+is($result, qq(5000), 'data replicated to subscriber after dropping index');
+
+# Triggers which execute non-immutable function are not allowed on the
+# subscriber side. Check the error reported by background worker in this case.
+$node_subscriber->safe_psql(
+ 'postgres', qq{
+CREATE FUNCTION trigger_func() RETURNS TRIGGER AS \$\$
+ BEGIN
+ RETURN NULL;
+ END
+\$\$ language plpgsql;
+CREATE TRIGGER insert_trig
+BEFORE INSERT ON test_tab
+FOR EACH ROW EXECUTE PROCEDURE trigger_func();
+ALTER TABLE test_tab ENABLE REPLICA TRIGGER insert_trig;
+});
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql('postgres', "DELETE FROM test_tab");
+
+$node_subscriber->wait_for_log(
+ qr/ERROR: cannot replicate relation. There is at least one non-immutable function/,
+ $offset);
+
+# Drop the trigger on the subscriber, now it works.
+$node_subscriber->safe_psql('postgres',
+ "DROP TRIGGER insert_trig ON test_tab");
+
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab");
+is($result, qq(0), 'data replicated to subscriber after dropping trigger');
+
+# It is not allowed that column default value expression contains a
+# non-immutable function on the subscriber side. Check the error reported by
+# background worker in this case.
+$node_subscriber->safe_psql('postgres',
+ "ALTER TABLE test_tab ALTER COLUMN b SET DEFAULT random()");
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1, 5000) s(i)"
+);
+
+$node_subscriber->wait_for_log(
+ qr/ERROR: cannot replicate relation. There is at least one non-immutable function/,
+ $offset);
+
+# Drop default value on the subscriber, now it works.
+$node_subscriber->safe_psql('postgres',
+ "ALTER TABLE test_tab ALTER COLUMN b DROP DEFAULT");
+
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab");
+is($result, qq(5000),
+ 'data replicated to subscriber after dropping default value expression');
+
+# It is not allowed that domain constraint expression contains a non-immutable
+# function on the subscriber side. Check the error reported by background
+# worker in this case.
+$node_subscriber->safe_psql(
+ 'postgres', qq{
+CREATE DOMAIN test_domain AS int CHECK(VALUE > random());
+ALTER TABLE test_tab ALTER COLUMN a TYPE test_domain;
+});
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql('postgres', "DELETE FROM test_tab");
+
+$node_subscriber->wait_for_log(
+ qr/ERROR: cannot replicate relation. There is at least one non-immutable function/,
+ $offset);
+
+# Drop domain constraint expression on the subscriber, now it works.
+$node_subscriber->safe_psql('postgres',
+ "ALTER TABLE test_tab ALTER COLUMN a TYPE int");
+
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab");
+is($result, qq(0),
+ 'data replicated to subscriber after dropping domain constraint expression'
+);
+
+# It is not allowed that constraint expression contains a non-immutable function
+# on the subscriber side. Check the error reported by background worker in this
+# case.
+$node_subscriber->safe_psql(
+ 'postgres', qq{
+ALTER TABLE test_tab ADD CONSTRAINT test_tab_con check (a > random());
+});
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1, 5000) s(i)"
+);
+
+$node_subscriber->wait_for_log(
+ qr/ERROR: cannot replicate relation. There is at least one non-immutable function/,
+ $offset);
+
+# Drop constraint on the subscriber, now it works.
+$node_subscriber->safe_psql('postgres',
+ "ALTER TABLE test_tab DROP CONSTRAINT test_tab_con");
+
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab");
+is($result, qq(5000),
+ 'data replicated to subscriber after dropping constraint expression');
+
+# It is not allowed that foreign key on the subscriber side. Check the error
+# reported by background worker in this case.
+$node_publisher->safe_psql('postgres', "DELETE FROM test_tab");
+$node_publisher->wait_for_catchup($appname);
+$node_subscriber->safe_psql(
+ 'postgres', qq{
+CREATE TABLE test_tab_f (a int primary key);
+ALTER TABLE test_tab ADD CONSTRAINT test_tabfk FOREIGN KEY(a) REFERENCES test_tab_f(a);
+});
+
+# Check the subscriber log from now on.
+$offset = -s $node_subscriber->logfile;
+
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(1, 5000) s(i)"
+);
+
+$node_subscriber->wait_for_log(
+ qr/ERROR: cannot replicate relation. There is at least one non-immutable function/,
+ $offset);
+
+# Drop the foreign key constraint on the subscriber, now it works.
+$node_subscriber->safe_psql('postgres',
+ "ALTER TABLE test_tab DROP CONSTRAINT test_tabfk");
+
+$node_publisher->wait_for_catchup($appname);
+
+$result =
+ $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab");
+is($result, qq(5000),
+ 'data replicated to subscriber after dropping the foreign key');
+
+$node_subscriber->stop;
+$node_publisher->stop;
+
+done_testing();
--
2.23.0.windows.1
[application/octet-stream] v10-0005-Retry-to-apply-streaming-xact-only-in-apply-work.patch (20.4K, ../OS3PR01MB62756BD9482EB6BB1CA4CD4D9EAA9@OS3PR01MB6275.jpnprd01.prod.outlook.com/6-v10-0005-Retry-to-apply-streaming-xact-only-in-apply-work.patch)
download | inline diff:
From 56c7537e5b8e5044866d7fe7a1a6d691b2afa30d Mon Sep 17 00:00:00 2001
From: wangw <[email protected]>
Date: Fri, 10 Jun 2022 15:04:52 +0800
Subject: [PATCH v10 5/5] Retry to apply streaming xact only in apply worker.
---
doc/src/sgml/catalogs.sgml | 11 +++
doc/src/sgml/ref/create_subscription.sgml | 4 +
src/backend/catalog/pg_subscription.c | 1 +
src/backend/catalog/system_views.sql | 4 +-
src/backend/commands/subscriptioncmds.c | 1 +
.../replication/logical/applybgwroker.c | 13 ++-
src/backend/replication/logical/worker.c | 89 ++++++++++++++++++
src/bin/pg_dump/pg_dump.c | 5 +-
src/include/catalog/pg_subscription.h | 4 +
.../subscription/t/032_streaming_apply.pl | 91 ++++++++++---------
10 files changed, 174 insertions(+), 49 deletions(-)
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 5cb39b18bd..928ad02ace 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -7907,6 +7907,17 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>subretry</structfield> <type>bool</type>
+ </para>
+ <para>
+ If true, the subscription will not try to apply streaming transaction
+ in <literal>apply</literal> mode. See
+ <xref linkend="sql-createsubscription"/> for more information.
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>subconninfo</structfield> <type>text</type>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 8de1a23ce4..70fbf81668 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -244,6 +244,10 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
is that the unique column should be the same between publisher and
subscriber; the second is that there should not be any non-immutable
function in subscriber-side replicated table.
+ When applying a streaming transaction, if either requirement is not
+ met, the background worker will exit with an error. And when retrying
+ later, we will try to apply this transaction in <literal>on</literal>
+ mode.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index add51caadf..1824390d7d 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -71,6 +71,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->stream = subform->substream;
sub->twophasestate = subform->subtwophasestate;
sub->disableonerr = subform->subdisableonerr;
+ sub->retry = subform->subretry;
/* Get conninfo */
datum = SysCacheGetAttr(SUBSCRIPTIONOID,
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index fedaed533b..10f4dd6785 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1298,8 +1298,8 @@ REVOKE ALL ON pg_replication_origin_status FROM public;
-- All columns of pg_subscription except subconninfo are publicly readable.
REVOKE ALL ON pg_subscription FROM public;
GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
- subbinary, substream, subtwophasestate, subdisableonerr, subslotname,
- subsynccommit, subpublications)
+ subbinary, substream, subtwophasestate, subdisableonerr,
+ subretry, subslotname, subsynccommit, subpublications)
ON pg_subscription TO public;
CREATE VIEW pg_stat_subscription_stats AS
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 4080bba987..38697184d9 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -663,6 +663,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
LOGICALREP_TWOPHASE_STATE_PENDING :
LOGICALREP_TWOPHASE_STATE_DISABLED);
values[Anum_pg_subscription_subdisableonerr - 1] = BoolGetDatum(opts.disableonerr);
+ values[Anum_pg_subscription_subretry - 1] = BoolGetDatum(false);
values[Anum_pg_subscription_subconninfo - 1] =
CStringGetTextDatum(conninfo);
if (opts.slot_name)
diff --git a/src/backend/replication/logical/applybgwroker.c b/src/backend/replication/logical/applybgwroker.c
index 7d3c61c2e9..02b1ef5e62 100644
--- a/src/backend/replication/logical/applybgwroker.c
+++ b/src/backend/replication/logical/applybgwroker.c
@@ -103,6 +103,13 @@ apply_bgworker_find_or_start(TransactionId xid, bool start)
if (start && !XLogRecPtrIsInvalid(MySubscription->skiplsn))
return NULL;
+ if (start && MySubscription->retry)
+ {
+ elog(DEBUG1, "retry to apply an streaming transaction in apply "
+ "background worker");
+ return NULL;
+ }
+
/*
* For streaming transactions that are being applied in apply background
* worker, we cannot decide whether to apply the change for a relation
@@ -794,8 +801,7 @@ apply_bgworker_relation_check(LogicalRepRelMapEntry *rel)
if (!rel->sameunique)
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
- errmsg("cannot replicate relation with different unique index"),
- errhint("Please change the streaming option to 'on' instead of 'apply'.")));
+ errmsg("cannot replicate relation with different unique index")));
/*
* Check if there is any non-immutable function present in expression in
@@ -804,7 +810,6 @@ apply_bgworker_relation_check(LogicalRepRelMapEntry *rel)
if (rel->volatility == FUNCTION_NONIMMUTABLE)
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
- errmsg("cannot replicate relation. There is at least one non-immutable function"),
- errhint("Please change the streaming option to 'on' instead of 'apply'.")));
+ errmsg("cannot replicate relation. There is at least one non-immutable function")));
}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 1b83a74685..b82f973626 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -374,6 +374,8 @@ static void clear_subscription_skip_lsn(XLogRecPtr finish_lsn);
static inline void set_apply_error_context_xact(TransactionId xid, XLogRecPtr lsn);
static inline void reset_apply_error_context_info(void);
+static void set_subscription_retry(bool retry);
+
/*
* Should this worker apply changes for given relation.
*
@@ -890,6 +892,9 @@ apply_handle_commit(StringInfo s)
apply_handle_commit_internal(&commit_data);
+ /* Set the flag that we will not retry later. */
+ set_subscription_retry(false);
+
/* Check the status of apply background worker if any. */
apply_bgworker_check_status();
@@ -1001,6 +1006,9 @@ apply_handle_prepare(StringInfo s)
in_remote_transaction = false;
+ /* Set the flag that we will not retry later. */
+ set_subscription_retry(false);
+
/* Check the status of apply background worker if any. */
apply_bgworker_check_status();
@@ -1054,6 +1062,9 @@ apply_handle_commit_prepared(StringInfo s)
store_flush_position(prepare_data.end_lsn);
in_remote_transaction = false;
+ /* Set the flag that we will not retry later. */
+ set_subscription_retry(false);
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(prepare_data.end_lsn);
@@ -1109,6 +1120,9 @@ apply_handle_rollback_prepared(StringInfo s)
store_flush_position(rollback_data.rollback_end_lsn);
in_remote_transaction = false;
+ /* Set the flag that we will not retry later. */
+ set_subscription_retry(false);
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(rollback_data.rollback_end_lsn);
@@ -1195,6 +1209,9 @@ apply_handle_stream_prepare(StringInfo s)
/* unlink the files with serialized changes and subxact info. */
stream_cleanup_files(MyLogicalRepWorker->subid, prepare_data.xid);
}
+
+ /* Set the flag that we will not retry later. */
+ set_subscription_retry(false);
}
in_remote_transaction = false;
@@ -1615,6 +1632,9 @@ apply_handle_stream_abort(StringInfo s)
*/
else
serialize_stream_abort(xid, subxid);
+
+ /* Set the flag that we will not retry later. */
+ set_subscription_retry(false);
}
}
@@ -2788,6 +2808,9 @@ apply_handle_stream_commit(StringInfo s)
/* unlink the files with serialized changes and subxact info */
stream_cleanup_files(MyLogicalRepWorker->subid, xid);
}
+
+ /* Set the flag that we will not retry later. */
+ set_subscription_retry(false);
}
/* Check the status of apply background worker if any. */
@@ -3856,6 +3879,9 @@ start_table_sync(XLogRecPtr *origin_startpos, char **myslotname)
}
PG_CATCH();
{
+ /* Set the flag that we will retry later. */
+ set_subscription_retry(true);
+
if (MySubscription->disableonerr)
DisableSubscriptionAndExit();
else
@@ -3894,6 +3920,9 @@ start_apply(XLogRecPtr origin_startpos)
}
PG_CATCH();
{
+ /* Set the flag that we will retry later. */
+ set_subscription_retry(true);
+
if (MySubscription->disableonerr)
DisableSubscriptionAndExit();
else
@@ -4395,3 +4424,63 @@ reset_apply_error_context_info(void)
apply_error_callback_arg.remote_attnum = -1;
set_apply_error_context_xact(InvalidTransactionId, InvalidXLogRecPtr);
}
+
+/*
+ * Set subretry of pg_subscription catalog.
+ *
+ * If retry is true, subscriber is about to exit with an error. Otherwise, it
+ * means that the changes was applied successfully.
+ */
+static void
+set_subscription_retry(bool retry)
+{
+ Relation rel;
+ HeapTuple tup;
+ bool started_tx = false;
+ bool nulls[Natts_pg_subscription];
+ bool replaces[Natts_pg_subscription];
+ Datum values[Natts_pg_subscription];
+
+ if (MySubscription->retry == retry ||
+ am_apply_bgworker())
+ return;
+
+ if (!IsTransactionState())
+ {
+ StartTransactionCommand();
+ started_tx = true;
+ }
+
+ /* Look up the subscription in the catalog */
+ rel = table_open(SubscriptionRelationId, RowExclusiveLock);
+ tup = SearchSysCacheCopy1(SUBSCRIPTIONOID,
+ ObjectIdGetDatum(MySubscription->oid));
+
+ if (!HeapTupleIsValid(tup))
+ elog(ERROR, "subscription \"%s\" does not exist", MySubscription->name);
+
+ LockSharedObject(SubscriptionRelationId, MySubscription->oid, 0,
+ AccessShareLock);
+
+ /* Form a new tuple. */
+ memset(values, 0, sizeof(values));
+ memset(nulls, false, sizeof(nulls));
+ memset(replaces, false, sizeof(replaces));
+
+ /* reset subretry */
+ values[Anum_pg_subscription_subretry - 1] = BoolGetDatum(retry);
+ replaces[Anum_pg_subscription_subretry - 1] = true;
+
+ tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
+ replaces);
+
+ /* Update the catalog. */
+ CatalogTupleUpdate(rel, &tup->t_self, tup);
+
+ /* Cleanup. */
+ heap_freetuple(tup);
+ table_close(rel, NoLock);
+
+ if (started_tx)
+ CommitTransactionCommand();
+}
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 6f8b30abb0..9b3ffe0888 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4472,8 +4472,9 @@ getSubscriptions(Archive *fout)
ntups = PQntuples(res);
/*
- * Get subscription fields. We don't include subskiplsn in the dump as
- * after restoring the dump this value may no longer be relevant.
+ * Get subscription fields. We don't include subskiplsn and subretry in
+ * the dump as after restoring the dump this value may no longer be
+ * relevant.
*/
i_tableoid = PQfnumber(res, "tableoid");
i_oid = PQfnumber(res, "oid");
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 9b394a45fe..dc7597bf80 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -76,6 +76,8 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
bool subdisableonerr; /* True if a worker error should cause the
* subscription to be disabled */
+ bool subretry; /* True if the previous apply change failed. */
+
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* Connection string to the publisher */
text subconninfo BKI_FORCE_NOT_NULL;
@@ -115,6 +117,8 @@ typedef struct Subscription
bool disableonerr; /* Indicates if the subscription should be
* automatically disabled if a worker error
* occurs */
+ bool retry; /* Indicates if the previous apply change
+ * failed. */
char *conninfo; /* Connection string to the publisher */
char *slotname; /* Name of the replication slot */
char *synccommit; /* Synchronous commit setting for worker */
diff --git a/src/test/subscription/t/032_streaming_apply.pl b/src/test/subscription/t/032_streaming_apply.pl
index 776f1390fa..51483b04a1 100644
--- a/src/test/subscription/t/032_streaming_apply.pl
+++ b/src/test/subscription/t/032_streaming_apply.pl
@@ -46,7 +46,9 @@ $node_subscriber->safe_psql(
$node_publisher->wait_for_catchup($appname);
# It is not allowed that the unique index on the publisher and the subscriber
-# is different. Check the error reported by background worker in this case.
+# is different. Check the error reported by background worker in this case. And
+# after retrying in apply worker, we check if the data is replicated
+# successfully.
$node_subscriber->safe_psql('postgres',
"CREATE UNIQUE INDEX test_tab_b_idx ON test_tab (b)");
@@ -69,17 +71,20 @@ $node_subscriber->wait_for_log(
qr/ERROR: cannot replicate relation with different unique index/,
$offset);
-# Drop the unique index on the subscriber, now it works.
-$node_subscriber->safe_psql('postgres', "DROP INDEX test_tab_b_idx");
-
+# Wait for this streaming transaction to be applied in the apply worker.
$node_publisher->wait_for_catchup($appname);
my $result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab");
-is($result, qq(5000), 'data replicated to subscriber after dropping index');
+is($result, qq(5000), 'data replicated to subscribers after retrying');
+
+# Drop the unique index on the subscriber.
+$node_subscriber->safe_psql('postgres', "DROP INDEX test_tab_b_idx");
# Triggers which execute non-immutable function are not allowed on the
# subscriber side. Check the error reported by background worker in this case.
+# And after retrying in apply worker, we check if the data is replicated
+# successfully.
$node_subscriber->safe_psql(
'postgres', qq{
CREATE FUNCTION trigger_func() RETURNS TRIGGER AS \$\$
@@ -102,19 +107,21 @@ $node_subscriber->wait_for_log(
qr/ERROR: cannot replicate relation. There is at least one non-immutable function/,
$offset);
-# Drop the trigger on the subscriber, now it works.
-$node_subscriber->safe_psql('postgres',
- "DROP TRIGGER insert_trig ON test_tab");
-
+# Wait for this streaming transaction to be applied in the apply worker.
$node_publisher->wait_for_catchup($appname);
$result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab");
-is($result, qq(0), 'data replicated to subscriber after dropping trigger');
+is($result, qq(0), 'data replicated to subscribers after retrying');
+
+# Drop the trigger on the subscriber.
+$node_subscriber->safe_psql('postgres',
+ "DROP TRIGGER insert_trig ON test_tab");
# It is not allowed that column default value expression contains a
# non-immutable function on the subscriber side. Check the error reported by
-# background worker in this case.
+# background worker in this case. And after retrying in apply worker, we check
+# if the data is replicated successfully.
$node_subscriber->safe_psql('postgres',
"ALTER TABLE test_tab ALTER COLUMN b SET DEFAULT random()");
@@ -129,20 +136,21 @@ $node_subscriber->wait_for_log(
qr/ERROR: cannot replicate relation. There is at least one non-immutable function/,
$offset);
-# Drop default value on the subscriber, now it works.
-$node_subscriber->safe_psql('postgres',
- "ALTER TABLE test_tab ALTER COLUMN b DROP DEFAULT");
-
+# Wait for this streaming transaction to be applied in the apply worker.
$node_publisher->wait_for_catchup($appname);
$result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab");
-is($result, qq(5000),
- 'data replicated to subscriber after dropping default value expression');
+is($result, qq(5000), 'data replicated to subscribers after retrying');
+
+# Drop default value on the subscriber.
+$node_subscriber->safe_psql('postgres',
+ "ALTER TABLE test_tab ALTER COLUMN b DROP DEFAULT");
# It is not allowed that domain constraint expression contains a non-immutable
# function on the subscriber side. Check the error reported by background
-# worker in this case.
+# worker in this case. And after retrying in apply worker, we check if the data
+# is replicated successfully.
$node_subscriber->safe_psql(
'postgres', qq{
CREATE DOMAIN test_domain AS int CHECK(VALUE > random());
@@ -158,21 +166,21 @@ $node_subscriber->wait_for_log(
qr/ERROR: cannot replicate relation. There is at least one non-immutable function/,
$offset);
-# Drop domain constraint expression on the subscriber, now it works.
-$node_subscriber->safe_psql('postgres',
- "ALTER TABLE test_tab ALTER COLUMN a TYPE int");
-
+# Wait for this streaming transaction to be applied in the apply worker.
$node_publisher->wait_for_catchup($appname);
$result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab");
-is($result, qq(0),
- 'data replicated to subscriber after dropping domain constraint expression'
-);
+is($result, qq(0), 'data replicated to subscribers after retrying');
+
+# Drop domain constraint expression on the subscriber.
+$node_subscriber->safe_psql('postgres',
+ "ALTER TABLE test_tab ALTER COLUMN a TYPE int");
-# It is not allowed that constraint expression contains a non-immutable function
-# on the subscriber side. Check the error reported by background worker in this
-# case.
+# It is not allowed that constraint expression contains a non-immutable
+# function on the subscriber side. Check the error reported by background
+# worker in this case. And after retrying in apply worker, we check if the data
+# is replicated successfully.
$node_subscriber->safe_psql(
'postgres', qq{
ALTER TABLE test_tab ADD CONSTRAINT test_tab_con check (a > random());
@@ -189,19 +197,20 @@ $node_subscriber->wait_for_log(
qr/ERROR: cannot replicate relation. There is at least one non-immutable function/,
$offset);
-# Drop constraint on the subscriber, now it works.
-$node_subscriber->safe_psql('postgres',
- "ALTER TABLE test_tab DROP CONSTRAINT test_tab_con");
-
+# Wait for this streaming transaction to be applied in the apply worker.
$node_publisher->wait_for_catchup($appname);
$result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab");
-is($result, qq(5000),
- 'data replicated to subscriber after dropping constraint expression');
+is($result, qq(5000), 'data replicated to subscribers after retrying');
+
+# Drop constraint on the subscriber.
+$node_subscriber->safe_psql('postgres',
+ "ALTER TABLE test_tab DROP CONSTRAINT test_tab_con");
# It is not allowed that foreign key on the subscriber side. Check the error
-# reported by background worker in this case.
+# reported by background worker in this case. And after retrying in apply
+# worker, we check if the data is replicated successfully.
$node_publisher->safe_psql('postgres', "DELETE FROM test_tab");
$node_publisher->wait_for_catchup($appname);
$node_subscriber->safe_psql(
@@ -221,16 +230,16 @@ $node_subscriber->wait_for_log(
qr/ERROR: cannot replicate relation. There is at least one non-immutable function/,
$offset);
-# Drop the foreign key constraint on the subscriber, now it works.
-$node_subscriber->safe_psql('postgres',
- "ALTER TABLE test_tab DROP CONSTRAINT test_tabfk");
-
+# Wait for this streaming transaction to be applied in the apply worker.
$node_publisher->wait_for_catchup($appname);
$result =
$node_subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tab");
-is($result, qq(5000),
- 'data replicated to subscriber after dropping the foreign key');
+is($result, qq(5000), 'data replicated to subscribers after retrying');
+
+# Drop the foreign key constraint on the subscriber.
+$node_subscriber->safe_psql('postgres',
+ "ALTER TABLE test_tab DROP CONSTRAINT test_tabfk");
$node_subscriber->stop;
$node_publisher->stop;
--
2.23.0.windows.1
view thread (633+ 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: Perform streaming logical transactions by background workers and parallel apply
In-Reply-To: <OS3PR01MB62756BD9482EB6BB1CA4CD4D9EAA9@OS3PR01MB6275.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