public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: 'Peter Smith' <[email protected]>
Cc: Amit Kapila <[email protected]>
Cc: [email protected] <[email protected]>
Cc: [email protected] <[email protected]>
Cc: vignesh C <[email protected]>
Cc: Greg Nancarrow <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Cc: Mark Dilger <[email protected]>
Cc: Smith, Peter <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: RE: Optionally automatically disable logical replication subscriptions on error
Date: Mon, 21 Feb 2022 00:25:22 +0000
Message-ID: <TYCPR01MB837326C879A9CDE756073A7FED3A9@TYCPR01MB8373.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAHut+PviMp-5MTgs-Wq=ryV87j6k=GdcPfkZ1R51wv7ecDqq4g@mail.gmail.com>
References: <TYCPR01MB83732011D83A027C32A2802CED6A9@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<CAJcOf-dkEVtRstWMchEAjb02xp_5cg_5g_0GUCthUx+o-bYdXg@mail.gmail.com>
<TYCPR01MB83737F77BBE63432611CE7C8ED6D9@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<CALDaNm1O8iBHSjfkv2-Ld6B7C4ay8F3-O1Q2QHVrM8H7Zu40iA@mail.gmail.com>
<TYCPR01MB8373DEBEB6EC28CB82F50D47ED759@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<CAJcOf-fY7VspwzN2PPDQq1c1DpvhxHuDv4DWq+=zDWb_60u-rQ@mail.gmail.com>
<TYCPR01MB837344D0306FFCAAE280F498ED779@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<OS3PR01MB6275C137E5032EC46290DB319E439@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<TYCPR01MB83732A590139C59DE0A83CFBED4B9@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<OS0PR01MB61132035CF205C142FF6E2E2FB4C9@OS0PR01MB6113.jpnprd01.prod.outlook.com>
<TYCPR01MB8373EEAA0EC0D45EC442F1C9ED4C9@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<CAA4eK1J3c1DJsw8RGc84sc2DmCe8gbW3pEf2CiA95bO+WOT=dA@mail.gmail.com>
<TYCPR01MB83736F1DE9A86E5FE64DC819ED349@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<TYCPR01MB8373CCCEFF852F17E1C5D85EED359@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<CAHut+PviMp-5MTgs-Wq=ryV87j6k=GdcPfkZ1R51wv7ecDqq4g@mail.gmail.com>
On Friday, February 18, 2022 3:27 PM Peter Smith <[email protected]> wrote:
> Hi. Below are my code review comments for v18.
Thank you for your review !
> ==========
>
> 1. Commit Message - wording
>
> BEFORE
> To partially remedy the situation, adding a new subscription_parameter named
> 'disable_on_error'.
>
> AFTER
> To partially remedy the situation, this patch adds a new
> subscription_parameter named 'disable_on_error'.
Fixed.
> ~~~
>
> 2. Commit message - wording
>
> BEFORE
> Require to bump catalog version.
>
> AFTER
> A catalog version bump is required.
Fixed.
> ~~~
>
> 3. doc/src/sgml/ref/alter_subscription.sgml - whitespace
>
> @@ -201,8 +201,8 @@ ALTER SUBSCRIPTION <replaceable
> class="parameter">name</replaceable> RENAME TO <
> information. The parameters that can be altered
> are <literal>slot_name</literal>,
> <literal>synchronous_commit</literal>,
> - <literal>binary</literal>, and
> - <literal>streaming</literal>.
> + <literal>binary</literal>,<literal>streaming</literal>, and
> + <literal>disable_on_error</literal>.
> </para>
>
> There is a missing space before <literal>streaming</literal>.
Fixed.
> ~~~
>
> 4. src/backend/replication/logical/worker.c - WorkerErrorRecovery
>
> @@ -2802,6 +2803,89 @@ LogicalRepApplyLoop(XLogRecPtr
> last_received) }
>
> /*
> + * Worker error recovery processing, in preparation for disabling the
> + * subscription.
> + */
> +static void
> +WorkerErrorRecovery(void)
>
> I was wondering about the need for this to be a separate function? It is only
> called immediately before calling 'DisableSubscriptionOnError'
> so would it maybe be better just to put this code inside
> DisableSubscriptionOnError with the appropriate comments?
I preferred to have one specific for error handling,
because from caller sides, when we catch error, it's apparent
that error recovery is done. But, the function name "DisableSubscriptionOnError"
by itself should have the nuance that we do something on error.
So, we can think that it's okay to have error recovery processing
in this function.
So, I removed the function and fixed some related comments.
> ~~~
>
> 5. src/backend/replication/logical/worker.c - DisableSubscriptionOnError
>
> + /*
> + * We would not be here unless this subscription's disableonerr field
> + was
> + * true when our worker began applying changes, but check whether that
> + * field has changed in the interim.
> + */
>
> Apparently, this function might just do nothing if it detects some situation
> where the flag was changed somehow, but I'm not 100% sure that the callers
> are properly catering for when nothing happens.
>
> IMO it would be better if this function would return true/false to mean "did
> disable subscription happen or not?" because that will give the calling code the
> chance to check the function return and do the right thing - e.g. if the caller first
> thought it should be disabled but then it turned out it did NOT disable...
I don't think we need to do something more.
After this function, table sync worker and the apply worker
just exit. IMO, we don't need to do additional work for
already-disabled subscription on the caller sides.
It should be sufficient to fulfill the purpose of
DisableSubscriptionOnError or confirm it has been fulfilled.
> ~~~
>
> 6. src/backend/replication/logical/worker.c - LogicalRepHandleTableSync
> name
>
> +/*
> + * Execute the initial sync with error handling. Disable the
> +subscription,
> + * if it's required.
> + */
> +static void
> +LogicalRepHandleTableSync(XLogRecPtr *origin_startpos,
> + char **myslotname, MemoryContext cctx)
>
> I felt that it is a bit overkill to put a "LogicalRep" prefix here because it is a static
> function.
>
> IMO this function should be renamed as 'SyncTableStartWrapper' because that
> describes better what it is doing.
Makes sense. Fixed.
> ~~~
>
> 7. src/backend/replication/logical/worker.c - LogicalRepHandleTableSync
> Assert
>
> Even though we can know this to be true because of where it is called from, I
> think the readability of the function will be improved if you add an assertion at
> the top:
>
> Assert(am_tablesync_worker());
Fixed.
> And then, because the function is clearly for Tablesync worker only there is no
> need to keep mentioning that in the subsequent comments...
>
> e.g.1
> /* This is table synchronization worker, call initial sync. */
> AFTER:
> /* Call initial sync. */
Fixed.
> e.g.2
> /*
> * Report the table sync error. There is no corresponding message type
> * for table synchronization.
> */
> AFTER
> /*
> * Report the error. There is no corresponding message type for table
> * synchronization.
> */
Agreed. Fixed
> ~~~
>
> 8. src/backend/replication/logical/worker.c - LogicalRepHandleTableSync
> unnecessarily complex
>
> +static void
> +LogicalRepHandleTableSync(XLogRecPtr *origin_startpos,
> + char **myslotname, MemoryContext cctx) {
> + char *syncslotname;
> + bool error_recovery_done = false;
>
> IMO this logic is way more complex than it needed to be. IIUC that
> 'error_recovery_done' and various conditions can be removed, and the whole
> thing be simplified quite a lot.
>
> I re-wrote this function as a POC. Please see the attached file [2].
> All the tests are still passing OK.
>
> (Perhaps the scenario for my comment #5 above still needs to be addressed?)
Removed the 'error_recovery_done' flag and fixed.
> ~~~
>
> 9. src/backend/replication/logical/worker.c -
> LogicalRepHandleApplyMessages name
>
> +/*
> + * Run the apply loop with error handling. Disable the subscription,
> + * if necessary.
> + */
> +static void
> +LogicalRepHandleApplyMessages(XLogRecPtr origin_startpos,
> + MemoryContext cctx)
>
> I felt that it is a bit overkill to put a "LogicalRep" prefix here because it is a static
> function.
>
> IMO this function should be renamed as 'ApplyLoopWrapper' because that
> describes better what it is doing.
Fixed.
> ~~~
>
> 10. src/backend/replication/logical/worker.c -
> LogicalRepHandleApplyMessages unnecessarily complex
>
> +static void
> +LogicalRepHandleApplyMessages(XLogRecPtr origin_startpos,
> + MemoryContext cctx)
> +{
> + bool error_recovery_done = false;
>
> IMO this logic is way more complex than it needed to be. IIUC that
> 'error_recovery_done' and various conditions can be removed, and the whole
> thing be simplified quite a lot.
>
> I re-wrote this function as a POC. Please see the attached file [2].
> All the tests are still passing OK.
>
> (Perhaps the scenario for my comment #5 above still needs to be addressed?)
Fixed.
> ~~~
>
> 11. src/bin/pg_dump/pg_dump.c - dumpSubscription
>
> @@ -4441,6 +4451,9 @@ dumpSubscription(Archive *fout, const
> SubscriptionInfo *subinfo)
> if (strcmp(subinfo->subtwophasestate, two_phase_disabled) != 0)
> appendPQExpBufferStr(query, ", two_phase = on");
>
> + if (strcmp(subinfo->subdisableonerr, "f") != 0)
> + appendPQExpBufferStr(query, ", disable_on_error = on");
> +
>
> I felt saying disable_on_err is "true" would look more natural than saying it is
> "on".
Fixed.
> ~~~
>
> 12. src/bin/psql/describe.c - describeSubscriptions typo
>
> @@ -6096,11 +6096,13 @@ describeSubscriptions(const char *pattern, bool
> verbose)
> gettext_noop("Binary"),
> gettext_noop("Streaming"));
>
> - /* Two_phase is only supported in v15 and higher */
> + /* Two_phase and disable_on_error is only supported in v15 and higher
> + */
>
> Typo
>
> "is only" --> "are only"
Fixed.
> ~~~
>
> 13. src/include/catalog/pg_subscription.h - comments
>
> @@ -103,6 +106,9 @@ typedef struct Subscription
> * binary format */
> bool stream; /* Allow streaming in-progress transactions. */
> char twophasestate; /* Allow streaming two-phase transactions */
> + bool disableonerr; /* Indicates if the subscription should be
> + * automatically disabled when subscription
> + * workers detect any errors. */
>
> It's not usual to have a full stop here.
> Maybe not needed to repeat the word "subscription".
> IMO, generally, it all can be simplified a bit.
>
> BEFORE
> Indicates if the subscription should be automatically disabled when
> subscription workers detect any errors.
>
> AFTER
> Indicates if the subscription should be automatically disabled if a worker error
> occurs
Fixed.
> ~~~
>
> 14. src/test/regress/sql/subscription.sql - missing test case.
>
> The "conflicting options" error from the below code is not currently being
> tested.
>
> @@ -249,6 +253,15 @@ parse_subscription_options(ParseState *pstate, List
> *stmt_options,
> opts->specified_opts |= SUBOPT_TWOPHASE_COMMIT;
> opts->twophase = defGetBoolean(defel);
> }
> + else if (IsSet(supported_opts, SUBOPT_DISABLE_ON_ERR) &&
> + strcmp(defel->defname, "disable_on_error") == 0) { if
> + (IsSet(opts->specified_opts, SUBOPT_DISABLE_ON_ERR))
> + errorConflictingDefElem(defel, pstate);
We don't have this test in other options as well.
So, this should be aligned.
> ~~~
>
> 15. src/test/subscription/t/028_disable_on_error.pl - 028 clash
>
> Just a heads-up that this 028 is going to clash with the Row-Filter patch 028
> which has been announced to be pushed soon, so be prepared to change this
> number again shortly :)
Thank you for letting me know.
> ~~~
>
> 16. src/test/subscription/t/028_disable_on_error.pl - done_testing
>
> AFAIK is a new style now for the TAP tests where it uses "done_testing();"
> instead of saying up-front how many tests there are.
> See here [1].
Fixed.
> ~~~
>
> 17. src/test/subscription/t/028_disable_on_error.pl - more comments
>
> +# Create an additional unique index in schema s1 on the subscriber
> +only. When # we create subscriptions, below, this should cause
> +subscription "s1" on the # subscriber to fail during initial
> +synchronization and to get automatically # disabled.
>
> I felt it could be made a bit more obvious upfront in a comment that 2 pairs of
> pub/sub will be created, and their names will same as the
> schemas:
> e.g.
> Publisher "s1" --> Subscriber "s1"
> Publisher "s2" --> Subscriber "s2"
Comments are fixed.
> ~~~
>
> 18. src/test/subscription/t/028_disable_on_error.pl - ALTER tests?
>
> The tests here are only using the hardwired 'disable_on_error' options set at
> CREATE SUBSCRIPTION time. There are no TAP tests for changing the
> disable_on_error using ALTER SUBSCRIPTION.
>
> Should there be?
I don't think so. Toggling the flag 'disable_on_error' is already tested
in the subscription.sql file. Both new paths for table sync and apply
worker to disable on error are already covered.
FYI : I skipped one change of worker.c.peter.txt
about "enabled" flag, which is independent from
disable_on_error option.
Kindly have a look at the attached v19.
Best Regards,
Takamichi Osumi
Attachments:
[application/octet-stream] v19-0001-Optionally-disable-subscriptions-on-error.patch (51.2K, ../TYCPR01MB837326C879A9CDE756073A7FED3A9@TYCPR01MB8373.jpnprd01.prod.outlook.com/2-v19-0001-Optionally-disable-subscriptions-on-error.patch)
download | inline diff:
From 8ac7afe1483be1707c84d029164593213f840872 Mon Sep 17 00:00:00 2001
From: Takamichi Osumi <[email protected]>
Date: Sun, 20 Feb 2022 23:55:06 +0000
Subject: [PATCH v19] Optionally disable subscriptions on error
Logical replication apply workers for a subscription can easily get
stuck in an infinite loop of attempting to apply a change,
triggering an error (such as a constraint violation), exiting with
an error written to the subscription worker log, and restarting.
To partially remedy the situation, this patch adds a new
subscription_parameter named 'disable_on_error'. To be consistent
with old behavior, the parameter defaults to false. When true, both
the table sync worker and apply worker catch any errors thrown and
disable the subscription in order to break the loop. The error is
still also written to the logs.
A catalog version bump is required.
Proposed and written originally by Mark Dilger
Taken over by Osumi Takamichi, Greg Nancarrow
Reviewed by Greg Nancarrow, Vignesh C, Amit Kapila, Wang wei, Tang Haiying, Peter Smith
Discussion : https://www.postgresql.org/message-id/DB35438F-9356-4841-89A0-412709EBD3AB%40enterprisedb.com
---
doc/src/sgml/catalogs.sgml | 10 +
doc/src/sgml/ref/alter_subscription.sgml | 4 +-
doc/src/sgml/ref/create_subscription.sgml | 12 ++
src/backend/catalog/pg_subscription.c | 1 +
src/backend/catalog/system_views.sql | 3 +-
src/backend/commands/subscriptioncmds.c | 27 ++-
src/backend/replication/logical/worker.c | 233 +++++++++++++++++-------
src/bin/pg_dump/pg_dump.c | 17 +-
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/psql/describe.c | 10 +-
src/bin/psql/tab-complete.c | 4 +-
src/include/catalog/pg_subscription.h | 6 +
src/test/regress/expected/subscription.out | 119 +++++++-----
src/test/regress/sql/subscription.sql | 14 ++
src/test/subscription/t/028_disable_on_error.pl | 200 ++++++++++++++++++++
15 files changed, 539 insertions(+), 122 deletions(-)
create mode 100644 src/test/subscription/t/028_disable_on_error.pl
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 5a1627a..9c7e80f 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -7762,6 +7762,16 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
<row>
<entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>subdisableonerr</structfield> <type>bool</type>
+ </para>
+ <para>
+ If true, the subscription will be disabled when subscription
+ workers detect any errors
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
<structfield>subconninfo</structfield> <type>text</type>
</para>
<para>
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 0b027cc..5783715 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -201,8 +201,8 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
information. The parameters that can be altered
are <literal>slot_name</literal>,
<literal>synchronous_commit</literal>,
- <literal>binary</literal>, and
- <literal>streaming</literal>.
+ <literal>binary</literal>, <literal>streaming</literal>, and
+ <literal>disable_on_error</literal>.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 990a41f..9ca0bb9 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -142,6 +142,18 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</varlistentry>
<varlistentry>
+ <term><literal>disable_on_error</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ Specifies whether the subscription should be automatically disabled
+ if any errors are detected by subscription workers during data
+ replication from the publisher. The default is
+ <literal>false</literal>.
+ </para>
+ </listitem>
+ </varlistentry>
+
+ <varlistentry>
<term><literal>enabled</literal> (<type>boolean</type>)</term>
<listitem>
<para>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index ca65a8b..d2beebc 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -69,6 +69,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->binary = subform->subbinary;
sub->stream = subform->substream;
sub->twophasestate = subform->subtwophasestate;
+ sub->disableonerr = subform->subdisableonerr;
/* Get conninfo */
datum = SysCacheGetAttr(SUBSCRIPTIONOID,
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 3cb69b1..430ac90 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1261,7 +1261,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, subname, subowner, subenabled, subbinary,
- substream, subtwophasestate, subslotname, subsynccommit, subpublications)
+ substream, subtwophasestate, subdisableonerr, subslotname,
+ subsynccommit, subpublications)
ON pg_subscription TO public;
CREATE VIEW pg_stat_subscription_workers AS
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 3ef6607..3922658 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -61,6 +61,7 @@
#define SUBOPT_BINARY 0x00000080
#define SUBOPT_STREAMING 0x00000100
#define SUBOPT_TWOPHASE_COMMIT 0x00000200
+#define SUBOPT_DISABLE_ON_ERR 0x00000400
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -82,6 +83,7 @@ typedef struct SubOpts
bool binary;
bool streaming;
bool twophase;
+ bool disableonerr;
} SubOpts;
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
@@ -130,6 +132,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->streaming = false;
if (IsSet(supported_opts, SUBOPT_TWOPHASE_COMMIT))
opts->twophase = false;
+ if (IsSet(supported_opts, SUBOPT_DISABLE_ON_ERR))
+ opts->disableonerr = false;
/* Parse options */
foreach(lc, stmt_options)
@@ -249,6 +253,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->specified_opts |= SUBOPT_TWOPHASE_COMMIT;
opts->twophase = defGetBoolean(defel);
}
+ else if (IsSet(supported_opts, SUBOPT_DISABLE_ON_ERR) &&
+ strcmp(defel->defname, "disable_on_error") == 0)
+ {
+ if (IsSet(opts->specified_opts, SUBOPT_DISABLE_ON_ERR))
+ errorConflictingDefElem(defel, pstate);
+
+ opts->specified_opts |= SUBOPT_DISABLE_ON_ERR;
+ opts->disableonerr = defGetBoolean(defel);
+ }
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -390,7 +403,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
supported_opts = (SUBOPT_CONNECT | SUBOPT_ENABLED | SUBOPT_CREATE_SLOT |
SUBOPT_SLOT_NAME | SUBOPT_COPY_DATA |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
- SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT);
+ SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
+ SUBOPT_DISABLE_ON_ERR);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/*
@@ -464,6 +478,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
CharGetDatum(opts.twophase ?
LOGICALREP_TWOPHASE_STATE_PENDING :
LOGICALREP_TWOPHASE_STATE_DISABLED);
+ values[Anum_pg_subscription_subdisableonerr - 1] = BoolGetDatum(opts.disableonerr);
values[Anum_pg_subscription_subconninfo - 1] =
CStringGetTextDatum(conninfo);
if (opts.slot_name)
@@ -864,7 +879,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
{
supported_opts = (SUBOPT_SLOT_NAME |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
- SUBOPT_STREAMING);
+ SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -913,6 +928,14 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
replaces[Anum_pg_subscription_substream - 1] = true;
}
+ if (IsSet(opts.specified_opts, SUBOPT_DISABLE_ON_ERR))
+ {
+ values[Anum_pg_subscription_subdisableonerr - 1]
+ = BoolGetDatum(opts.disableonerr);
+ replaces[Anum_pg_subscription_subdisableonerr - 1]
+ = true;
+ }
+
update_tuple = true;
break;
}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 5d9acc6..a3c240f 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -136,6 +136,7 @@
#include "access/xact.h"
#include "access/xlog_internal.h"
#include "catalog/catalog.h"
+#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/partition.h"
#include "catalog/pg_inherits.h"
@@ -2802,6 +2803,81 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
}
/*
+ * Disable the current subscription, after error recovery processing.
+ */
+static void
+DisableSubscriptionOnError(void)
+{
+ Relation rel;
+ bool nulls[Natts_pg_subscription];
+ bool replaces[Natts_pg_subscription];
+ Datum values[Natts_pg_subscription];
+ HeapTuple tup;
+ Form_pg_subscription subform;
+
+ /* Emit the error */
+ EmitErrorReport();
+ /* Abort any active transaction */
+ AbortOutOfAnyTransaction();
+ /* Reset the ErrorContext */
+ FlushErrorState();
+
+ /* Disable the subscription in a fresh transaction */
+ StartTransactionCommand();
+
+ /* Look up our subscription in the catalogs */
+ rel = table_open(SubscriptionRelationId, RowExclusiveLock);
+ tup = SearchSysCacheCopy2(SUBSCRIPTIONNAME, MyDatabaseId,
+ CStringGetDatum(MySubscription->name));
+ if (!HeapTupleIsValid(tup))
+ ereport(ERROR,
+ errcode(ERRCODE_UNDEFINED_OBJECT),
+ errmsg("subscription \"%s\" does not exist",
+ MySubscription->name));
+
+ subform = (Form_pg_subscription) GETSTRUCT(tup);
+
+ /*
+ * We would not be here unless this subscription's disableonerr field was
+ * true when our worker began applying changes, but check whether that
+ * field has changed in the interim.
+ */
+ if (!subform->subdisableonerr)
+ {
+ heap_freetuple(tup);
+ table_close(rel, RowExclusiveLock);
+ CommitTransactionCommand();
+ return;
+ }
+
+ /* Notify the subscription will be no longer valid */
+ ereport(LOG,
+ errmsg("logical replication subscription \"%s\" will be disabled due to an error",
+ MySubscription->name));
+
+ LockSharedObject(SubscriptionRelationId, subform->oid, 0, AccessExclusiveLock);
+
+ /* Form a new tuple. */
+ memset(values, 0, sizeof(values));
+ memset(nulls, false, sizeof(nulls));
+ memset(replaces, false, sizeof(replaces));
+
+ /* Set the subscription to disabled. */
+ values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(false);
+ replaces[Anum_pg_subscription_subenabled - 1] = true;
+
+ /* Update the catalog */
+ tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
+ replaces);
+ CatalogTupleUpdate(rel, &tup->t_self, tup);
+ heap_freetuple(tup);
+
+ table_close(rel, RowExclusiveLock);
+
+ CommitTransactionCommand();
+}
+
+/*
* Send a Standby Status Update message to server.
*
* 'recvpos' is the latest LSN we've received data to, force is set if we need
@@ -3372,6 +3448,98 @@ TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid)
snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid);
}
+/*
+ * Execute the initial sync with error handling. Disable the subscription,
+ * if it's required.
+ */
+static void
+SyncTableStartWrapper(XLogRecPtr *origin_startpos,
+ char **myslotname, MemoryContext cctx)
+{
+ char *syncslotname;
+
+ Assert(am_tablesync_worker());
+
+ PG_TRY();
+ {
+ /* Call initial sync. */
+ syncslotname = LogicalRepSyncTableStart(origin_startpos);
+ }
+ PG_CATCH();
+ {
+ MemoryContext ecxt = MemoryContextSwitchTo(cctx);
+ ErrorData *errdata = CopyErrorData();
+
+ /*
+ * Report the error. There is no corresponding message type for table
+ * synchronization.
+ */
+ pgstat_report_subworker_error(MyLogicalRepWorker->subid,
+ MyLogicalRepWorker->relid,
+ MyLogicalRepWorker->relid,
+ 0, /* message type */
+ InvalidTransactionId,
+ errdata->message);
+
+ if (MySubscription->disableonerr)
+ {
+ DisableSubscriptionOnError();
+ proc_exit(0);
+ }
+
+ MemoryContextSwitchTo(ecxt);
+ PG_RE_THROW();
+ }
+ PG_END_TRY();
+
+ /* allocate slot name in long-lived context */
+ *myslotname = MemoryContextStrdup(ApplyContext, syncslotname);
+ pfree(syncslotname);
+}
+
+/*
+ * Run the apply loop with error handling. Disable the subscription,
+ * if necessary.
+ */
+static void
+ApplyLoopWrapper(XLogRecPtr origin_startpos, MemoryContext cctx)
+{
+ PG_TRY();
+ {
+ LogicalRepApplyLoop(origin_startpos);
+ }
+ PG_CATCH();
+ {
+ MemoryContext ecxt = MemoryContextSwitchTo(cctx);
+
+ /* report the apply error */
+ if (apply_error_callback_arg.command != 0)
+ {
+ ErrorData *errdata = CopyErrorData();
+
+ pgstat_report_subworker_error(MyLogicalRepWorker->subid,
+ MyLogicalRepWorker->relid,
+ apply_error_callback_arg.rel != NULL
+ ? apply_error_callback_arg.rel->localreloid
+ : InvalidOid,
+ apply_error_callback_arg.command,
+ apply_error_callback_arg.remote_xid,
+ errdata->message);
+ }
+
+ if (MySubscription->disableonerr)
+ {
+ /* Disable the subscription */
+ DisableSubscriptionOnError();
+ return;
+ }
+
+ MemoryContextSwitchTo(ecxt);
+ PG_RE_THROW();
+ }
+ PG_END_TRY();
+}
+
/* Logical Replication Apply worker entry point */
void
ApplyWorkerMain(Datum main_arg)
@@ -3380,8 +3548,8 @@ ApplyWorkerMain(Datum main_arg)
MemoryContext cctx = CurrentMemoryContext;
MemoryContext oldctx;
char originname[NAMEDATALEN];
- XLogRecPtr origin_startpos;
- char *myslotname;
+ XLogRecPtr origin_startpos = InvalidXLogRecPtr;
+ char *myslotname = NULL;
WalRcvStreamOptions options;
int server_version;
@@ -3475,39 +3643,7 @@ ApplyWorkerMain(Datum main_arg)
MySubscription->conninfo);
if (am_tablesync_worker())
- {
- char *syncslotname;
-
- PG_TRY();
- {
- /* This is table synchronization worker, call initial sync. */
- syncslotname = LogicalRepSyncTableStart(&origin_startpos);
- }
- PG_CATCH();
- {
- MemoryContext ecxt = MemoryContextSwitchTo(cctx);
- ErrorData *errdata = CopyErrorData();
-
- /*
- * Report the table sync error. There is no corresponding message
- * type for table synchronization.
- */
- pgstat_report_subworker_error(MyLogicalRepWorker->subid,
- MyLogicalRepWorker->relid,
- MyLogicalRepWorker->relid,
- 0, /* message type */
- InvalidTransactionId,
- errdata->message);
- MemoryContextSwitchTo(ecxt);
- PG_RE_THROW();
- }
- PG_END_TRY();
-
- /* allocate slot name in long-lived context */
- myslotname = MemoryContextStrdup(ApplyContext, syncslotname);
-
- pfree(syncslotname);
- }
+ SyncTableStartWrapper(&origin_startpos, &myslotname, cctx);
else
{
/* This is main apply worker */
@@ -3619,32 +3755,7 @@ ApplyWorkerMain(Datum main_arg)
}
/* Run the main loop. */
- PG_TRY();
- {
- LogicalRepApplyLoop(origin_startpos);
- }
- PG_CATCH();
- {
- /* report the apply error */
- if (apply_error_callback_arg.command != 0)
- {
- MemoryContext ecxt = MemoryContextSwitchTo(cctx);
- ErrorData *errdata = CopyErrorData();
-
- pgstat_report_subworker_error(MyLogicalRepWorker->subid,
- MyLogicalRepWorker->relid,
- apply_error_callback_arg.rel != NULL
- ? apply_error_callback_arg.rel->localreloid
- : InvalidOid,
- apply_error_callback_arg.command,
- apply_error_callback_arg.remote_xid,
- errdata->message);
- MemoryContextSwitchTo(ecxt);
- }
-
- PG_RE_THROW();
- }
- PG_END_TRY();
+ ApplyLoopWrapper(origin_startpos, cctx);
proc_exit(0);
}
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 4485ea8..46a0deb 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4271,6 +4271,7 @@ getSubscriptions(Archive *fout)
int i_subowner;
int i_substream;
int i_subtwophasestate;
+ int i_subdisableonerr;
int i_subconninfo;
int i_subslotname;
int i_subsynccommit;
@@ -4318,12 +4319,18 @@ getSubscriptions(Archive *fout)
appendPQExpBufferStr(query, " false AS substream,\n");
if (fout->remoteVersion >= 150000)
- appendPQExpBufferStr(query, " s.subtwophasestate\n");
+ appendPQExpBufferStr(query, " s.subtwophasestate,\n");
else
appendPQExpBuffer(query,
- " '%c' AS subtwophasestate\n",
+ " '%c' AS subtwophasestate,\n",
LOGICALREP_TWOPHASE_STATE_DISABLED);
+ if (fout->remoteVersion >= 150000)
+ appendPQExpBuffer(query, " s.subdisableonerr\n");
+ else
+ appendPQExpBuffer(query,
+ " false AS subdisableonerr\n");
+
appendPQExpBufferStr(query,
"FROM pg_subscription s\n"
"WHERE s.subdbid = (SELECT oid FROM pg_database\n"
@@ -4344,6 +4351,7 @@ getSubscriptions(Archive *fout)
i_subbinary = PQfnumber(res, "subbinary");
i_substream = PQfnumber(res, "substream");
i_subtwophasestate = PQfnumber(res, "subtwophasestate");
+ i_subdisableonerr = PQfnumber(res, "subdisableonerr");
subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo));
@@ -4371,6 +4379,8 @@ getSubscriptions(Archive *fout)
pg_strdup(PQgetvalue(res, i, i_substream));
subinfo[i].subtwophasestate =
pg_strdup(PQgetvalue(res, i, i_subtwophasestate));
+ subinfo[i].subdisableonerr =
+ pg_strdup(PQgetvalue(res, i, i_subdisableonerr));
/* Decide whether we want to dump it */
selectDumpableObject(&(subinfo[i].dobj), fout);
@@ -4441,6 +4451,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
if (strcmp(subinfo->subtwophasestate, two_phase_disabled) != 0)
appendPQExpBufferStr(query, ", two_phase = on");
+ if (strcmp(subinfo->subdisableonerr, "f") != 0)
+ appendPQExpBufferStr(query, ", disable_on_error = true");
+
if (strcmp(subinfo->subsynccommit, "off") != 0)
appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit));
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 9965ac2..d7616d3 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -656,6 +656,7 @@ typedef struct _SubscriptionInfo
char *subbinary;
char *substream;
char *subtwophasestate;
+ char *subdisableonerr;
char *subsynccommit;
char *subpublications;
} SubscriptionInfo;
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 654ef2d..cb06f99 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6062,7 +6062,7 @@ describeSubscriptions(const char *pattern, bool verbose)
PGresult *res;
printQueryOpt myopt = pset.popt;
static const bool translate_columns[] = {false, false, false, false,
- false, false, false, false, false};
+ false, false, false, false, false, false};
if (pset.sversion < 100000)
{
@@ -6096,11 +6096,13 @@ describeSubscriptions(const char *pattern, bool verbose)
gettext_noop("Binary"),
gettext_noop("Streaming"));
- /* Two_phase is only supported in v15 and higher */
+ /* Two_phase and disable_on_error are only supported in v15 and higher */
if (pset.sversion >= 150000)
appendPQExpBuffer(&buf,
- ", subtwophasestate AS \"%s\"\n",
- gettext_noop("Two phase commit"));
+ ", subtwophasestate AS \"%s\"\n"
+ ", subdisableonerr AS \"%s\"\n",
+ gettext_noop("Two phase commit"),
+ gettext_noop("Disable on error"));
appendPQExpBuffer(&buf,
", subsynccommit AS \"%s\"\n"
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 010edb6..f7e35ba 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1820,7 +1820,7 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH("(", "PUBLICATION");
/* ALTER SUBSCRIPTION <name> SET ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
- COMPLETE_WITH("binary", "slot_name", "streaming", "synchronous_commit");
+ COMPLETE_WITH("binary", "slot_name", "streaming", "synchronous_commit", "disable_on_error");
/* ALTER SUBSCRIPTION <name> SET PUBLICATION */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "PUBLICATION"))
{
@@ -3079,7 +3079,7 @@ psql_completion(const char *text, int start, int end)
else if (HeadMatches("CREATE", "SUBSCRIPTION") && TailMatches("WITH", "("))
COMPLETE_WITH("binary", "connect", "copy_data", "create_slot",
"enabled", "slot_name", "streaming",
- "synchronous_commit", "two_phase");
+ "synchronous_commit", "two_phase", "disable_on_error");
/* CREATE TRIGGER --- is allowed inside CREATE SCHEMA, so use TailMatches */
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 18c2912..9ebead9 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -67,6 +67,9 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
char subtwophasestate; /* Stream two-phase transactions */
+ bool subdisableonerr; /* True if occurrence of apply errors
+ * should disable the subscription */
+
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* Connection string to the publisher */
text subconninfo BKI_FORCE_NOT_NULL;
@@ -103,6 +106,9 @@ typedef struct Subscription
* binary format */
bool 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
+ * occurs */
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/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 80aae83..ad8003f 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -76,10 +76,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo
------------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+-----------------------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | off | dbname=regress_doesnotexist
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -94,10 +94,10 @@ ERROR: subscription "regress_doesnotexist" does not exist
ALTER SUBSCRIPTION regress_testsub SET (create_slot = false);
ERROR: unrecognized subscription parameter: "create_slot"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+--------------------+------------------------------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | f | d | off | dbname=regress_doesnotexist2
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------------------+------------------------------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | f | d | f | off | dbname=regress_doesnotexist2
(1 row)
BEGIN;
@@ -129,10 +129,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
ERROR: invalid value for parameter "synchronous_commit": "foobar"
HINT: Available values: local, remote_write, remote_apply, on, off.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+--------------------+------------------------------
- regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | f | d | local | dbname=regress_doesnotexist2
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------------------+------------------------------
+ regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | f | d | f | local | dbname=regress_doesnotexist2
(1 row)
-- rename back to keep the rest simple
@@ -165,19 +165,19 @@ ERROR: binary requires a Boolean value
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, binary = true);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo
------------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+-----------------------------
- regress_testsub | regress_subscription_user | f | {testpub} | t | f | d | off | dbname=regress_doesnotexist
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | f | d | f | off | dbname=regress_doesnotexist
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (binary = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo
------------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+-----------------------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | off | dbname=regress_doesnotexist
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -188,19 +188,19 @@ ERROR: streaming requires a Boolean value
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
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo
------------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+-----------------------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | t | d | off | dbname=regress_doesnotexist
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | t | d | f | off | dbname=regress_doesnotexist
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo
------------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+-----------------------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | off | dbname=regress_doesnotexist
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist
(1 row)
-- fail - publication already exists
@@ -215,10 +215,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
ERROR: publication "testpub1" is already in subscription "regress_testsub"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+--------------------+-----------------------------
- regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | f | d | off | dbname=regress_doesnotexist
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
+ regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | f | d | f | off | dbname=regress_doesnotexist
(1 row)
-- fail - publication used more then once
@@ -233,10 +233,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub"
-- ok - delete publications
ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo
------------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+-----------------------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | off | dbname=regress_doesnotexist
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -270,10 +270,10 @@ ERROR: two_phase requires a Boolean value
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo
------------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+-----------------------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | f | p | off | dbname=regress_doesnotexist
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | f | p | f | off | dbname=regress_doesnotexist
(1 row)
--fail - alter of two_phase option not supported.
@@ -282,10 +282,10 @@ ERROR: unrecognized subscription parameter: "two_phase"
-- but can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo
------------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+-----------------------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | t | p | off | dbname=regress_doesnotexist
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | t | p | f | off | dbname=regress_doesnotexist
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -294,10 +294,33 @@ DROP SUBSCRIPTION regress_testsub;
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, streaming = true, two_phase = true);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo
------------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+-----------------------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | t | p | off | dbname=regress_doesnotexist
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | t | p | f | off | dbname=regress_doesnotexist
+(1 row)
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+-- fail - disable_on_error must be boolean
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, disable_on_error = foo);
+ERROR: disable_on_error requires a Boolean value
+-- now it works
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, disable_on_error = false);
+WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist
+(1 row)
+
+ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | t | off | dbname=regress_doesnotexist
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index bd0f4af..dbac8e2 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -228,6 +228,20 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
+-- fail - disable_on_error must be boolean
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, disable_on_error = foo);
+
+-- now it works
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, disable_on_error = false);
+
+\dRs+
+
+ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
+
+\dRs+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+
RESET SESSION AUTHORIZATION;
DROP ROLE regress_subscription_user;
DROP ROLE regress_subscription_user2;
diff --git a/src/test/subscription/t/028_disable_on_error.pl b/src/test/subscription/t/028_disable_on_error.pl
new file mode 100644
index 0000000..f15deb3
--- /dev/null
+++ b/src/test/subscription/t/028_disable_on_error.pl
@@ -0,0 +1,200 @@
+
+# Copyright (c) 2021-2022, PostgreSQL Global Development Group
+
+# Test of logical replication subscription self-disabling feature
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Wait for the named subscriptions to catch up or to be disabled.
+sub wait_for_subscriptions
+{
+ my ($node_name, $dbname, @subscriptions) = @_;
+
+ # Unique-ify the subscriptions passed by the caller
+ my %unique = map { $_ => 1 } @subscriptions;
+ my @unique = sort keys %unique;
+ my $unique_count = scalar(@unique);
+
+ # Construct a SQL list from the unique subscription names
+ my $sublist = join(', ', map { "'$_'" } @unique);
+
+ my $polling_sql = qq(
+ SELECT COUNT(1) = $unique_count FROM
+ (SELECT s.oid
+ FROM pg_catalog.pg_subscription s
+ LEFT JOIN pg_catalog.pg_subscription_rel sr
+ ON sr.srsubid = s.oid
+ WHERE sr.srsubstate IN ('s', 'r')
+ AND s.subname IN ($sublist)
+ AND s.subenabled IS TRUE
+ UNION
+ SELECT s.oid
+ FROM pg_catalog.pg_subscription s
+ WHERE s.subname IN ($sublist)
+ AND s.subenabled IS FALSE
+ ) AS synced_or_disabled
+ );
+ return $node_name->poll_query_until($dbname, $polling_sql);
+}
+
+my @schemas = qw(s1 s2);
+my ($schema, $cmd);
+
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->start;
+
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->start;
+
+# Create identical schema, table and index on both the publisher and
+# subscriber
+for $schema (@schemas)
+{
+ $cmd = qq(
+CREATE SCHEMA $schema;
+CREATE TABLE $schema.tbl (i INT);
+ALTER TABLE $schema.tbl REPLICA IDENTITY FULL;
+CREATE INDEX ${schema}_tbl_idx ON $schema.tbl(i));
+ $node_publisher->safe_psql('postgres', $cmd);
+ $node_subscriber->safe_psql('postgres', $cmd);
+}
+
+# Create non-unique data in both schemas on the publisher.
+for $schema (@schemas)
+{
+ $cmd = qq(INSERT INTO $schema.tbl (i) VALUES (1), (1), (1));
+ $node_publisher->safe_psql('postgres', $cmd);
+}
+
+# Create an additional unique index in schema s1 on the subscriber only.
+# Then, we'll set up 2 pairs of pub/sub, whose names are same as schemas.
+# Publisher "s1" --> Subscriber "s1"
+# Publisher "s2" --> Subscriber "s2"
+# This should cause subscription "s1" on the subscriber to fail during
+# initial synchronization and to get automatically disabled.
+$cmd = qq(CREATE UNIQUE INDEX s1_tbl_unique ON s1.tbl (i));
+$node_subscriber->safe_psql('postgres', $cmd);
+
+# Create publications and subscriptions linking the schemas on
+# the publisher with those on the subscriber. This tests that the
+# uniqueness violations cause subscription "s1" to fail during
+# initial synchronization.
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+for $schema (@schemas)
+{
+ # Create the publication for this table
+ $cmd = qq(
+CREATE PUBLICATION $schema FOR TABLE $schema.tbl);
+ $node_publisher->safe_psql('postgres', $cmd);
+
+ # Create the subscription for this table
+ $cmd = qq(
+CREATE SUBSCRIPTION $schema
+ CONNECTION '$publisher_connstr'
+ PUBLICATION $schema
+ WITH (disable_on_error = true));
+ $node_subscriber->safe_psql('postgres', $cmd);
+}
+
+# Wait for the initial subscription synchronizations to finish or fail.
+wait_for_subscriptions($node_subscriber, 'postgres', @schemas)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+# Subscription "s1" should have disabled itself due to error.
+$cmd = qq(
+SELECT subenabled FROM pg_catalog.pg_subscription WHERE subname = 's1');
+is($node_subscriber->safe_psql('postgres', $cmd),
+ "f", "subscription s1 no longer enabled");
+
+# Subscription "s2" should have copied the initial data without incident.
+$cmd = qq(
+SELECT subenabled FROM pg_catalog.pg_subscription WHERE subname = 's2');
+is($node_subscriber->safe_psql('postgres', $cmd),
+ "t", "subscription s2 still enabled");
+$cmd = qq(SELECT i, COUNT(*) FROM s2.tbl GROUP BY i);
+is($node_subscriber->safe_psql('postgres', $cmd),
+ "1|3", "subscription s2 replicated initial data");
+
+# Enter unique data for both schemas on the publisher. This should succeed on
+# the publisher node, and not cause any additional problems on the subscriber
+# side either, though disabled subscription "s1" should not replicate anything.
+for $schema (@schemas)
+{
+ $cmd = qq(INSERT INTO $schema.tbl (i) VALUES (2));
+ $node_publisher->safe_psql('postgres', $cmd);
+}
+
+# Wait for the data to replicate for the subscriptions. This tests that the
+# problems encountered by subscription "s1" do not cause subscription "s2" to
+# get stuck. Subscription "s1" should still be disabled.
+$cmd = qq(
+SELECT subenabled FROM pg_catalog.pg_subscription WHERE subname = 's1');
+is($node_subscriber->safe_psql('postgres', $cmd),
+ "f", "subscription s1 still disabled");
+
+# Subscription "s2" should still be enabled and have replicated all changes
+$cmd = qq(
+SELECT subenabled FROM pg_catalog.pg_subscription WHERE subname = 's2');
+is($node_subscriber->safe_psql('postgres', $cmd),
+ "t", "subscription s2 still enabled");
+$cmd = q(SELECT COUNT(1) = 1 FROM s2.tbl WHERE i = 2);
+$node_subscriber->poll_query_until('postgres', $cmd)
+ or die "Timed out while waiting for subscriber to apply data";
+
+# Drop the unique index on "s1" which caused the subscription to be disabled
+$cmd = qq(DROP INDEX s1.s1_tbl_unique);
+$node_subscriber->safe_psql('postgres', $cmd);
+
+# Re-enable the subscription "s1"
+$cmd = q(ALTER SUBSCRIPTION s1 ENABLE);
+$node_subscriber->safe_psql('postgres', $cmd);
+
+# Wait for the data to replicate
+wait_for_subscriptions($node_subscriber, 'postgres', @schemas)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+# Check that we have the new data in s1.tbl
+$cmd = q(SELECT MAX(i), COUNT(*) FROM s1.tbl);
+is($node_subscriber->safe_psql('postgres', $cmd),
+ "2|4", "subscription s1 replicated data");
+
+# Delete the data from the subscriber only, and recreate the unique index
+$cmd = q(
+DELETE FROM s1.tbl;
+CREATE UNIQUE INDEX s1_tbl_unique ON s1.tbl (i));
+$node_subscriber->safe_psql('postgres', $cmd);
+
+# Add more non-unique data to the publisher
+for $schema (@schemas)
+{
+ $cmd = qq(INSERT INTO $schema.tbl (i) VALUES (3), (3), (3));
+ $node_publisher->safe_psql('postgres', $cmd);
+}
+
+# Wait for the data to replicate for the subscriptions. This tests that
+# uniqueness violations encountered during replication cause s1 to be disabled.
+$cmd = qq(
+SELECT count(1) = 1 FROM pg_catalog.pg_subscription s
+WHERE s.subname = 's1' AND s.subenabled IS FALSE
+);
+$node_subscriber->poll_query_until('postgres', $cmd)
+ or die "Timed out while waiting for subscription s1 to be disabled";
+
+# Subscription "s2" should have copied the initial data without incident.
+$cmd = qq(
+SELECT subenabled FROM pg_catalog.pg_subscription WHERE subname = 's2');
+is($node_subscriber->safe_psql('postgres', $cmd),
+ "t", "subscription s2 still enabled");
+$cmd = qq(SELECT MAX(i), COUNT(*) FROM s2.tbl);
+is($node_subscriber->safe_psql('postgres', $cmd),
+ "3|7", "subscription s2 replicated additional data");
+
+$node_subscriber->stop;
+$node_publisher->stop;
+
+done_testing();
--
1.8.3.1
view thread (4+ messages)
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: RE: Optionally automatically disable logical replication subscriptions on error
In-Reply-To: <TYCPR01MB837326C879A9CDE756073A7FED3A9@TYCPR01MB8373.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