public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v7 2/7] Implement CLUSTER of partitioned table.. 4+ messages / 3 participants [nested] [flat]
* [PATCH v7 2/7] Implement CLUSTER of partitioned table.. @ 2020-06-07 21:58 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 4+ messages in thread From: Justin Pryzby @ 2020-06-07 21:58 UTC (permalink / raw) This requires either specification of a partitioned index on which to cluster, or that an partitioned index was previously set clustered. --- doc/src/sgml/ref/cluster.sgml | 6 + src/backend/commands/cluster.c | 174 +++++++++++++++++++------- src/bin/psql/tab-complete.c | 1 + src/include/commands/cluster.h | 1 + src/test/regress/expected/cluster.out | 58 ++++++++- src/test/regress/sql/cluster.sql | 24 +++- 6 files changed, 209 insertions(+), 55 deletions(-) diff --git a/doc/src/sgml/ref/cluster.sgml b/doc/src/sgml/ref/cluster.sgml index 5dd21a0189..fb5deddb35 100644 --- a/doc/src/sgml/ref/cluster.sgml +++ b/doc/src/sgml/ref/cluster.sgml @@ -192,6 +192,12 @@ CLUSTER [VERBOSE] are periodically reclustered. </para> + <para> + Clustering a partitioned table clusters each of its partitions using the + index partition of the given partitioned index or (if not specified) the + partitioned index marked as clustered. + </para> + </refsect1> <refsect1> diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 096a06f7b3..9b6673867c 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -32,7 +32,9 @@ #include "catalog/index.h" #include "catalog/namespace.h" #include "catalog/objectaccess.h" +#include "catalog/partition.h" #include "catalog/pg_am.h" +#include "catalog/pg_inherits.h" #include "catalog/toasting.h" #include "commands/cluster.h" #include "commands/defrem.h" @@ -73,6 +75,9 @@ static void copy_table_data(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex, bool verbose, bool *pSwapToastByContent, TransactionId *pFreezeXid, MultiXactId *pCutoffMulti); static List *get_tables_to_cluster(MemoryContext cluster_context); +static List *get_tables_to_cluster_partitioned(MemoryContext cluster_context, + Oid indexOid); +static void cluster_multiple_rels(List *rvs, int options); /*--------------------------------------------------------------------------- @@ -135,7 +140,7 @@ cluster(ParseState *pstate, ClusterStmt *stmt, bool isTopLevel) AccessExclusiveLock, 0, RangeVarCallbackOwnsTable, NULL); - rel = table_open(tableOid, NoLock); + rel = table_open(tableOid, ShareUpdateExclusiveLock); /* * Reject clustering a remote temp table ... their local buffer @@ -146,14 +151,6 @@ cluster(ParseState *pstate, ClusterStmt *stmt, bool isTopLevel) (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("cannot cluster temporary tables of other sessions"))); - /* - * Reject clustering a partitioned table. - */ - if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) - ereport(ERROR, - (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("cannot cluster a partitioned table"))); - if (stmt->indexname == NULL) { ListCell *index; @@ -189,10 +186,34 @@ cluster(ParseState *pstate, ClusterStmt *stmt, bool isTopLevel) } /* close relation, keep lock till commit */ - table_close(rel, NoLock); + table_close(rel, ShareUpdateExclusiveLock); - /* Do the job. */ - cluster_rel(tableOid, indexOid, ¶ms); + if (rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE) + { + /* Do the job. */ + cluster_rel(tableOid, indexOid, ¶ms); + } + else + { + List *rvs; + MemoryContext cluster_context; + + /* Refuse to hold strong locks in a user transaction */ + PreventInTransactionBlock(isTopLevel, "CLUSTER"); + + cluster_context = AllocSetContextCreate(PortalContext, + "Cluster", + ALLOCSET_DEFAULT_SIZES); + + rvs = get_tables_to_cluster_partitioned(cluster_context, indexOid); + cluster_multiple_rels(rvs, params.options); + + /* Start a new transaction for the cleanup work. */ + StartTransactionCommand(); + + /* Clean up working storage */ + MemoryContextDelete(cluster_context); + } } else { @@ -202,7 +223,6 @@ cluster(ParseState *pstate, ClusterStmt *stmt, bool isTopLevel) */ MemoryContext cluster_context; List *rvs; - ListCell *rv; /* * We cannot run this form of CLUSTER inside a user transaction block; @@ -225,28 +245,7 @@ cluster(ParseState *pstate, ClusterStmt *stmt, bool isTopLevel) * cluster_context. */ rvs = get_tables_to_cluster(cluster_context); - - /* Commit to get out of starting transaction */ - PopActiveSnapshot(); - CommitTransactionCommand(); - - /* Ok, now that we've got them all, cluster them one by one */ - foreach(rv, rvs) - { - RelToCluster *rvtc = (RelToCluster *) lfirst(rv); - ClusterParams cluster_params = params; - - /* Start a new transaction for each relation. */ - StartTransactionCommand(); - /* functions in indexes may want a snapshot set */ - PushActiveSnapshot(GetTransactionSnapshot()); - /* Do the job. */ - cluster_params.options |= CLUOPT_RECHECK; - cluster_rel(rvtc->tableOid, rvtc->indexOid, - &cluster_params); - PopActiveSnapshot(); - CommitTransactionCommand(); - } + cluster_multiple_rels(rvs, params.options | CLUOPT_RECHECK_ISCLUSTERED); /* Start a new transaction for the cleanup work. */ StartTransactionCommand(); @@ -352,9 +351,10 @@ cluster_rel(Oid tableOid, Oid indexOid, ClusterParams *params) } /* - * Check that the index is still the one with indisclustered set. + * Check that the index is still the one with indisclustered set, if needed. */ - if (!get_index_isclustered(indexOid)) + if ((params->options & CLUOPT_RECHECK_ISCLUSTERED) != 0 && + !get_index_isclustered(indexOid)) { relation_close(OldHeap, AccessExclusiveLock); pgstat_progress_end_command(); @@ -398,8 +398,13 @@ cluster_rel(Oid tableOid, Oid indexOid, ClusterParams *params) /* Check heap and index are valid to cluster on */ if (OidIsValid(indexOid)) + { check_index_is_clusterable(OldHeap, indexOid, recheck, AccessExclusiveLock); + /* Mark the index as clustered */ + mark_index_clustered(OldHeap, indexOid, true); + } + /* * Quietly ignore the request if this is a materialized view which has not * been populated from its query. No harm is done because there is no data @@ -415,6 +420,14 @@ cluster_rel(Oid tableOid, Oid indexOid, ClusterParams *params) return; } + /* For a partitioned rel, we're done. */ + if (!RELKIND_HAS_STORAGE(get_rel_relkind(tableOid))) + { + relation_close(OldHeap, AccessExclusiveLock); + pgstat_progress_end_command(); + return; + } + /* * All predicate locks on the tuples or pages are about to be made * invalid, because we move tuples around. Promote them to relation @@ -483,6 +496,9 @@ check_index_is_clusterable(Relation OldHeap, Oid indexOid, bool recheck, LOCKMOD * the worst consequence of following broken HOT chains would be that we * might put recently-dead tuples out-of-order in the new table, and there * is little harm in that.) + * + * This also refuses to cluster on an "incomplete" partitioned index + * created with "ON ONLY". */ if (!OldIndex->rd_index->indisvalid) ereport(ERROR, @@ -507,12 +523,6 @@ mark_index_clustered(Relation rel, Oid indexOid, bool is_internal) Relation pg_index; ListCell *index; - /* Disallow applying to a partitioned table */ - if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) - ereport(ERROR, - (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("cannot mark index clustered in partitioned table"))); - /* * If the index is already marked clustered, no need to do anything. */ @@ -584,10 +594,6 @@ rebuild_relation(Relation OldHeap, Oid indexOid, bool verbose) TransactionId frozenXid; MultiXactId cutoffMulti; - /* Mark the correct index as clustered */ - if (OidIsValid(indexOid)) - mark_index_clustered(OldHeap, indexOid, true); - /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; is_system_catalog = IsSystemRelation(OldHeap); @@ -1582,3 +1588,77 @@ get_tables_to_cluster(MemoryContext cluster_context) return rvs; } + +/* + * Return a List of tables and associated index, where each index is a + * partition of the given index + */ +static List * +get_tables_to_cluster_partitioned(MemoryContext cluster_context, Oid indexOid) +{ + List *inhoids; + ListCell *lc; + List *rvs = NIL; + MemoryContext old_context; + + inhoids = find_all_inheritors(indexOid, NoLock, NULL); + + foreach(lc, inhoids) + { + Oid indexrelid = lfirst_oid(lc); + Oid relid = IndexGetRelation(indexrelid, false); + RelToCluster *rvtc; + + /* + * Partitioned rels are also processed by cluster_rel, to + * call check_index_is_clusterable() and mark_index_clustered(). + */ + + /* + * We have to build the list in a different memory context so it will + * survive the cross-transaction processing + */ + old_context = MemoryContextSwitchTo(cluster_context); + + rvtc = (RelToCluster *) palloc(sizeof(RelToCluster)); + rvtc->tableOid = relid; + rvtc->indexOid = indexrelid; + rvs = lappend(rvs, rvtc); + + MemoryContextSwitchTo(old_context); + } + + return rvs; +} + +/* Cluster each relation in a separate transaction */ +static void +cluster_multiple_rels(List *rvs, int options) +{ + ListCell *lc; + + /* Commit to get out of starting transaction */ + PopActiveSnapshot(); + CommitTransactionCommand(); + + /* Ok, now that we've got them all, cluster them one by one */ + foreach(lc, rvs) + { + RelToCluster *rvtc = (RelToCluster *) lfirst(lc); + ClusterParams cluster_params = { .options = options, }; + + /* Start a new transaction for each relation. */ + StartTransactionCommand(); + + /* functions in indexes may want a snapshot set */ + PushActiveSnapshot(GetTransactionSnapshot()); + + /* Do the job. */ + cluster_params.options |= CLUOPT_RECHECK; + cluster_rel(rvtc->tableOid, rvtc->indexOid, + &cluster_params); + + PopActiveSnapshot(); + CommitTransactionCommand(); + } +} diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c index a75647b1cc..781aa95abc 100644 --- a/src/bin/psql/tab-complete.c +++ b/src/bin/psql/tab-complete.c @@ -588,6 +588,7 @@ static const SchemaQuery Query_for_list_of_clusterables = { .catname = "pg_catalog.pg_class c", .selcondition = "c.relkind IN (" CppAsString2(RELKIND_RELATION) ", " + CppAsString2(RELKIND_PARTITIONED_TABLE) ", " CppAsString2(RELKIND_MATVIEW) ")", .viscondition = "pg_catalog.pg_table_is_visible(c.oid)", .namespace = "c.relnamespace", diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index a941f2accd..c30ca01726 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -22,6 +22,7 @@ /* flag bits for ClusterParams->flags */ #define CLUOPT_RECHECK 0x01 /* recheck relation state */ #define CLUOPT_VERBOSE 0x02 /* print progress info */ +#define CLUOPT_RECHECK_ISCLUSTERED 0x04 /* recheck relation state for indisclustered */ /* options for CLUSTER */ typedef struct ClusterParams diff --git a/src/test/regress/expected/cluster.out b/src/test/regress/expected/cluster.out index bdae8fe00c..c74cfa88cc 100644 --- a/src/test/regress/expected/cluster.out +++ b/src/test/regress/expected/cluster.out @@ -439,14 +439,62 @@ select * from clstr_temp; drop table clstr_temp; RESET SESSION AUTHORIZATION; --- Check that partitioned tables cannot be clustered +-- Check that partitioned tables can be clustered CREATE TABLE clstrpart (a int) PARTITION BY RANGE (a); +CREATE TABLE clstrpart1 PARTITION OF clstrpart FOR VALUES FROM (1)TO(10) PARTITION BY RANGE (a); +CREATE TABLE clstrpart11 PARTITION OF clstrpart1 FOR VALUES FROM (1)TO(10); +CREATE TABLE clstrpart12 PARTITION OF clstrpart1 FOR VALUES FROM (10)TO(20) PARTITION BY RANGE(a); +CREATE TABLE clstrpart2 PARTITION OF clstrpart FOR VALUES FROM (20)TO(30); +CREATE TABLE clstrpart3 PARTITION OF clstrpart DEFAULT PARTITION BY RANGE(a); +CREATE TABLE clstrpart33 PARTITION OF clstrpart3 DEFAULT; +ALTER TABLE clstrpart SET WITHOUT CLUSTER; +CREATE INDEX clstrpart_only_idx ON ONLY clstrpart (a); +CLUSTER clstrpart USING clstrpart_only_idx; -- fails +ERROR: cannot cluster on invalid index "clstrpart_only_idx" +DROP INDEX clstrpart_only_idx; CREATE INDEX clstrpart_idx ON clstrpart (a); -ALTER TABLE clstrpart CLUSTER ON clstrpart_idx; -ERROR: cannot mark index clustered in partitioned table +-- Check that clustering sets new relfilenodes: +CREATE TEMP TABLE old_cluster_info AS SELECT relname, level, relfilenode, relkind FROM pg_partition_tree('clstrpart'::regclass) AS tree JOIN pg_class c ON c.oid=tree.relid ; CLUSTER clstrpart USING clstrpart_idx; -ERROR: cannot cluster a partitioned table -DROP TABLE clstrpart; +CREATE TEMP TABLE new_cluster_info AS SELECT relname, level, relfilenode, relkind FROM pg_partition_tree('clstrpart'::regclass) AS tree JOIN pg_class c ON c.oid=tree.relid ; +SELECT relname, old.level, old.relkind, old.relfilenode = new.relfilenode FROM old_cluster_info AS old JOIN new_cluster_info AS new USING (relname) ORDER BY relname COLLATE "C"; + relname | level | relkind | ?column? +-------------+-------+---------+---------- + clstrpart | 0 | p | t + clstrpart1 | 1 | p | t + clstrpart11 | 2 | r | f + clstrpart12 | 2 | p | t + clstrpart2 | 1 | r | f + clstrpart3 | 1 | p | t + clstrpart33 | 2 | r | f +(7 rows) + +-- Check that clustering sets new indisclustered: +SELECT relname, relkind, indisclustered FROM pg_partition_tree('clstrpart_idx'::regclass) AS tree JOIN pg_index i ON i.indexrelid=tree.relid JOIN pg_class c ON c.oid=indexrelid ORDER BY relname COLLATE "C"; + relname | relkind | indisclustered +-------------------+---------+---------------- + clstrpart11_a_idx | i | t + clstrpart12_a_idx | I | t + clstrpart1_a_idx | I | t + clstrpart2_a_idx | i | t + clstrpart33_a_idx | i | t + clstrpart3_a_idx | I | t + clstrpart_idx | I | t +(7 rows) + +CLUSTER clstrpart1 USING clstrpart1_a_idx; -- partition which is itself partitioned +CLUSTER clstrpart12 USING clstrpart12_a_idx; -- partition which is itself partitioned, no childs +CLUSTER clstrpart2 USING clstrpart2_a_idx; -- leaf +\d clstrpart + Partitioned table "public.clstrpart" + Column | Type | Collation | Nullable | Default +--------+---------+-----------+----------+--------- + a | integer | | | +Partition key: RANGE (a) +Indexes: + "clstrpart_idx" btree (a) CLUSTER +Number of partitions: 3 (Use \d+ to list them.) + -- Test CLUSTER with external tuplesorting create table clstr_4 as select * from tenk1; create index cluster_sort on clstr_4 (hundred, thousand, tenthous); diff --git a/src/test/regress/sql/cluster.sql b/src/test/regress/sql/cluster.sql index 188183647c..9bcc77695c 100644 --- a/src/test/regress/sql/cluster.sql +++ b/src/test/regress/sql/cluster.sql @@ -196,12 +196,30 @@ drop table clstr_temp; RESET SESSION AUTHORIZATION; --- Check that partitioned tables cannot be clustered +-- Check that partitioned tables can be clustered CREATE TABLE clstrpart (a int) PARTITION BY RANGE (a); +CREATE TABLE clstrpart1 PARTITION OF clstrpart FOR VALUES FROM (1)TO(10) PARTITION BY RANGE (a); +CREATE TABLE clstrpart11 PARTITION OF clstrpart1 FOR VALUES FROM (1)TO(10); +CREATE TABLE clstrpart12 PARTITION OF clstrpart1 FOR VALUES FROM (10)TO(20) PARTITION BY RANGE(a); +CREATE TABLE clstrpart2 PARTITION OF clstrpart FOR VALUES FROM (20)TO(30); +CREATE TABLE clstrpart3 PARTITION OF clstrpart DEFAULT PARTITION BY RANGE(a); +CREATE TABLE clstrpart33 PARTITION OF clstrpart3 DEFAULT; +ALTER TABLE clstrpart SET WITHOUT CLUSTER; +CREATE INDEX clstrpart_only_idx ON ONLY clstrpart (a); +CLUSTER clstrpart USING clstrpart_only_idx; -- fails +DROP INDEX clstrpart_only_idx; CREATE INDEX clstrpart_idx ON clstrpart (a); -ALTER TABLE clstrpart CLUSTER ON clstrpart_idx; +-- Check that clustering sets new relfilenodes: +CREATE TEMP TABLE old_cluster_info AS SELECT relname, level, relfilenode, relkind FROM pg_partition_tree('clstrpart'::regclass) AS tree JOIN pg_class c ON c.oid=tree.relid ; CLUSTER clstrpart USING clstrpart_idx; -DROP TABLE clstrpart; +CREATE TEMP TABLE new_cluster_info AS SELECT relname, level, relfilenode, relkind FROM pg_partition_tree('clstrpart'::regclass) AS tree JOIN pg_class c ON c.oid=tree.relid ; +SELECT relname, old.level, old.relkind, old.relfilenode = new.relfilenode FROM old_cluster_info AS old JOIN new_cluster_info AS new USING (relname) ORDER BY relname COLLATE "C"; +-- Check that clustering sets new indisclustered: +SELECT relname, relkind, indisclustered FROM pg_partition_tree('clstrpart_idx'::regclass) AS tree JOIN pg_index i ON i.indexrelid=tree.relid JOIN pg_class c ON c.oid=indexrelid ORDER BY relname COLLATE "C"; +CLUSTER clstrpart1 USING clstrpart1_a_idx; -- partition which is itself partitioned +CLUSTER clstrpart12 USING clstrpart12_a_idx; -- partition which is itself partitioned, no childs +CLUSTER clstrpart2 USING clstrpart2_a_idx; -- leaf +\d clstrpart -- Test CLUSTER with external tuplesorting -- 2.17.0 --AA9g+nFNFPYNJKiL Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v7-0003-Propagate-changes-to-indisclustered-to-child-pare.patch" ^ permalink raw reply [nested|flat] 4+ messages in thread
* RE: Optionally automatically disable logical replication subscriptions on error @ 2022-01-05 12:53 [email protected] <[email protected]> 0 siblings, 1 reply; 4+ messages in thread From: [email protected] @ 2022-01-05 12:53 UTC (permalink / raw) To: [email protected] <[email protected]>; +Cc: vignesh C <[email protected]>; 'Greg Nancarrow' <[email protected]>; Amit Kapila <[email protected]>; Masahiko Sawada <[email protected]>; Mark Dilger <[email protected]>; Smith, Peter <[email protected]>; PostgreSQL Hackers <[email protected]> On Tuesday, December 28, 2021 11:53 AM Wang, Wei/王 威 <[email protected]> wrote: > On Thursday, December 16, 2021 8:51 PM [email protected] > <[email protected]> wrote: > > Attached the updated patch v14. > > A comment to the timing of printing a log: Thank you for your review ! > After the log[1] was printed, I altered subscription's option > (DISABLE_ON_ERROR) from true to false before invoking > DisableSubscriptionOnError to disable subscription. Subscription was not > disabled. > [1] "LOG: logical replication subscription "sub1" will be disabled due to an > error" > > I found this log is printed in function WorkerErrorRecovery: > + ereport(LOG, > + errmsg("logical replication subscription \"%s\" will > be disabled due to an error", > + MySubscription->name)); > This log is printed here, but in DisableSubscriptionOnError, there is a check to > confirm subscription's disableonerr field. If disableonerr is found changed from > true to false in DisableSubscriptionOnError, subscription will not be disabled. > > In this case, "disable subscription" is printed, but subscription will not be > disabled actually. > I think it is a little confused to user, so what about moving this message after > the check which is mentioned above in DisableSubscriptionOnError? Makes sense. I moved the log print after the check of the necessity to disable the subscription. Also, I've scrutinized and refined the new TAP test as well for refactoring. As a result, I fixed wait_for_subscriptions() so that some extra codes that can be simplified, such as escaped variable and one part of WHERE clause, are removed. Other change I did is to replace two calls of wait_for_subscriptions() with polling_query_until() for the subscriber, in order to make the tests better and more suitable for the test purposes. Again, for this refinement, I've conducted a tight loop test to check no timing issue and found no problem. Best Regards, Takamichi Osumi Attachments: [application/octet-stream] v15-0001-Optionally-disable-subscriptions-on-error.patch (48.7K, ../../TYCPR01MB83732A590139C59DE0A83CFBED4B9@TYCPR01MB8373.jpnprd01.prod.outlook.com/2-v15-0001-Optionally-disable-subscriptions-on-error.patch) download | inline diff: From 89128db3109b02ccfad88139960d67febe4e3cfa Mon Sep 17 00:00:00 2001 From: Takamichi Osumi <[email protected]> Date: Wed, 5 Jan 2022 11:36:09 +0000 Subject: [PATCH v15] 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, adding 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. Require to bump catalog version. Proposed and written originally by Mark Dilger Taken over by Osumi Takamichi, Greg Nancarrow Reviewed by Greg Nancarrow, Vignesh C, Amit Kapila, Wang wei 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 | 140 ++++++++++++++++- 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 | 4 + src/test/regress/expected/subscription.out | 119 ++++++++------ src/test/regress/sql/subscription.sql | 14 ++ src/test/subscription/t/027_disable_on_error.pl | 196 ++++++++++++++++++++++++ 15 files changed, 493 insertions(+), 69 deletions(-) create mode 100644 src/test/subscription/t/027_disable_on_error.pl diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml index 03e2537..f89257d 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -7725,6 +7725,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..3109ee9 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 25021e2..9b416dd 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 61b515c..9d0ed61 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1259,7 +1259,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 2b65808..cebc51a 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 2e79302..efd9767 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" @@ -2755,6 +2756,93 @@ LogicalRepApplyLoop(XLogRecPtr last_received) } /* + * Worker error recovery processing, in preparation for disabling the + * subscription. + */ +static void +WorkerErrorRecovery(void) +{ + /* Emit the error */ + EmitErrorReport(); + /* Abort any active transaction */ + AbortOutOfAnyTransaction(); + /* Reset the ErrorContext */ + FlushErrorState(); +} + +/* + * Disable the current subscription. + */ +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; + + /* 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) + { + /* + * Disabling the subscription has been done already. No need of + * additional work. + */ + 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, and note the reason. */ + 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 @@ -3339,6 +3427,7 @@ ApplyWorkerMain(Datum main_arg) char *myslotname; WalRcvStreamOptions options; int server_version; + bool error_recovery_done = false; /* Attach to slot */ logicalrep_worker_attach(worker_slot); @@ -3453,15 +3542,33 @@ ApplyWorkerMain(Datum main_arg) 0, /* message type */ InvalidTransactionId, errdata->message); - MemoryContextSwitchTo(ecxt); - PG_RE_THROW(); + + if (MySubscription->disableonerr) + { + WorkerErrorRecovery(); + error_recovery_done = true; + } + else + { + MemoryContextSwitchTo(ecxt); + PG_RE_THROW(); + } } PG_END_TRY(); - /* allocate slot name in long-lived context */ - myslotname = MemoryContextStrdup(ApplyContext, syncslotname); + /* If we caught an error above, disable the subscription */ + if (error_recovery_done) + { + DisableSubscriptionOnError(); + proc_exit(0); + } + else + { + /* allocate slot name in long-lived context */ + myslotname = MemoryContextStrdup(ApplyContext, syncslotname); - pfree(syncslotname); + pfree(syncslotname); + } } else { @@ -3580,10 +3687,11 @@ ApplyWorkerMain(Datum main_arg) } PG_CATCH(); { + MemoryContext ecxt = MemoryContextSwitchTo(cctx); + /* report the apply error */ if (apply_error_callback_arg.command != 0) { - MemoryContext ecxt = MemoryContextSwitchTo(cctx); ErrorData *errdata = CopyErrorData(); pgstat_report_subworker_error(MyLogicalRepWorker->subid, @@ -3594,13 +3702,29 @@ ApplyWorkerMain(Datum main_arg) apply_error_callback_arg.command, apply_error_callback_arg.remote_xid, errdata->message); - MemoryContextSwitchTo(ecxt); + + if (!MySubscription->disableonerr) + { + /* + * Some work in error recovery work is done. Switch to the old + * memory context and rethrow. + */ + MemoryContextSwitchTo(ecxt); + PG_RE_THROW(); + } } + else if (!MySubscription->disableonerr) + PG_RE_THROW(); - PG_RE_THROW(); + /* Prepare to disable the subscription */ + WorkerErrorRecovery(); + error_recovery_done = true; } PG_END_TRY(); + if (error_recovery_done) + DisableSubscriptionOnError(); + proc_exit(0); } diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index 4fec88b..2d3d91a 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -4239,6 +4239,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; @@ -4286,12 +4287,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" @@ -4312,6 +4319,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)); @@ -4339,6 +4347,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); @@ -4409,6 +4419,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"); + 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 f9deb32..3912a31 100644 --- a/src/bin/pg_dump/pg_dump.h +++ b/src/bin/pg_dump/pg_dump.h @@ -655,6 +655,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 c28788e..d625b9b 100644 --- a/src/bin/psql/describe.c +++ b/src/bin/psql/describe.c @@ -6021,7 +6021,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) { @@ -6055,11 +6055,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 */ 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 b81a04c..6442fc5 100644 --- a/src/bin/psql/tab-complete.c +++ b/src/bin/psql/tab-complete.c @@ -1690,7 +1690,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")) { @@ -2950,7 +2950,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 2106149..7b76842 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,7 @@ typedef struct Subscription * binary format */ bool stream; /* Allow streaming in-progress transactions. */ char twophasestate; /* Allow streaming two-phase transactions */ + bool disableonerr; /* Whether errors automatically disable */ 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/027_disable_on_error.pl b/src/test/subscription/t/027_disable_on_error.pl new file mode 100644 index 0000000..8838805 --- /dev/null +++ b/src/test/subscription/t/027_disable_on_error.pl @@ -0,0 +1,196 @@ + +# Copyright (c) 2021, 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 tests => 8; + +# 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. When +# we create subscriptions, below, 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; -- 1.8.3.1 ^ permalink raw reply [nested|flat] 4+ messages in thread
* RE: Optionally automatically disable logical replication subscriptions on error @ 2022-01-06 03:16 [email protected] <[email protected]> parent: [email protected] <[email protected]> 0 siblings, 1 reply; 4+ messages in thread From: [email protected] @ 2022-01-06 03:16 UTC (permalink / raw) To: [email protected] <[email protected]>; [email protected] <[email protected]>; +Cc: vignesh C <[email protected]>; 'Greg Nancarrow' <[email protected]>; Amit Kapila <[email protected]>; Masahiko Sawada <[email protected]>; Mark Dilger <[email protected]>; Smith, Peter <[email protected]>; PostgreSQL Hackers <[email protected]> On Wednesday, January 5, 2022 8:53 PM [email protected] <[email protected]> wrote: > > On Tuesday, December 28, 2021 11:53 AM Wang, Wei/王 威 > <[email protected]> wrote: > > On Thursday, December 16, 2021 8:51 PM [email protected] > > <[email protected]> wrote: > > > Attached the updated patch v14. > > > > A comment to the timing of printing a log: > Thank you for your review ! > > > After the log[1] was printed, I altered subscription's option > > (DISABLE_ON_ERROR) from true to false before invoking > > DisableSubscriptionOnError to disable subscription. Subscription was not > > disabled. > > [1] "LOG: logical replication subscription "sub1" will be disabled due to an > > error" > > > > I found this log is printed in function WorkerErrorRecovery: > > + ereport(LOG, > > + errmsg("logical replication subscription \"%s\" will > > be disabled due to an error", > > + MySubscription->name)); > > This log is printed here, but in DisableSubscriptionOnError, there is a check to > > confirm subscription's disableonerr field. If disableonerr is found changed from > > true to false in DisableSubscriptionOnError, subscription will not be disabled. > > > > In this case, "disable subscription" is printed, but subscription will not be > > disabled actually. > > I think it is a little confused to user, so what about moving this message after > > the check which is mentioned above in DisableSubscriptionOnError? > Makes sense. I moved the log print after > the check of the necessity to disable the subscription. > > Also, I've scrutinized and refined the new TAP test as well for refactoring. > As a result, I fixed wait_for_subscriptions() > so that some extra codes that can be simplified, > such as escaped variable and one part of WHERE clause, are removed. > Other change I did is to replace two calls of wait_for_subscriptions() > with polling_query_until() for the subscriber, in order to > make the tests better and more suitable for the test purposes. > Again, for this refinement, I've conducted a tight loop test > to check no timing issue and found no problem. > Thanks for updating the patch. Here are some comments: 1) + /* + * 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) + { + /* + * Disabling the subscription has been done already. No need of + * additional work. + */ + heap_freetuple(tup); + table_close(rel, RowExclusiveLock); + CommitTransactionCommand(); + return; + } I don't understand what does "Disabling the subscription has been done already" mean, I think we only run here when subdisableonerr is changed in the interim. Should we modify this comment? Or remove it because there are already some explanations before. 2) + /* Set the subscription to disabled, and note the reason. */ + values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(false); + replaces[Anum_pg_subscription_subenabled - 1] = true; I didn't see the code corresponding to "note the reason". Should we modify the comment? 3) + bool disableonerr; /* Whether errors automatically disable */ This comment is hard to understand. Maybe it can be changed to: Indicates if the subscription should be automatically disabled when subscription workers detect any errors. Regards, Tang ^ permalink raw reply [nested|flat] 4+ messages in thread
* RE: Optionally automatically disable logical replication subscriptions on error @ 2022-01-06 05:53 [email protected] <[email protected]> parent: [email protected] <[email protected]> 0 siblings, 0 replies; 4+ messages in thread From: [email protected] @ 2022-01-06 05:53 UTC (permalink / raw) To: [email protected] <[email protected]>; [email protected] <[email protected]>; +Cc: vignesh C <[email protected]>; 'Greg Nancarrow' <[email protected]>; Amit Kapila <[email protected]>; Masahiko Sawada <[email protected]>; Mark Dilger <[email protected]>; Smith, Peter <[email protected]>; PostgreSQL Hackers <[email protected]> On Thursday, January 6, 2022 12:17 PM Tang, Haiying/唐 海英 <[email protected]> wrote: > Thanks for updating the patch. Here are some comments: Thank you for your review ! > 1) > + /* > + * 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) > + { > + /* > + * Disabling the subscription has been done already. No need > of > + * additional work. > + */ > + heap_freetuple(tup); > + table_close(rel, RowExclusiveLock); > + CommitTransactionCommand(); > + return; > + } > > I don't understand what does "Disabling the subscription has been done > already" > mean, I think we only run here when subdisableonerr is changed in the interim. > Should we modify this comment? Or remove it because there are already some > explanations before. Removed. The description you pointed out was redundant. > 2) > + /* Set the subscription to disabled, and note the reason. */ > + values[Anum_pg_subscription_subenabled - 1] = > BoolGetDatum(false); > + replaces[Anum_pg_subscription_subenabled - 1] = true; > > I didn't see the code corresponding to "note the reason". Should we modify the > comment? Fixed the comment by removing the part. We come here when an error occurred and the reason is printed as log so no need to note more reason. > 3) > + bool disableonerr; /* Whether errors automatically > disable */ > > This comment is hard to understand. Maybe it can be changed to: > > Indicates if the subscription should be automatically disabled when > subscription workers detect any errors. Agreed. Fixed. Kindly have a look at the attached v16. Best Regards, Takamichi Osumi Attachments: [application/octet-stream] v16-0001-Optionally-disable-subscriptions-on-error.patch (48.7K, ../../TYCPR01MB8373EEAA0EC0D45EC442F1C9ED4C9@TYCPR01MB8373.jpnprd01.prod.outlook.com/2-v16-0001-Optionally-disable-subscriptions-on-error.patch) download | inline diff: From 60f0009fec206b654233efadd9942c25ecee0285 Mon Sep 17 00:00:00 2001 From: Takamichi Osumi <[email protected]> Date: Thu, 6 Jan 2022 05:13:04 +0000 Subject: [PATCH v16] 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, adding 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. Require to bump catalog version. 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 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 | 136 +++++++++++++++- 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/027_disable_on_error.pl | 196 ++++++++++++++++++++++++ 15 files changed, 491 insertions(+), 69 deletions(-) create mode 100644 src/test/subscription/t/027_disable_on_error.pl diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml index 03e2537..f89257d 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -7725,6 +7725,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..3109ee9 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 25021e2..9b416dd 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 61b515c..9d0ed61 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1259,7 +1259,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 2b65808..cebc51a 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 2e79302..16376df 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" @@ -2755,6 +2756,89 @@ LogicalRepApplyLoop(XLogRecPtr last_received) } /* + * Worker error recovery processing, in preparation for disabling the + * subscription. + */ +static void +WorkerErrorRecovery(void) +{ + /* Emit the error */ + EmitErrorReport(); + /* Abort any active transaction */ + AbortOutOfAnyTransaction(); + /* Reset the ErrorContext */ + FlushErrorState(); +} + +/* + * Disable the current subscription. + */ +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; + + /* 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 @@ -3339,6 +3423,7 @@ ApplyWorkerMain(Datum main_arg) char *myslotname; WalRcvStreamOptions options; int server_version; + bool error_recovery_done = false; /* Attach to slot */ logicalrep_worker_attach(worker_slot); @@ -3453,15 +3538,33 @@ ApplyWorkerMain(Datum main_arg) 0, /* message type */ InvalidTransactionId, errdata->message); - MemoryContextSwitchTo(ecxt); - PG_RE_THROW(); + + if (MySubscription->disableonerr) + { + WorkerErrorRecovery(); + error_recovery_done = true; + } + else + { + MemoryContextSwitchTo(ecxt); + PG_RE_THROW(); + } } PG_END_TRY(); - /* allocate slot name in long-lived context */ - myslotname = MemoryContextStrdup(ApplyContext, syncslotname); + /* If we caught an error above, disable the subscription */ + if (error_recovery_done) + { + DisableSubscriptionOnError(); + proc_exit(0); + } + else + { + /* allocate slot name in long-lived context */ + myslotname = MemoryContextStrdup(ApplyContext, syncslotname); - pfree(syncslotname); + pfree(syncslotname); + } } else { @@ -3580,10 +3683,11 @@ ApplyWorkerMain(Datum main_arg) } PG_CATCH(); { + MemoryContext ecxt = MemoryContextSwitchTo(cctx); + /* report the apply error */ if (apply_error_callback_arg.command != 0) { - MemoryContext ecxt = MemoryContextSwitchTo(cctx); ErrorData *errdata = CopyErrorData(); pgstat_report_subworker_error(MyLogicalRepWorker->subid, @@ -3594,13 +3698,29 @@ ApplyWorkerMain(Datum main_arg) apply_error_callback_arg.command, apply_error_callback_arg.remote_xid, errdata->message); - MemoryContextSwitchTo(ecxt); + + if (!MySubscription->disableonerr) + { + /* + * Some work in error recovery work is done. Switch to the old + * memory context and rethrow. + */ + MemoryContextSwitchTo(ecxt); + PG_RE_THROW(); + } } + else if (!MySubscription->disableonerr) + PG_RE_THROW(); - PG_RE_THROW(); + /* Prepare to disable the subscription */ + WorkerErrorRecovery(); + error_recovery_done = true; } PG_END_TRY(); + if (error_recovery_done) + DisableSubscriptionOnError(); + proc_exit(0); } diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index 4fec88b..2d3d91a 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -4239,6 +4239,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; @@ -4286,12 +4287,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" @@ -4312,6 +4319,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)); @@ -4339,6 +4347,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); @@ -4409,6 +4419,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"); + 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 f9deb32..3912a31 100644 --- a/src/bin/pg_dump/pg_dump.h +++ b/src/bin/pg_dump/pg_dump.h @@ -655,6 +655,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 c28788e..d625b9b 100644 --- a/src/bin/psql/describe.c +++ b/src/bin/psql/describe.c @@ -6021,7 +6021,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) { @@ -6055,11 +6055,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 */ 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 b81a04c..6442fc5 100644 --- a/src/bin/psql/tab-complete.c +++ b/src/bin/psql/tab-complete.c @@ -1690,7 +1690,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")) { @@ -2950,7 +2950,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 2106149..66c8015 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 when subscription + * workers detect any errors. */ 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/027_disable_on_error.pl b/src/test/subscription/t/027_disable_on_error.pl new file mode 100644 index 0000000..8838805 --- /dev/null +++ b/src/test/subscription/t/027_disable_on_error.pl @@ -0,0 +1,196 @@ + +# Copyright (c) 2021, 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 tests => 8; + +# 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. When +# we create subscriptions, below, 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; -- 1.8.3.1 ^ permalink raw reply [nested|flat] 4+ messages in thread
end of thread, other threads:[~2022-01-06 05:53 UTC | newest] Thread overview: 4+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2020-06-07 21:58 [PATCH v7 2/7] Implement CLUSTER of partitioned table.. Justin Pryzby <[email protected]> 2022-01-05 12:53 RE: Optionally automatically disable logical replication subscriptions on error [email protected] <[email protected]> 2022-01-06 03:16 ` RE: Optionally automatically disable logical replication subscriptions on error [email protected] <[email protected]> 2022-01-06 05:53 ` RE: Optionally automatically disable logical replication subscriptions on error [email protected] <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox