Received: from malur.postgresql.org ([217.196.149.56]) by arkaria.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.92) (envelope-from ) id 1luOSt-00025f-6R for pgsql-hackers@arkaria.postgresql.org; Sat, 19 Jun 2021 00:03:59 +0000 Received: from localhost ([127.0.0.1] helo=malur.postgresql.org) by malur.postgresql.org with esmtp (Exim 4.92) (envelope-from ) id 1luOSs-0000Xm-24 for pgsql-hackers@arkaria.postgresql.org; Sat, 19 Jun 2021 00:03:58 +0000 Received: from magus.postgresql.org ([2a02:c0:301:0:ffff::29]) by malur.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.92) (envelope-from ) id 1luOSr-0000Xe-KB for pgsql-hackers@lists.postgresql.org; Sat, 19 Jun 2021 00:03:57 +0000 Received: from mail-pf1-x431.google.com ([2607:f8b0:4864:20::431]) by magus.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_128_GCM_SHA256:128) (Exim 4.92) (envelope-from ) id 1luOSk-0003XJ-7g for pgsql-hackers@lists.postgresql.org; Sat, 19 Jun 2021 00:03:57 +0000 Received: by mail-pf1-x431.google.com with SMTP id p13so9021333pfw.0 for ; Fri, 18 Jun 2021 17:03:49 -0700 (PDT) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=enterprisedb-com.20150623.gappssmtp.com; s=20150623; h=from:message-id:mime-version:subject:date:in-reply-to:cc:to :references; bh=bH/iY0KjTWSNC9OcoIF4dXvVO1XWFgUNbNAyO0O+T8A=; b=c9awapWVWMBpnIMEOu6gKCR8klT2gADhtTOWh8tcMbIV/CFZMpo38tp7zHvfKc4tWf hVfM0G5cSlBlzuwh+hNTFN6TBn7+TRQsw5iZvi0THW/bYd7AF7Z1kOUVLV9/3Ao2AVPq G0srsYjPkK/fvEX3HdwArwGPk2hKJ/qhESkhStGEgkD9Vi++T+vjBFx+RLRGKTGSY/Qa L6E6FIizG5vBHihZBFSchfVJ6//Kl0MXmUwRYkC6BlrsyD8tSIv1Kti3sGw1cMD8gNke G8t9KuiWCYXqgBPVJj9/wUu6W67W/AuCzn9bCScTNkhoL+DyKGSUhIFMHQCIggm6pv5A L4bw== X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20161025; h=x-gm-message-state:from:message-id:mime-version:subject:date :in-reply-to:cc:to:references; bh=bH/iY0KjTWSNC9OcoIF4dXvVO1XWFgUNbNAyO0O+T8A=; b=hgi2Z56q+uNLZyAOuXm9tczZGP6TzJ1/sjsGRUF2ahyPdXhWIuXJjsJ3oon6oC5SfV 8NFDJy/s5k6MYd0PDlKJfTQSQfXvdmjYGwA+Dj2kujNUlKwdiiwDxORXzgG7bM+Yd9Nl GN6AHE/HC1pCAQehwOwwpn7wenen+0bG+1JWQk4Jeak/sHZ0Rtjcv3MF6piMPxJKm2IO 7YVhPaE6k6OedCqlBItuLmBT/X0Qay95yY82GtUCvj4djqffjW18wbCpyHQut/NwOHgc Xi5i9CcCCNJp1fOaJE+sznb8JM9aLVTowdbL19W8Irh9gaSNDetiDvff3pAhklnBpERM OMjw== X-Gm-Message-State: AOAM531MF1wc2AgS1gQEf1YaeWRzpfTYkJrY6MtjOUSQGq3PeG3qDdW7 /3WxdJ6rrXOGJZ1Kks03B1WqTwexW30lGh3VsN8yAQw7MyMaMaW/rOz61+ocaY8oIt5w3H7GsNX iHfRWiA3r6kN4SfpwnQ84dsImkSSMkGi0B3ydhwfcjwrhbvFDjcaj5KK2yrwGJ0aFShXC613bYq BhNLa3SHaDhUmO0L9dU+ghzr5of8ErKWwxEZEPesXuXcdpyZ3vyV5y X-Google-Smtp-Source: ABdhPJzhezJ3m2KNLmzuH82oRHeSJl9iD5oSg51TY8ihaXI5gngg9DVkRtAq01qPDa4b6PR3OwNijA== X-Received: by 2002:a63:2ac4:: with SMTP id q187mr12297962pgq.370.1624061027660; Fri, 18 Jun 2021 17:03:47 -0700 (PDT) Received: from ?IPv6:2600:6c50:157f:eb87:7450:3329:5fae:6f3e? ([2600:6c50:157f:eb87:7450:3329:5fae:6f3e]) by smtp.gmail.com with ESMTPSA id t14sm10944978pgm.9.2021.06.18.17.03.46 (version=TLS1_2 cipher=ECDHE-ECDSA-AES128-GCM-SHA256 bits=128/128); Fri, 18 Jun 2021 17:03:47 -0700 (PDT) From: Mark Dilger Message-Id: <38706E50-9190-4623-9FD7-4E9E3DD3F81A@enterprisedb.com> Content-Type: multipart/mixed; boundary="Apple-Mail=_E69A0652-B6B2-4D25-BE6D-3FCC2DC247AB" Mime-Version: 1.0 (Mac OS X Mail 13.4 \(3608.120.23.2.6\)) Subject: Re: Optionally automatically disable logical replication subscriptions on error Date: Fri, 18 Jun 2021 17:03:45 -0700 In-Reply-To: Cc: PostgreSQL Hackers To: Peter Smith , Amit Kapila References: X-Mailer: Apple Mail (2.3608.120.23.2.6) X-CLOUD-SEC-AV-Info: enterprisedb,google_mail,monitor X-CLOUD-SEC-AV-Sent: true X-Gm-Spam: 0 X-Gm-Phishy: 0 List-Id: List-Help: List-Subscribe: List-Post: List-Owner: List-Archive: Archived-At: Precedence: bulk --Apple-Mail=_E69A0652-B6B2-4D25-BE6D-3FCC2DC247AB Content-Transfer-Encoding: quoted-printable Content-Type: text/plain; charset=us-ascii > On Jun 17, 2021, at 11:34 PM, Peter Smith = wrote: >=20 > I tried your patch. Thanks for the quick and thorough review! > (2) New column "disabled_by_error". >=20 > I wondered if there was actually any need for this column. Isn't the > same information conveyed by just having "subenabled" =3D false, at = same > time as as non-empty "suberrmsg"? This would remove any confusion for > having 2 booleans which both indicate disabled. Yeah, I wondered about that before posting v1. I removed the = disabled_by_error field for v2. > (3) New columns "disabled_by_error", "disabled_on_error". >=20 > All other columns of the pg_subscription have a "sub" prefix. I don't feel strongly about this. How about "subdisableonerr"? I used = that in v2. > I did not find any code using that newly added member "errhint". Thanks for catching that. I had tried to remove all references to = "errhint" before posting v1. The original idea was that both the = message and hint of the error would be kept, but in testing I found the = hint field was typically empty, so I removed it. Sorry that I left one = mention of it lying around. > (5) dump.c I didn't bother getting pg_dump working before posting v1, and I still = have not done so, as I mainly want to solicit feedback on whether the = basic direction I am going will work for the community. > (6) Patch only handles errors only from the Apply worker. >=20 > Tablesync can give similar errors (e.g. PK violation during DATASYNC > phase) which will trigger re-launch forever regardless of the setting > of "disabled_on_error". > (confirmed by observations below) Yes, this is a good point, and also mentioned by Amit. I have fixed it = in v2 and adjusted the regression test to trigger an automatic disabling = for initial table sync as well as for change replication. > 2021-06-18 15:12:45.905 AEST [25904] LOG: edata is true for > subscription 'tap_sub': message =3D "duplicate key value violates = unique > constraint "test_tab_pkey"", hint =3D "" You didn't call this out, but FYI, I don't intend to leave this = particular log message in the patch. It was for development only. I = have removed it for v2 and have added a different log message much = sooner after catching the error, to avoid squashing the error in case = some other action fails. The regression test shows this, if you open = tmp_check/log/022_disable_on_error_subscriber.log: 2021-06-18 16:25:20.138 PDT [56926] LOG: logical replication = subscription "s1" will be disabled due to error: duplicate key value = violates unique constraint "s1_tbl_unique" 2021-06-18 16:25:20.139 PDT [56926] ERROR: duplicate key value violates = unique constraint "s1_tbl_unique" 2021-06-18 16:25:20.139 PDT [56926] DETAIL: Key (i)=3D(1) already = exists. 2021-06-18 16:25:20.139 PDT [56926] CONTEXT: COPY tbl, line 2 The first line logs the error prior to attempting to disable the = subscription, and the next three lines are due to rethrowing the error = after committing the successful disabling of the subscription. If the = attempt to disable the subscription itself throws, these additional = three lines won't show up, but the first one should. Amit mentioned = this upthread. Do you think this will be ok, or would you like to also = have a suberrdetail field so that the detail doesn't get lost? I = haven't added such an extra field, and am inclined to think it would be = excessive, but maybe others feel differently? > =3D=3D=3D=3D=3D=3D >=20 > Test: Cause a PK violation in the Tablesync copy (DATASYNC) phase. > (when disable_on_error =3D true) > Observation: This patch changes nothing for this case. The Tablesyn > re-launchs in a forever loop the same as current functionality. In v2, tablesync copy errors should also be caught. The test has been = extended to cover this also. --Apple-Mail=_E69A0652-B6B2-4D25-BE6D-3FCC2DC247AB Content-Disposition: attachment; filename=v2-0001-Optionally-disabling-subscriptions-on-error.patch Content-Type: application/octet-stream; x-unix-mode=0644; name="v2-0001-Optionally-disabling-subscriptions-on-error.patch" Content-Transfer-Encoding: quoted-printable =46rom=2084cdb2bf2aef276255971be47cf48c6dde7e0564=20Mon=20Sep=2017=20= 00:00:00=202001=0AFrom:=20Mark=20Dilger=20=0A= Date:=20Thu,=2017=20Jun=202021=2012:39:19=20-0700=0ASubject:=20[PATCH=20= v2]=20Optionally=20disabling=20subscriptions=20on=20error=0A=0ALogical=20= replication=20apply=20workers=20for=20a=20subscription=20can=20easily=20= get=0Astuck=20in=20an=20infinite=20loop=20of=20attempting=20to=20apply=20= a=20change,=0Atriggering=20an=20error=20(such=20as=20a=20constraint=20= violation),=20exiting=20with=0Aan=20error=20written=20to=20the=20= subscription=20worker=20log,=20and=20restarting.=0A=0ATo=20partially=20= remedy=20the=20situation,=20adding=20a=20new=0Asubscription_parameter=20= named=20'disable_on_error'.=20=20To=20be=20consistent=0Awith=20old=20= behavior,=20the=20parameter=20defaults=20to=20false.=20=20When=20true,=20= the=0Aapply=20worker=20catches=20errors=20thrown,=20and=20for=20errors=20= that=20are=20deemed=0Anot=20to=20be=20transient,=20disables=20the=20= subscription=20in=20order=20to=20break=20the=0Aloop.=20=20A=20new=20= column=20in=20the=20pg_subscription=20table=20helps=20diagnose=20the=0A= situation:=20=20when=20this=20has=20occurred,=20'suberrmsg'=20includes=20= the=20message=0Afield=20of=20the=20error.=20=20The=20error=20is=20still=20= also=20written=20to=20the=20logs.=0A=0AIn=20addition=20to=20helping=20on=20= production=20systems,=20this=20makes=20writing=20TAP=0Atests=20involving=20= error=20conditions=20simpler.=20=20Rather=20than=20having=20to=0Aopen=20= and=20parse=20the=20apply=20worker's=20log=20file,=20the=20test=20can=20= query=20the=0Apg_subscription=20table.=20=20It=20also=20helps=20that=20= the=20workers=20don't=20go=20into=0Aan=20infinite=20loop=20during=20the=20= test.=0A---=0A=20doc/src/sgml/catalogs.sgml=20=20=20=20=20=20=20=20=20=20= =20=20=20=20=20=20=20=20=20=20|=20=2010=20+=0A=20= doc/src/sgml/ref/create_subscription.sgml=20=20=20=20=20|=20=2012=20+=0A=20= src/backend/catalog/pg_subscription.c=20=20=20=20=20=20=20=20=20|=20=20=20= 9=20+=0A=20src/backend/catalog/system_views.sql=20=20=20=20=20=20=20=20=20= =20|=20=20=208=20+-=0A=20src/backend/commands/subscriptioncmds.c=20=20=20= =20=20=20=20|=20=2057=20+++++=0A=20= src/backend/replication/logical/launcher.c=20=20=20=20|=20=20=201=20+=0A=20= src/backend/replication/logical/worker.c=20=20=20=20=20=20|=20189=20= +++++++++++++++-=0A=20src/bin/pg_dump/pg_dump.c=20=20=20=20=20=20=20=20=20= =20=20=20=20=20=20=20=20=20=20=20=20|=20=20=206=20+-=0A=20= src/bin/pg_dump/pg_dump.h=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20= =20=20=20=20=20|=20=20=201=20+=0A=20= src/include/catalog/pg_subscription.h=20=20=20=20=20=20=20=20=20|=20=20=20= 8=20+=0A=20src/test/perl/PostgresNode.pm=20=20=20=20=20=20=20=20=20=20=20= =20=20=20=20=20=20|=20=2040=20++++=0A=20= .../subscription/t/022_disable_on_error.pl=20=20=20=20|=20205=20= ++++++++++++++++++=0A=2012=20files=20changed,=20540=20insertions(+),=206=20= deletions(-)=0A=20create=20mode=20100644=20= src/test/subscription/t/022_disable_on_error.pl=0A=0Adiff=20--git=20= a/doc/src/sgml/catalogs.sgml=20b/doc/src/sgml/catalogs.sgml=0Aindex=20= f517a7d4af..6fd95cb7ca=20100644=0A---=20a/doc/src/sgml/catalogs.sgml=0A= +++=20b/doc/src/sgml/catalogs.sgml=0A@@=20-7671,6=20+7671,16=20@@=20= SCRAM-SHA-256$<iteration=20= count>:&l=0A=20=20=20=20=20=20=20= =0A=20=20=20=20=20=20=0A=20=0A+=20=20=20=20=20=0A= +=20=20=20=20=20=20=0A+=20=20=20=20=20=20=20= suberrmsg=20text=0A+=20=20=20=20=20= =20=0A+=20=20=20=20=20=20=0A+=20=20=20=20=20=20=20The=20= message=20from=20the=20error=20which=20disabled=20the=20subscription,=20= if=20it=20has=0A+=20=20=20=20=20=20=20been=20automatically=20disabled.=0A= +=20=20=20=20=20=20=0A+=20=20=20=20=20=0A+=0A=20=20=20= =20=20=20=0A=20=20=20=20=20=20=20=0A=20=20= =20=20=20=20=20=20subpublications=20= text[]=0Adiff=20--git=20= a/doc/src/sgml/ref/create_subscription.sgml=20= b/doc/src/sgml/ref/create_subscription.sgml=0Aindex=20= e812beee37..6ec6524901=20100644=0A---=20= a/doc/src/sgml/ref/create_subscription.sgml=0A+++=20= b/doc/src/sgml/ref/create_subscription.sgml=0A@@=20-117,6=20+117,18=20@@=20= CREATE=20SUBSCRIPTION=20subscription_name=0A=20=20=20=20=20=20=20=20=0A=20=0A+=20=20= =20=20=20=20=20=0A+=20=20=20=20=20=20=20=20= disable_on_error=20= (boolean)=0A+=20=20=20=20=20=20=20=20=0A+=20= =20=20=20=20=20=20=20=20=0A+=20=20=20=20=20=20=20=20=20=20= Specifies=20whether=20the=20subscription=20should=20be=20automatically=20= disabled=0A+=20=20=20=20=20=20=20=20=20=20if=20replicating=20data=20from=20= the=20publisher=20triggers=20non-transient=20errors=0A+=20=20=20=20=20=20= =20=20=20=20such=20as=20referential=20integrity=20or=20permissions=20= errors.=20=20The=20default=20is=0A+=20=20=20=20=20=20=20=20=20=20= false.=0A+=20=20=20=20=20=20=20=20=20=0A+=20=20= =20=20=20=20=20=20=0A+=20=20=20=20=20=20=20=0A= +=0A=20=20=20=20=20=20=20=20=0A=20=20=20=20=20=20=20=20=20= enabled=20(boolean)=0A=20=20= =20=20=20=20=20=20=20=0Adiff=20--git=20= a/src/backend/catalog/pg_subscription.c=20= b/src/backend/catalog/pg_subscription.c=0Aindex=2029fc4218cd..5781da062c=20= 100644=0A---=20a/src/backend/catalog/pg_subscription.c=0A+++=20= b/src/backend/catalog/pg_subscription.c=0A@@=20-66,6=20+66,7=20@@=20= GetSubscription(Oid=20subid,=20bool=20missing_ok)=0A=20=09sub->name=20=3D=20= pstrdup(NameStr(subform->subname));=0A=20=09sub->owner=20=3D=20= subform->subowner;=0A=20=09sub->enabled=20=3D=20subform->subenabled;=0A+=09= sub->disableonerr=20=3D=20subform->subdisableonerr;=0A=20=09sub->binary=20= =3D=20subform->subbinary;=0A=20=09sub->stream=20=3D=20= subform->substream;=0A=20=0A@@=20-95,6=20+96,14=20@@=20= GetSubscription(Oid=20subid,=20bool=20missing_ok)=0A=20=09= Assert(!isnull);=0A=20=09sub->synccommit=20=3D=20= TextDatumGetCString(datum);=0A=20=0A+=09/*=20Get=20errmsg=20*/=0A+=09= datum=20=3D=20SysCacheGetAttr(SUBSCRIPTIONOID,=0A+=09=09=09=09=09=09=09= tup,=0A+=09=09=09=09=09=09=09Anum_pg_subscription_suberrmsg,=0A+=09=09=09= =09=09=09=09&isnull);=0A+=09Assert(!isnull);=0A+=09sub->errmsg=20=3D=20= TextDatumGetCString(datum);=0A+=0A=20=09/*=20Get=20publications=20*/=0A=20= =09datum=20=3D=20SysCacheGetAttr(SUBSCRIPTIONOID,=0A=20=09=09=09=09=09=09= =09tup,=0Adiff=20--git=20a/src/backend/catalog/system_views.sql=20= b/src/backend/catalog/system_views.sql=0Aindex=20999d984068..0e840c1a34=20= 100644=0A---=20a/src/backend/catalog/system_views.sql=0A+++=20= b/src/backend/catalog/system_views.sql=0A@@=20-1252,8=20+1252,10=20@@=20= CREATE=20VIEW=20pg_replication_origin_status=20AS=0A=20=0A=20REVOKE=20= ALL=20ON=20pg_replication_origin_status=20FROM=20public;=0A=20=0A---=20= All=20columns=20of=20pg_subscription=20except=20subconninfo=20are=20= publicly=20readable.=0A+--=20All=20columns=20of=20pg_subscription=20= except=20subconninfo=20and=20suberrmsg=20are=20publicly=0A+--=20= readable.=0A=20REVOKE=20ALL=20ON=20pg_subscription=20FROM=20public;=0A= -GRANT=20SELECT=20(oid,=20subdbid,=20subname,=20subowner,=20subenabled,=20= subbinary,=0A-=20=20=20=20=20=20=20=20=20=20=20=20=20=20substream,=20= subslotname,=20subsynccommit,=20subpublications)=0A+GRANT=20SELECT=20= (oid,=20subdbid,=20subname,=20subowner,=20subenabled,=20subdisableonerr,=0A= +=09=09=09=20=20subbinary,=20substream,=20subslotname,=20subsynccommit,=0A= +=09=09=09=20=20subpublications)=0A=20=20=20=20=20ON=20pg_subscription=20= TO=20public;=0Adiff=20--git=20a/src/backend/commands/subscriptioncmds.c=20= b/src/backend/commands/subscriptioncmds.c=0Aindex=20= 75e195f286..9f8adf529c=20100644=0A---=20= a/src/backend/commands/subscriptioncmds.c=0A+++=20= b/src/backend/commands/subscriptioncmds.c=0A@@=20-63,6=20+63,7=20@@=20= static=20void=0A=20parse_subscription_options(List=20*options,=0A=20=09=09= =09=09=09=09=20=20=20bool=20*connect,=0A=20=09=09=09=09=09=09=20=20=20= bool=20*enabled_given,=20bool=20*enabled,=0A+=09=09=09=09=09=09=20=20=20= bool=20*disableonerr_given,=20bool=20*disableonerr,=0A=20=09=09=09=09=09=09= =20=20=20bool=20*create_slot,=0A=20=09=09=09=09=09=09=20=20=20bool=20= *slot_name_given,=20char=20**slot_name,=0A=20=09=09=09=09=09=09=20=20=20= bool=20*copy_data,=0A@@=20-84,13=20+85,24=20@@=20= parse_subscription_options(List=20*options,=0A=20=09=09*connect=20=3D=20= true;=0A=20=09if=20(enabled)=0A=20=09{=0A+=09=09Assert(enabled_given);=0A= +=0A=20=09=09*enabled_given=20=3D=20false;=0A=20=09=09*enabled=20=3D=20= true;=0A=20=09}=0A+=09if=20(disableonerr)=0A+=09{=0A+=09=09= Assert(disableonerr_given);=0A+=0A+=09=09*disableonerr_given=20=3D=20= false;=0A+=09=09*disableonerr=20=3D=20false;=0A+=09}=0A=20=09if=20= (create_slot)=0A=20=09=09*create_slot=20=3D=20true;=0A=20=09if=20= (slot_name)=0A=20=09{=0A+=09=09Assert(slot_name_given);=0A+=0A=20=09=09= *slot_name_given=20=3D=20false;=0A=20=09=09*slot_name=20=3D=20NULL;=0A=20= =09}=0A@@=20-102,11=20+114,15=20@@=20parse_subscription_options(List=20= *options,=0A=20=09=09*refresh=20=3D=20true;=0A=20=09if=20(binary)=0A=20=09= {=0A+=09=09Assert(binary_given);=0A+=0A=20=09=09*binary_given=20=3D=20= false;=0A=20=09=09*binary=20=3D=20false;=0A=20=09}=0A=20=09if=20= (streaming)=0A=20=09{=0A+=09=09Assert(streaming_given);=0A+=0A=20=09=09= *streaming_given=20=3D=20false;=0A=20=09=09*streaming=20=3D=20false;=0A=20= =09}=0A@@=20-136,6=20+152,16=20@@=20parse_subscription_options(List=20= *options,=0A=20=09=09=09*enabled_given=20=3D=20true;=0A=20=09=09=09= *enabled=20=3D=20defGetBoolean(defel);=0A=20=09=09}=0A+=09=09else=20if=20= (strcmp(defel->defname,=20"disable_on_error")=20=3D=3D=200=20&&=0A+=09=09= =09=09=20disableonerr)=0A+=09=09{=0A+=09=09=09if=20(*disableonerr_given)=0A= +=09=09=09=09ereport(ERROR,=0A+=09=09=09=09=09=09= (errcode(ERRCODE_SYNTAX_ERROR),=0A+=09=09=09=09=09=09=20= errmsg("conflicting=20or=20redundant=20options")));=0A+=09=09=09= *disableonerr_given=20=3D=20true;=0A+=09=09=09*disableonerr=20=3D=20= defGetBoolean(defel);=0A+=09=09}=0A=20=09=09else=20if=20= (strcmp(defel->defname,=20"create_slot")=20=3D=3D=200=20&&=20= create_slot)=0A=20=09=09{=0A=20=09=09=09if=20(create_slot_given)=0A@@=20= -334,6=20+360,8=20@@=20CreateSubscription(CreateSubscriptionStmt=20= *stmt,=20bool=20isTopLevel)=0A=20=09bool=09=09connect;=0A=20=09bool=09=09= enabled_given;=0A=20=09bool=09=09enabled;=0A+=09bool=09=09= disableonerr_given;=0A+=09bool=09=09disableonerr;=0A=20=09bool=09=09= copy_data;=0A=20=09bool=09=09streaming;=0A=20=09bool=09=09= streaming_given;=0A@@=20-355,6=20+383,8=20@@=20= CreateSubscription(CreateSubscriptionStmt=20*stmt,=20bool=20isTopLevel)=0A= =20=09parse_subscription_options(stmt->options,=0A=20=09=09=09=09=09=09=09= =20=20=20&connect,=0A=20=09=09=09=09=09=09=09=20=20=20&enabled_given,=20= &enabled,=0A+=09=09=09=09=09=09=09=20=20=20&disableonerr_given,=0A+=09=09= =09=09=09=09=09=20=20=20&disableonerr,=0A=20=09=09=09=09=09=09=09=20=20=20= &create_slot,=0A=20=09=09=09=09=09=09=09=20=20=20&slotname_given,=20= &slotname,=0A=20=09=09=09=09=09=09=09=20=20=20©_data,=0A@@=20-427,6=20= +457,8=20@@=20CreateSubscription(CreateSubscriptionStmt=20*stmt,=20bool=20= isTopLevel)=0A=20=09=09DirectFunctionCall1(namein,=20= CStringGetDatum(stmt->subname));=0A=20=09= values[Anum_pg_subscription_subowner=20-=201]=20=3D=20= ObjectIdGetDatum(owner);=0A=20=09values[Anum_pg_subscription_subenabled=20= -=201]=20=3D=20BoolGetDatum(enabled);=0A+=09= values[Anum_pg_subscription_subdisableonerr=20-=201]=20=3D=0A+=09=09= BoolGetDatum(disableonerr);=0A=20=09= values[Anum_pg_subscription_subbinary=20-=201]=20=3D=20= BoolGetDatum(binary);=0A=20=09values[Anum_pg_subscription_substream=20-=20= 1]=20=3D=20BoolGetDatum(streaming);=0A=20=09= values[Anum_pg_subscription_subconninfo=20-=201]=20=3D=0A@@=20-438,6=20= +470,8=20@@=20CreateSubscription(CreateSubscriptionStmt=20*stmt,=20bool=20= isTopLevel)=0A=20=09=09nulls[Anum_pg_subscription_subslotname=20-=201]=20= =3D=20true;=0A=20=09values[Anum_pg_subscription_subsynccommit=20-=201]=20= =3D=0A=20=09=09CStringGetTextDatum(synchronous_commit);=0A+=09= values[Anum_pg_subscription_suberrmsg=20-=201]=20=3D=0A+=09=09= CStringGetTextDatum("");=0A=20=09= values[Anum_pg_subscription_subpublications=20-=201]=20=3D=0A=20=09=09= publicationListToArray(publications);=0A=20=0A@@=20-799,6=20+833,8=20@@=20= AlterSubscription(AlterSubscriptionStmt=20*stmt,=20bool=20isTopLevel)=0A=20= =09{=0A=20=09=09case=20ALTER_SUBSCRIPTION_OPTIONS:=0A=20=09=09=09{=0A+=09= =09=09=09bool=09=09disableonerr;=0A+=09=09=09=09bool=09=09= disableonerr_given;=0A=20=09=09=09=09char=09=20=20=20*slotname;=0A=20=09=09= =09=09bool=09=09slotname_given;=0A=20=09=09=09=09char=09=20=20=20= *synchronous_commit;=0A@@=20-810,6=20+846,8=20@@=20= AlterSubscription(AlterSubscriptionStmt=20*stmt,=20bool=20isTopLevel)=0A=20= =09=09=09=09parse_subscription_options(stmt->options,=0A=20=09=09=09=09=09= =09=09=09=09=09=20=20=20NULL,=09/*=20no=20"connect"=20*/=0A=20=09=09=09=09= =09=09=09=09=09=09=20=20=20NULL,=20NULL,=09/*=20no=20"enabled"=20*/=0A+=09= =09=09=09=09=09=09=09=09=09=20=20=20&disableonerr_given,=0A+=09=09=09=09=09= =09=09=09=09=09=20=20=20&disableonerr,=0A=20=09=09=09=09=09=09=09=09=09=09= =20=20=20NULL,=09/*=20no=20"create_slot"=20*/=0A=20=09=09=09=09=09=09=09=09= =09=09=20=20=20&slotname_given,=20&slotname,=0A=20=09=09=09=09=09=09=09=09= =09=09=20=20=20NULL,=09/*=20no=20"copy_data"=20*/=0A@@=20-818,6=20= +856,14=20@@=20AlterSubscription(AlterSubscriptionStmt=20*stmt,=20bool=20= isTopLevel)=0A=20=09=09=09=09=09=09=09=09=09=09=20=20=20&binary_given,=20= &binary,=0A=20=09=09=09=09=09=09=09=09=09=09=20=20=20&streaming_given,=20= &streaming);=0A=20=0A+=09=09=09=09if=20(disableonerr_given)=0A+=09=09=09=09= {=0A+=09=09=09=09=09values[Anum_pg_subscription_subdisableonerr=20-=201]=20= =3D=0A+=09=09=09=09=09=09BoolGetDatum(disableonerr);=0A+=09=09=09=09=09= replaces[Anum_pg_subscription_subdisableonerr=20-=201=20]=20=3D=0A+=09=09= =09=09=09=09true;=0A+=09=09=09=09}=0A+=0A=20=09=09=09=09if=20= (slotname_given)=0A=20=09=09=09=09{=0A=20=09=09=09=09=09if=20= (sub->enabled=20&&=20!slotname)=0A@@=20-867,6=20+913,7=20@@=20= AlterSubscription(AlterSubscriptionStmt=20*stmt,=20bool=20isTopLevel)=0A=20= =09=09=09=09parse_subscription_options(stmt->options,=0A=20=09=09=09=09=09= =09=09=09=09=09=20=20=20NULL,=09/*=20no=20"connect"=20*/=0A=20=09=09=09=09= =09=09=09=09=09=09=20=20=20&enabled_given,=20&enabled,=0A+=09=09=09=09=09= =09=09=09=09=09=20=20=20NULL,=20NULL,=09/*=20no=20"disableonerr"=20*/=0A=20= =09=09=09=09=09=09=09=09=09=09=20=20=20NULL,=09/*=20no=20"create_slot"=20= */=0A=20=09=09=09=09=09=09=09=09=09=09=20=20=20NULL,=20NULL,=09/*=20no=20= "slot_name"=20*/=0A=20=09=09=09=09=09=09=09=09=09=09=20=20=20NULL,=09/*=20= no=20"copy_data"=20*/=0A@@=20-885,6=20+932,13=20@@=20= AlterSubscription(AlterSubscriptionStmt=20*stmt,=20bool=20isTopLevel)=0A=20= =09=09=09=09=09BoolGetDatum(enabled);=0A=20=09=09=09=09= replaces[Anum_pg_subscription_subenabled=20-=201]=20=3D=20true;=0A=20=0A= +=09=09=09=09/*=0A+=09=09=09=09=20*=20Manually=20enabling=20or=20= disabling=20a=20subscription=20clears=20any=0A+=09=09=09=09=20*=20error=20= which=20automatically=20disabled=20it.=0A+=09=09=09=09=20*/=0A+=09=09=09=09= values[Anum_pg_subscription_suberrmsg=20-=201]=20=3D=20= CStringGetTextDatum("");=0A+=09=09=09=09= replaces[Anum_pg_subscription_suberrmsg=20-=201]=20=3D=20true;=0A+=0A=20=09= =09=09=09if=20(enabled)=0A=20=09=09=09=09=09= ApplyLauncherWakeupAtCommit();=0A=20=0A@@=20-912,6=20+966,7=20@@=20= AlterSubscription(AlterSubscriptionStmt=20*stmt,=20bool=20isTopLevel)=0A=20= =09=09=09=09parse_subscription_options(stmt->options,=0A=20=09=09=09=09=09= =09=09=09=09=09=20=20=20NULL,=09/*=20no=20"connect"=20*/=0A=20=09=09=09=09= =09=09=09=09=09=09=20=20=20NULL,=20NULL,=09/*=20no=20"enabled"=20*/=0A+=09= =09=09=09=09=09=09=09=09=09=20=20=20NULL,=20NULL,=09/*=20no=20= "disableonerr"=20*/=0A=20=09=09=09=09=09=09=09=09=09=09=20=20=20NULL,=09= /*=20no=20"create_slot"=20*/=0A=20=09=09=09=09=09=09=09=09=09=09=20=20=20= NULL,=20NULL,=09/*=20no=20"slot_name"=20*/=0A=20=09=09=09=09=09=09=09=09=09= =09=20=20=20©_data,=0A@@=20-958,6=20+1013,7=20@@=20= AlterSubscription(AlterSubscriptionStmt=20*stmt,=20bool=20isTopLevel)=0A=20= =09=09=09=09parse_subscription_options(stmt->options,=0A=20=09=09=09=09=09= =09=09=09=09=09=20=20=20NULL,=09/*=20no=20"connect"=20*/=0A=20=09=09=09=09= =09=09=09=09=09=09=20=20=20NULL,=20NULL,=09/*=20no=20"enabled"=20*/=0A+=09= =09=09=09=09=09=09=09=09=09=20=20=20NULL,=20NULL,=09/*=20no=20= "disableonerr"=20*/=0A=20=09=09=09=09=09=09=09=09=09=09=20=20=20NULL,=09= /*=20no=20"create_slot"=20*/=0A=20=09=09=09=09=09=09=09=09=09=09=20=20=20= NULL,=20NULL,=09/*=20no=20"slot_name"=20*/=0A=20=09=09=09=09=09=09=09=09=09= =09=20=20=20isadd=20?=20©_data=20:=20NULL,=09/*=20for=20drop,=20no=0A= @@=20-1005,6=20+1061,7=20@@=20AlterSubscription(AlterSubscriptionStmt=20= *stmt,=20bool=20isTopLevel)=0A=20=09=09=09=09= parse_subscription_options(stmt->options,=0A=20=09=09=09=09=09=09=09=09=09= =09=20=20=20NULL,=09/*=20no=20"connect"=20*/=0A=20=09=09=09=09=09=09=09=09= =09=09=20=20=20NULL,=20NULL,=09/*=20no=20"enabled"=20*/=0A+=09=09=09=09=09= =09=09=09=09=09=20=20=20NULL,=20NULL,=09/*=20no=20"disableonerr"=20*/=0A=20= =09=09=09=09=09=09=09=09=09=09=20=20=20NULL,=09/*=20no=20"create_slot"=20= */=0A=20=09=09=09=09=09=09=09=09=09=09=20=20=20NULL,=20NULL,=09/*=20no=20= "slot_name"=20*/=0A=20=09=09=09=09=09=09=09=09=09=09=20=20=20©_data,=0A= diff=20--git=20a/src/backend/replication/logical/launcher.c=20= b/src/backend/replication/logical/launcher.c=0Aindex=20= e3b11daa89..e7279ea86b=20100644=0A---=20= a/src/backend/replication/logical/launcher.c=0A+++=20= b/src/backend/replication/logical/launcher.c=0A@@=20-132,6=20+132,7=20@@=20= get_subscription_list(void)=0A=20=09=09sub->dbid=20=3D=20= subform->subdbid;=0A=20=09=09sub->owner=20=3D=20subform->subowner;=0A=20=09= =09sub->enabled=20=3D=20subform->subenabled;=0A+=09=09sub->disableonerr=20= =3D=20subform->subdisableonerr;=0A=20=09=09sub->name=20=3D=20= pstrdup(NameStr(subform->subname));=0A=20=09=09/*=20We=20don't=20fill=20= fields=20we=20are=20not=20interested=20in.=20*/=0A=20=0Adiff=20--git=20= a/src/backend/replication/logical/worker.c=20= b/src/backend/replication/logical/worker.c=0Aindex=20= bbb659dad0..54fd61afe5=20100644=0A---=20= a/src/backend/replication/logical/worker.c=0A+++=20= b/src/backend/replication/logical/worker.c=0A@@=20-62,6=20+62,7=20@@=0A=20= #include=20"access/xact.h"=0A=20#include=20"access/xlog_internal.h"=0A=20= #include=20"catalog/catalog.h"=0A+#include=20"catalog/indexing.h"=0A=20= #include=20"catalog/namespace.h"=0A=20#include=20"catalog/partition.h"=0A= =20#include=20"catalog/pg_inherits.h"=0A@@=20-2412,6=20+2413,160=20@@=20= LogicalRepApplyLoop(XLogRecPtr=20last_received)=0A=20=09= walrcv_endstreaming(LogRepWorkerWalRcvConn,=20&tli);=0A=20}=0A=20=0A+/*=0A= +=20*=20Errors=20which=20are=20transient,=20network=20protocol=20= related,=20or=20resource=20exhaustion=0A+=20*=20related,=20should=20not=20= disable=20a=20subscription.=20=20These=20may=20clear=20up=20without=20= user=0A+=20*=20intervention=20in=20the=20subscription,=20schema,=20or=20= data=20being=20replicated.=0A+=20*/=0A+static=20bool=0A= +IsSubscriptionDisablingError(void)=0A+{=0A+=09switch=20(geterrcode())=0A= +=09{=0A+=09=09case=20ERRCODE_CONNECTION_EXCEPTION:=0A+=09=09case=20= ERRCODE_CONNECTION_DOES_NOT_EXIST:=0A+=09=09case=20= ERRCODE_CONNECTION_FAILURE:=0A+=09=09case=20= ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION:=0A+=09=09case=20= ERRCODE_SQLSERVER_REJECTED_ESTABLISHMENT_OF_SQLCONNECTION:=0A+=09=09case=20= ERRCODE_TRANSACTION_RESOLUTION_UNKNOWN:=0A+=09=09case=20= ERRCODE_PROTOCOL_VIOLATION:=0A+=09=09case=20= ERRCODE_INSUFFICIENT_RESOURCES:=0A+=09=09case=20ERRCODE_DISK_FULL:=0A+=09= =09case=20ERRCODE_OUT_OF_MEMORY:=0A+=09=09case=20= ERRCODE_TOO_MANY_CONNECTIONS:=0A+=09=09case=20= ERRCODE_CONFIGURATION_LIMIT_EXCEEDED:=0A+=09=09case=20= ERRCODE_PROGRAM_LIMIT_EXCEEDED:=0A+=09=09case=20= ERRCODE_STATEMENT_TOO_COMPLEX:=0A+=09=09case=20ERRCODE_TOO_MANY_COLUMNS:=0A= +=09=09case=20ERRCODE_TOO_MANY_ARGUMENTS:=0A+=09=09case=20= ERRCODE_OPERATOR_INTERVENTION:=0A+=09=09case=20ERRCODE_QUERY_CANCELED:=0A= +=09=09case=20ERRCODE_ADMIN_SHUTDOWN:=0A+=09=09case=20= ERRCODE_CRASH_SHUTDOWN:=0A+=09=09case=20ERRCODE_CANNOT_CONNECT_NOW:=0A+=09= =09case=20ERRCODE_DATABASE_DROPPED:=0A+=09=09case=20= ERRCODE_IDLE_SESSION_TIMEOUT:=0A+=09=09=09return=20false;=0A+=09=09= default:=0A+=09=09=09break;=0A+=09}=0A+=0A+=09return=20true;=0A+}=0A+=0A= +/*=0A+=20*=20Recover=20from=20a=20possibly=20aborted=20transaction=20= state=20and=20disable=20the=20current=0A+=20*=20subscription.=0A+=20*/=0A= +static=20ErrorData=20*=0A+DisableSubscriptionOnError(MemoryContext=20= mcxt)=0A+{=0A+=09Relation=09rel;=0A+=09bool=09=09= nulls[Natts_pg_subscription];=0A+=09bool=09=09= replaces[Natts_pg_subscription];=0A+=09Datum=09=09= values[Natts_pg_subscription];=0A+=09HeapTuple=09tup;=0A+=09Oid=09=09=09= subid;=0A+=09Form_pg_subscription=20subform;=0A+=09ErrorData=20=20= *edata;=0A+=0A+=09/*=0A+=09=20*=20Clean=20up=20from=20the=20error=20and=20= get=20a=20fresh=20transaction=20in=20which=20to=0A+=09=20*=20disable=20= the=20subscription.=0A+=09=20*/=0A+=09MemoryContextSwitchTo(mcxt);=0A+=09= edata=20=3D=20CopyErrorData();=0A+=0A+=09ereport(LOG,=0A+=09=09=09= (errmsg("logical=20replication=20subscription=20\"%s\"=20will=20be=20= disabled=20due=20to=20error:=20%s",=0A+=09=09=09=09=09= MySubscription->name,=20edata->message)));=0A+=0A+=09= AbortOutOfAnyTransaction();=0A+=09FlushErrorState();=0A+=0A+=09= StartTransactionCommand();=0A+=0A+=09/*=20Look=20up=20our=20subscription=20= in=20the=20catalogs=20*/=0A+=09rel=20=3D=20= table_open(SubscriptionRelationId,=20RowExclusiveLock);=0A+=09tup=20=3D=20= SearchSysCacheCopy2(SUBSCRIPTIONNAME,=20MyDatabaseId,=0A+=09=09=09=09=09=09= =09=20=20CStringGetDatum(MySubscription->name));=0A+=09if=20= (!HeapTupleIsValid(tup))=0A+=09=09ereport(ERROR,=0A+=09=09=09=09= (errcode(ERRCODE_UNDEFINED_OBJECT),=0A+=09=09=09=09=20= errmsg("subscription=20\"%s\"=20does=20not=20exist",=0A+=09=09=09=09=09=09= MySubscription->name)));=0A+=0A+=09subform=20=3D=20= (Form_pg_subscription)=20GETSTRUCT(tup);=0A+=09subid=20=3D=20= subform->oid;=0A+=09LockSharedObject(SubscriptionRelationId,=20subid,=20= 0,=20AccessExclusiveLock);=0A+=0A+=09/*=0A+=09=20*=20We=20would=20not=20= be=20here=20unless=20this=20subscription's=20disableonerr=0A+=09=20*=20= field=20was=20true=20when=20our=20worker=20began=20applying=20changes,=20= but=20check=0A+=09=20*=20whether=20that=20field=20has=20changed=20in=20= the=20interim.=0A+=09=20*/=0A+=09if=20(!subform->subdisableonerr)=0A+=09=09= ReThrowError(edata);=0A+=0A+=09/*=20Form=20a=20new=20tuple.=20*/=0A+=09= memset(values,=200,=20sizeof(values));=0A+=09memset(nulls,=20false,=20= sizeof(nulls));=0A+=09memset(replaces,=20false,=20sizeof(replaces));=0A+=0A= +=09/*=20Set=20the=20subscription=20to=20disabled,=20and=20note=20the=20= reason.=20*/=0A+=09values[Anum_pg_subscription_subenabled=20-=201]=20=3D=20= BoolGetDatum(false);=0A+=09replaces[Anum_pg_subscription_subenabled=20-=20= 1]=20=3D=20true;=0A+=09values[Anum_pg_subscription_suberrmsg=20-=201]=20= =3D=0A+=09=09CStringGetTextDatum(edata->message=20?=20edata->message=20:=20= "");=0A+=09replaces[Anum_pg_subscription_suberrmsg=20-=201]=20=3D=20= true;=0A+=0A+=09/*=20Update=20the=20catalog=20*/=0A+=09tup=20=3D=20= heap_modify_tuple(tup,=20RelationGetDescr(rel),=20values,=20nulls,=0A+=09= =09=09=09=09=09=09replaces);=0A+=09CatalogTupleUpdate(rel,=20= &tup->t_self,=20tup);=0A+=09heap_freetuple(tup);=0A+=0A+=09= table_close(rel,=20RowExclusiveLock);=0A+=0A+=09= CommitTransactionCommand();=0A+=0A+=09return=20edata;=0A+}=0A+=0A+/*=0A+=20= *=20Apply=20main=20loop=20with=20logic=20to=20disable=20a=20stuck=20= subscription=0A+=20*=0A+=20*=20Applies=20changes,=20and=20for=20= non-transient=20errors,=20catches=20the=20error=20and=0A+=20*=20disables=20= the=20subscription=20before=20rethrowing.=0A+=20*/=0A+static=20void=0A= +LogicalRepApplyLoopTry(XLogRecPtr=20last_received)=0A+{=0A+=09= MemoryContext=20mcxt=20=3D=20CurrentMemoryContext;=0A+=09bool=09=09= did_error=20=3D=20false;=0A+=0A+=09PG_TRY();=0A+=09{=0A+=09=09= LogicalRepApplyLoop(last_received);=0A+=09}=0A+=09PG_CATCH();=0A+=09{=0A= +=09=09if=20(IsSubscriptionDisablingError())=0A+=09=09=09did_error=20=3D=20= true;=0A+=09=09else=0A+=09=09=09PG_RE_THROW();=0A+=09}=0A+=09= PG_END_TRY();=0A+=0A+=09/*=0A+=09=20*=20If=20we=20caught=20an=20error=20= above,=20disable=20the=20subscription=20and=20record=20copies=0A+=09=20*=20= of=20the=20relevant=20error=20text,=20then=20rethrow=20the=20error=20we=20= caught=20so=20it=20gets=0A+=09=20*=20logged=20and=20our=20process=20= exits=20appropriately.=0A+=09=20*/=0A+=09if=20(did_error)=0A+=09=09= ReThrowError(DisableSubscriptionOnError(mcxt));=0A+}=0A+=0A=20/*=0A=20=20= *=20Send=20a=20Standby=20Status=20Update=20message=20to=20server.=0A=20=20= *=0A@@=20-3167,7=20+3322,34=20@@=20ApplyWorkerMain(Datum=20main_arg)=0A=20= =09=09char=09=20=20=20*syncslotname;=0A=20=0A=20=09=09/*=20This=20is=20= table=20synchronization=20worker,=20call=20initial=20sync.=20*/=0A-=09=09= syncslotname=20=3D=20LogicalRepSyncTableStart(&origin_startpos);=0A+=09=09= if=20(MySubscription->disableonerr)=0A+=09=09{=0A+=09=09=09MemoryContext=20= mcxt=20=3D=20CurrentMemoryContext;=0A+=09=09=09bool=09did_error=20=3D=20= false;=0A+=0A+=09=09=09PG_TRY();=0A+=09=09=09{=0A+=09=09=09=09= syncslotname=20=3D=20LogicalRepSyncTableStart(&origin_startpos);=0A+=09=09= =09}=0A+=09=09=09PG_CATCH();=0A+=09=09=09{=0A+=09=09=09=09if=20= (IsSubscriptionDisablingError())=0A+=09=09=09=09=09did_error=20=3D=20= true;=0A+=09=09=09=09else=0A+=09=09=09=09=09PG_RE_THROW();=0A+=09=09=09}=0A= +=09=09=09PG_END_TRY();=0A+=0A+=09=09=09/*=0A+=09=09=09=20*=20If=20we=20= caught=20an=20error=20above,=20disable=20the=20subscription=20and=20= record=20copies=0A+=09=09=09=20*=20of=20the=20relevant=20error=20text,=20= then=20rethrow=20the=20error=20we=20caught=20so=20it=20gets=0A+=09=09=09=20= *=20logged=20and=20our=20process=20exits=20appropriately.=0A+=09=09=09=20= */=0A+=09=09=09if=20(did_error)=0A+=09=09=09=09= ReThrowError(DisableSubscriptionOnError(mcxt));=0A+=09=09}=0A+=09=09else=0A= +=09=09=09syncslotname=20=3D=20= LogicalRepSyncTableStart(&origin_startpos);=0A=20=0A=20=09=09/*=20= allocate=20slot=20name=20in=20long-lived=20context=20*/=0A=20=09=09= myslotname=20=3D=20MemoryContextStrdup(ApplyContext,=20syncslotname);=0A= @@=20-3241,7=20+3423,10=20@@=20ApplyWorkerMain(Datum=20main_arg)=0A=20=09= walrcv_startstreaming(LogRepWorkerWalRcvConn,=20&options);=0A=20=0A=20=09= /*=20Run=20the=20main=20loop.=20*/=0A-=09= LogicalRepApplyLoop(origin_startpos);=0A+=09if=20= (MySubscription->disableonerr)=0A+=09=09= LogicalRepApplyLoopTry(origin_startpos);=0A+=09else=0A+=09=09= LogicalRepApplyLoop(origin_startpos);=0A=20=0A=20=09proc_exit(0);=0A=20}=0A= diff=20--git=20a/src/bin/pg_dump/pg_dump.c=20b/src/bin/pg_dump/pg_dump.c=0A= index=208f53cc7c3b..16d9d006d1=20100644=0A---=20= a/src/bin/pg_dump/pg_dump.c=0A+++=20b/src/bin/pg_dump/pg_dump.c=0A@@=20= -4307,6=20+4307,7=20@@=20getSubscriptions(Archive=20*fout)=0A=20=09int=09= =09=09i_subconninfo;=0A=20=09int=09=09=09i_subslotname;=0A=20=09int=09=09= =09i_subsynccommit;=0A+=09int=09=09=09i_suberrmsg;=0A=20=09int=09=09=09= i_subpublications;=0A=20=09int=09=09=09i_subbinary;=0A=20=09int=09=09=09= i,=0A@@=20-4338,7=20+4339,7=20@@=20getSubscriptions(Archive=20*fout)=0A=20= =09=09=09=09=09=20=20"SELECT=20s.tableoid,=20s.oid,=20s.subname,\n"=0A=20= =09=09=09=09=09=20=20"=20(%s=20s.subowner)=20AS=20rolname,\n"=0A=20=09=09= =09=09=09=20=20"=20s.subconninfo,=20s.subslotname,=20s.subsynccommit,\n"=0A= -=09=09=09=09=09=20=20"=20s.subpublications,\n",=0A+=09=09=09=09=09=20=20= "=20s.suberrmsg,=20s.subpublications,\n",=0A=20=09=09=09=09=09=20=20= username_subquery);=0A=20=0A=20=09if=20(fout->remoteVersion=20>=3D=20= 140000)=0A@@=20-4367,6=20+4368,7=20@@=20getSubscriptions(Archive=20= *fout)=0A=20=09i_subconninfo=20=3D=20PQfnumber(res,=20"subconninfo");=0A=20= =09i_subslotname=20=3D=20PQfnumber(res,=20"subslotname");=0A=20=09= i_subsynccommit=20=3D=20PQfnumber(res,=20"subsynccommit");=0A+=09= i_suberrmsg=20=3D=20PQfnumber(res,=20"suberrmsg");=0A=20=09= i_subpublications=20=3D=20PQfnumber(res,=20"subpublications");=0A=20=09= i_subbinary=20=3D=20PQfnumber(res,=20"subbinary");=0A=20=09i_substream=20= =3D=20PQfnumber(res,=20"substream");=0A@@=20-4389,6=20+4391,8=20@@=20= getSubscriptions(Archive=20*fout)=0A=20=09=09=09subinfo[i].subslotname=20= =3D=20pg_strdup(PQgetvalue(res,=20i,=20i_subslotname));=0A=20=09=09= subinfo[i].subsynccommit=20=3D=0A=20=09=09=09pg_strdup(PQgetvalue(res,=20= i,=20i_subsynccommit));=0A+=09=09subinfo[i].suberrmsg=20=3D=0A+=09=09=09= pg_strdup(PQgetvalue(res,=20i,=20i_suberrmsg));=0A=20=09=09= subinfo[i].subpublications=20=3D=0A=20=09=09=09pg_strdup(PQgetvalue(res,=20= i,=20i_subpublications));=0A=20=09=09subinfo[i].subbinary=20=3D=0Adiff=20= --git=20a/src/bin/pg_dump/pg_dump.h=20b/src/bin/pg_dump/pg_dump.h=0A= index=2049e1b0a09c..3c2a8dd38c=20100644=0A---=20= a/src/bin/pg_dump/pg_dump.h=0A+++=20b/src/bin/pg_dump/pg_dump.h=0A@@=20= -638,6=20+638,7=20@@=20typedef=20struct=20_SubscriptionInfo=0A=20=09char=09= =20=20=20*subbinary;=0A=20=09char=09=20=20=20*substream;=0A=20=09char=09=20= =20=20*subsynccommit;=0A+=09char=09=20=20=20*suberrmsg;=0A=20=09char=09=20= =20=20*subpublications;=0A=20}=20SubscriptionInfo;=0A=20=0Adiff=20--git=20= a/src/include/catalog/pg_subscription.h=20= b/src/include/catalog/pg_subscription.h=0Aindex=200060ebfb40..b094bd082f=20= 100644=0A---=20a/src/include/catalog/pg_subscription.h=0A+++=20= b/src/include/catalog/pg_subscription.h=0A@@=20-52,6=20+52,9=20@@=20= CATALOG(pg_subscription,6100,SubscriptionRelationId)=20= BKI_SHARED_RELATION=20BKI_ROW=0A=20=09bool=09=09subenabled;=09=09/*=20= True=20if=20the=20subscription=20is=20enabled=20(the=0A=20=09=09=09=09=09= =09=09=09=20*=20worker=20should=20be=20running)=20*/=0A=20=0A+=09bool=09=09= subdisableonerr;=09/*=20True=20if=20apply=20errors=20should=0A+=09=09=09=09= =09=09=09=09=09=20*=20disable=20the=20subscription=20upon=20error=20*/=0A= +=0A=20=09bool=09=09subbinary;=09=09/*=20True=20if=20the=20subscription=20= wants=20the=0A=20=09=09=09=09=09=09=09=09=20*=20publisher=20to=20send=20= data=20in=20binary=20*/=0A=20=0A@@=20-67,6=20+70,9=20@@=20= CATALOG(pg_subscription,6100,SubscriptionRelationId)=20= BKI_SHARED_RELATION=20BKI_ROW=0A=20=09/*=20Synchronous=20commit=20= setting=20for=20worker=20*/=0A=20=09text=09=09subsynccommit=20= BKI_FORCE_NOT_NULL;=0A=20=0A+=09/*=20Message=20from=20error=20which=20= disabled=20this=20subscription=20*/=0A+=09text=09=09suberrmsg=20= BKI_FORCE_NOT_NULL;=0A+=0A=20=09/*=20List=20of=20publications=20= subscribed=20to=20*/=0A=20=09text=09=09subpublications[1]=20= BKI_FORCE_NOT_NULL;=0A=20#endif=0A@@=20-91,12=20+97,14=20@@=20typedef=20= struct=20Subscription=0A=20=09char=09=20=20=20*name;=09=09=09/*=20Name=20= of=20the=20subscription=20*/=0A=20=09Oid=09=09=09owner;=09=09=09/*=20Oid=20= of=20the=20subscription=20owner=20*/=0A=20=09bool=09=09enabled;=09=09/*=20= Indicates=20if=20the=20subscription=20is=20enabled=20*/=0A+=09bool=09=09= disableonerr;=09/*=20Whether=20errors=20automatically=20disable=20*/=0A=20= =09bool=09=09binary;=09=09=09/*=20Indicates=20if=20the=20subscription=20= wants=20data=20in=0A=20=09=09=09=09=09=09=09=09=20*=20binary=20format=20= */=0A=20=09bool=09=09stream;=09=09=09/*=20Allow=20streaming=20= in-progress=20transactions.=20*/=0A=20=09char=09=20=20=20*conninfo;=09=09= /*=20Connection=20string=20to=20the=20publisher=20*/=0A=20=09char=09=20=20= =20*slotname;=09=09/*=20Name=20of=20the=20replication=20slot=20*/=0A=20=09= char=09=20=20=20*synccommit;=09=09/*=20Synchronous=20commit=20setting=20= for=20worker=20*/=0A+=09char=09=20=20=20*errmsg;=09=09=09/*=20Message=20= from=20error=20which=20disabled=20*/=0A=20=09List=09=20=20=20= *publications;=09/*=20List=20of=20publication=20names=20to=20subscribe=20= to=20*/=0A=20}=20Subscription;=0A=20=0Adiff=20--git=20= a/src/test/perl/PostgresNode.pm=20b/src/test/perl/PostgresNode.pm=0A= index=202027cbf43d..33dc9d5367=20100644=0A---=20= a/src/test/perl/PostgresNode.pm=0A+++=20b/src/test/perl/PostgresNode.pm=0A= @@=20-2442,6=20+2442,46=20@@=20sub=20wait_for_slot_catchup=0A=20=09= return;=0A=20}=0A=20=0A+=3Dpot=0A+=0A+=3Ditem=20= $node->wait_for_subscription($dbname,=20@subcriptions)=0A+=0A+Wait=20for=20= the=20named=20subscriptions=20to=20catch=20up=20or=20to=20be=20disabled.=0A= +=0A+=3Dcut=0A+=0A+sub=20wait_for_subscriptions=0A+{=0A+=09my=20($self,=20= $dbname,=20@subscriptions)=20=3D=20@_;=0A+=0A+=09#=20Unique-ify=20the=20= subscriptions=20passed=20by=20the=20caller=0A+=09my=20%unique=20=3D=20= map=20{=20$_=20=3D>=201=20}=20@subscriptions;=0A+=09my=20@unique=20=3D=20= sort=20keys=20%unique;=0A+=09my=20$unique_count=20=3D=20scalar(@unique);=0A= +=0A+=09#=20Construct=20a=20SQL=20list=20from=20the=20unique=20= subscription=20names=0A+=09my=20@escaped=20=3D=20map=20{=20s/'/''/g;=20= s/\\/\\\\/g;=20$_=20}=20@unique;=0A+=09my=20$sublist=20=3D=20join(',=20= ',=20map=20{=20"'$_'"=20}=20@escaped);=0A+=0A+=09my=20$polling_sql=20=3D=20= qq(=0A+=09=09SELECT=20COUNT(1)=20=3D=20$unique_count=20FROM=0A+=09=09=09= (SELECT=20s.oid=0A+=09=09=09=09FROM=20pg_catalog.pg_subscription=20s=0A+=09= =09=09=09LEFT=20JOIN=20pg_catalog.pg_subscription_rel=20sr=0A+=09=09=09=09= ON=20sr.srsubid=20=3D=20s.oid=0A+=09=09=09=09WHERE=20(sr=20IS=20NULL=20= OR=20sr.srsubstate=20IN=20('s',=20'r'))=0A+=09=09=09=09=20=20AND=20= s.subname=20IN=20($sublist)=0A+=09=09=09=09=20=20AND=20s.subenabled=20IS=20= TRUE=0A+=09=09=09=20UNION=0A+=09=09=09=20SELECT=20s.oid=0A+=09=09=09=09= FROM=20pg_catalog.pg_subscription=20s=0A+=09=09=09=09WHERE=20s.subname=20= IN=20($sublist)=0A+=09=09=09=09=20=20AND=20s.subenabled=20IS=20FALSE=0A+=09= =09=09)=20AS=20synced_or_disabled=0A+=09=09);=0A+=09return=20= $self->poll_query_until($dbname,=20$polling_sql);=0A+}=0A+=0A=20=3Dpod=0A= =20=0A=20=3Ditem=20$node->query_hash($dbname,=20$query,=20@columns)=0A= diff=20--git=20a/src/test/subscription/t/022_disable_on_error.pl=20= b/src/test/subscription/t/022_disable_on_error.pl=0Anew=20file=20mode=20= 100644=0Aindex=200000000000..5c10981a6b=0A---=20/dev/null=0A+++=20= b/src/test/subscription/t/022_disable_on_error.pl=0A@@=20-0,0=20+1,205=20= @@=0A+=0A+#=20Copyright=20(c)=202021,=20PostgreSQL=20Global=20= Development=20Group=0A+=0A+#=20Test=20of=20logical=20replication=20= subscription=20self-disabling=20feature=0A+use=20strict;=0A+use=20= warnings;=0A+use=20PostgresNode;=0A+use=20TestLib;=0A+use=20Test::More=20= tests=20=3D>=2013;=0A+=0A+my=20@schemas=20=3D=20qw(s1=20s2);=0A+my=20= ($schema,=20$cmd);=0A+=0A+my=20$node_publisher=20=3D=20= get_new_node('publisher');=0A+$node_publisher->init(allows_streaming=20= =3D>=20'logical');=0A+$node_publisher->start;=0A+=0A+my=20= $node_subscriber=20=3D=20get_new_node('subscriber');=0A= +$node_subscriber->init;=0A+$node_subscriber->start;=0A+=0A+#=20Create=20= identical=20schema,=20table=20and=20index=20on=20both=20the=20publisher=20= and=0A+#=20subscriber=0A+#=0A+for=20$schema=20(@schemas)=0A+{=0A+=09$cmd=20= =3D=20qq(=0A+CREATE=20SCHEMA=20$schema;=0A+CREATE=20TABLE=20$schema.tbl=20= (i=20INT);=0A+ALTER=20TABLE=20$schema.tbl=20REPLICA=20IDENTITY=20FULL;=0A= +CREATE=20INDEX=20${schema}_tbl_idx=20ON=20$schema.tbl(i));=0A+=09= $node_publisher->safe_psql('postgres',=20$cmd);=0A+=09= $node_subscriber->safe_psql('postgres',=20$cmd);=0A+}=0A+=0A+#=20Create=20= non-unique=20data=20in=20both=20schemas=20on=20the=20publisher.=0A+#=0A= +for=20$schema=20(@schemas)=0A+{=0A+=09$cmd=20=3D=20qq(INSERT=20INTO=20= $schema.tbl=20(i)=20VALUES=20(1),=20(1),=20(1));=0A+=09= $node_publisher->safe_psql('postgres',=20$cmd);=0A+}=0A+=0A+#=20Create=20= an=20additional=20unique=20index=20in=20schema=20s1=20on=20the=20= subscriber=20only.=20=20When=0A+#=20we=20create=20subscriptions,=20= below,=20this=20should=20cause=20subscription=20"s1"=20on=20the=0A+#=20= subscriber=20to=20fail=20during=20initial=20synchronization=20and=20to=20= get=20automatically=0A+#=20disabled.=0A+#=0A+$cmd=20=3D=20qq(CREATE=20= UNIQUE=20INDEX=20s1_tbl_unique=20ON=20s1.tbl=20(i));=0A= +$node_subscriber->safe_psql('postgres',=20$cmd);=0A+=0A+#=20Create=20= publications=20and=20subscriptions=20linking=20the=20schemas=20on=0A+#=20= the=20publisher=20with=20those=20on=20the=20subscriber.=20=20This=20= tests=20that=20the=0A+#=20uniqueness=20violations=20cause=20subscription=20= "s1"=20to=20fail=20during=0A+#=20initial=20synchronization.=0A+#=0A+my=20= $publisher_connstr=20=3D=20$node_publisher->connstr=20.=20'=20= dbname=3Dpostgres';=0A+for=20$schema=20(@schemas)=0A+{=0A+=09#=20Create=20= the=20publication=20for=20this=20table=0A+=09$cmd=20=3D=20qq(=0A+CREATE=20= PUBLICATION=20$schema=20FOR=20TABLE=20$schema.tbl);=0A+=09= $node_publisher->safe_psql('postgres',=20$cmd);=0A+=0A+=09#=20Create=20= the=20subscription=20for=20this=20table=0A+=09$cmd=20=3D=20qq(=0A+CREATE=20= SUBSCRIPTION=20$schema=0A+=09CONNECTION=20'$publisher_connstr'=0A+=09= PUBLICATION=20$schema=0A+=09WITH=20(disable_on_error=20=3D=20true));=0A+=09= $node_subscriber->safe_psql('postgres',=20$cmd);=0A+}=0A+=0A+#=20Wait=20= for=20the=20initial=20subscription=20synchronizations=20to=20finish=20or=20= fail.=0A+#=0A+$node_subscriber->wait_for_subscriptions('postgres',=20= @schemas)=0A+=09or=20die=20"Timed=20out=20while=20waiting=20for=20= subscriber=20to=20synchronize=20data";=0A+=0A+#=20Subscription=20"s1"=20= should=20have=20disabled=20itself=20due=20to=20error.=0A+#=0A+$cmd=20=3D=20= qq(=0A+SELECT=20subenabled=20FROM=20pg_catalog.pg_subscription=20WHERE=20= subname=20=3D=20's1');=0A+is=20($node_subscriber->safe_psql('postgres',=20= $cmd),=0A+=09"f",=20"subscription=20s1=20no=20longer=20enabled");=0A= +$cmd=20=3D=20qq(=0A+SELECT=20suberrmsg=20FROM=20= pg_catalog.pg_subscription=20WHERE=20subname=20=3D=20's1');=0A+is=20= ($node_subscriber->safe_psql('postgres',=20$cmd),=0A+=09qq(duplicate=20= key=20value=20violates=20unique=20constraint=20"s1_tbl_unique"),=0A+=09= "subscription=20s1=20disabled=20during=20synchronization=20by=20unique=20= key=20violation");=0A+=0A+#=20Subscription=20"s2"=20should=20have=20= copied=20the=20initial=20data=20without=20incident.=0A+#=0A+$cmd=20=3D=20= qq(=0A+SELECT=20subenabled=20FROM=20pg_catalog.pg_subscription=20WHERE=20= subname=20=3D=20's2');=0A+is=20($node_subscriber->safe_psql('postgres',=20= $cmd),=0A+=09"t",=20"subscription=20s2=20still=20enabled");=0A+$cmd=20=3D=20= qq(SELECT=20i,=20COUNT(*)=20FROM=20s2.tbl=20GROUP=20BY=20i);=0A+is=20= ($node_subscriber->safe_psql('postgres',=20$cmd),=0A+=09"1|3",=0A+=09= "subscription=20s2=20replicated=20initial=20data");=0A+=0A+#=20Enter=20= unique=20data=20for=20both=20schemas=20on=20the=20publisher.=20=20This=20= should=20succeed=20on=0A+#=20the=20publisher=20node,=20and=20not=20cause=20= any=20additional=20problems=20on=20the=20subscriber=0A+#=20side=20= either,=20though=20disabled=20subscription=20"s1"=20should=20not=20= replicate=20anything.=0A+#=0A+for=20$schema=20(@schemas)=0A+{=0A+=09$cmd=20= =3D=20qq(INSERT=20INTO=20$schema.tbl=20(i)=20VALUES=20(2));=0A+=09= $node_publisher->safe_psql('postgres',=20$cmd);=0A+}=0A+=0A+#=20Wait=20= for=20the=20data=20to=20replicate=20for=20the=20subscriptions.=20=20This=20= tests=20that=20the=0A+#=20problems=20encountered=20by=20subscription=20= "s1"=20do=20not=20cause=20subscription=20"s2"=20to=0A+#=20get=20stuck.=0A= +$node_subscriber->wait_for_subscriptions('postgres',=20@schemas)=0A+=09= or=20die=20"Timed=20out=20while=20waiting=20for=20subscriber=20to=20= synchronize=20data";=0A+=0A+#=20Subscription=20"s1"=20should=20still=20= be=20disabled=20and=20have=20replicated=20no=20data=0A+#=0A+$cmd=20=3D=20= qq(=0A+SELECT=20subenabled=20FROM=20pg_catalog.pg_subscription=20WHERE=20= subname=20=3D=20's1');=0A+is=20($node_subscriber->safe_psql('postgres',=20= $cmd),=0A+=09"f",=20"subscription=20s1=20still=20disabled");=0A+$cmd=20=3D= =20qq(=0A+SELECT=20suberrmsg=20FROM=20pg_catalog.pg_subscription=20WHERE=20= subname=20=3D=20's1');=0A+is=20($node_subscriber->safe_psql('postgres',=20= $cmd),=0A+=09qq(duplicate=20key=20value=20violates=20unique=20constraint=20= "s1_tbl_unique"),=0A+=09"subscription=20s1=20still=20disabled=20by=20= unique=20key=20violation");=0A+=0A+#=20Subscription=20"s2"=20should=20= still=20be=20enabled=20and=20have=20replicated=20all=20changes=0A+#=0A= +$cmd=20=3D=20qq(=0A+SELECT=20subenabled=20FROM=20= pg_catalog.pg_subscription=20WHERE=20subname=20=3D=20's2');=0A+is=20= ($node_subscriber->safe_psql('postgres',=20$cmd),=0A+=09"t",=20= "subscription=20s2=20still=20enabled");=0A+$cmd=20=3D=20q(SELECT=20= MAX(i),=20COUNT(*)=20FROM=20s2.tbl);=0A+is=20= ($node_subscriber->safe_psql('postgres',=20$cmd),=0A+=09"2|4",=20= "subscription=20s2=20replicated=20data");=0A+=0A+#=20Drop=20the=20unique=20= index=20on=20"s1"=20which=20caused=20the=20subscription=20to=20be=20= disabled=0A+#=0A+$cmd=20=3D=20qq(DROP=20INDEX=20s1.s1_tbl_unique);=0A= +$node_subscriber->safe_psql('postgres',=20$cmd);=0A+=0A+#=20Re-enable=20= the=20subscription=20"s1"=0A+#=0A+$cmd=20=3D=20q(ALTER=20SUBSCRIPTION=20= s1=20ENABLE);=0A+$node_subscriber->safe_psql('postgres',=20$cmd);=0A+=0A= +#=20Wait=20for=20the=20data=20to=20replicate=0A+#=0A= +$node_subscriber->wait_for_subscriptions('postgres',=20@schemas)=0A+=09= or=20die=20"Timed=20out=20while=20waiting=20for=20subscriber=20to=20= synchronize=20data";=0A+=0A+#=20Check=20that=20we=20have=20the=20new=20= data=20in=20s1.tbl=0A+#=0A+$cmd=20=3D=20q(SELECT=20MAX(i),=20COUNT(*)=20= FROM=20s1.tbl);=0A+is=20($node_subscriber->safe_psql('postgres',=20= $cmd),=0A+=09"2|4",=20"subscription=20s1=20replicated=20data");=0A+=0A+#=20= Delete=20the=20data=20from=20the=20subscriber=20only,=20and=20recreate=20= the=20unique=20index=0A+#=0A+$cmd=20=3D=20q(=0A+DELETE=20FROM=20s1.tbl;=0A= +CREATE=20UNIQUE=20INDEX=20s1_tbl_unique=20ON=20s1.tbl=20(i));=0A= +$node_subscriber->safe_psql('postgres',=20$cmd);=0A+=0A+#=20Add=20more=20= non-unique=20data=20to=20the=20publisher=0A+for=20$schema=20(@schemas)=0A= +{=0A+=09$cmd=20=3D=20qq(INSERT=20INTO=20$schema.tbl=20(i)=20VALUES=20= (3),=20(3),=20(3));=0A+=09$node_publisher->safe_psql('postgres',=20= $cmd);=0A+}=0A+=0A+#=20Wait=20for=20the=20data=20to=20replicate=20for=20= the=20subscriptions.=20=20This=20tests=20that=0A+#=20uniqueness=20= violations=20encountered=20during=20replication=20cause=20s1=20to=20be=20= disabled.=0A+#=0A+$node_subscriber->wait_for_subscriptions('postgres',=20= @schemas)=0A+=09or=20die=20"Timed=20out=20while=20waiting=20for=20= subscriber=20to=20synchronize=20data";=0A+=0A+#=20Subscription=20"s1"=20= should=20have=20disabled=20itself=20due=20to=20error.=0A+#=0A+$cmd=20=3D=20= qq(=0A+SELECT=20subenabled=20FROM=20pg_catalog.pg_subscription=20WHERE=20= subname=20=3D=20's1');=0A+is=20($node_subscriber->safe_psql('postgres',=20= $cmd),=0A+=09"f",=20"subscription=20s1=20no=20longer=20enabled");=0A= +$cmd=20=3D=20qq(=0A+SELECT=20suberrmsg=20FROM=20= pg_catalog.pg_subscription=20WHERE=20subname=20=3D=20's1');=0A+is=20= ($node_subscriber->safe_psql('postgres',=20$cmd),=0A+=09qq(duplicate=20= key=20value=20violates=20unique=20constraint=20"s1_tbl_unique"),=0A+=09= "subscription=20s1=20disabled=20during=20replication=20by=20unique=20key=20= violation");=0A+=0A+#=20Subscription=20"s2"=20should=20have=20copied=20= the=20initial=20data=20without=20incident.=0A+#=0A+$cmd=20=3D=20qq(=0A= +SELECT=20subenabled=20FROM=20pg_catalog.pg_subscription=20WHERE=20= subname=20=3D=20's2');=0A+is=20($node_subscriber->safe_psql('postgres',=20= $cmd),=0A+=09"t",=20"subscription=20s2=20still=20enabled");=0A+$cmd=20=3D=20= qq(SELECT=20MAX(i),=20COUNT(*)=20FROM=20s2.tbl);=0A+is=20= ($node_subscriber->safe_psql('postgres',=20$cmd),=0A+=09"3|7",=0A+=09= "subscription=20s2=20replicated=20additional=20data");=0A+=0A= +$node_subscriber->stop;=0A+$node_publisher->stop;=0A--=20=0A2.21.1=20= (Apple=20Git-122.3)=0A=0A= --Apple-Mail=_E69A0652-B6B2-4D25-BE6D-3FCC2DC247AB Content-Transfer-Encoding: quoted-printable Content-Type: text/plain; charset=utf-8 =E2=80=94 Mark Dilger EnterpriseDB: http://www.enterprisedb.com The Enterprise PostgreSQL Company --Apple-Mail=_E69A0652-B6B2-4D25-BE6D-3FCC2DC247AB--