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 1ltyTP-0000gX-Kv for pgsql-hackers@arkaria.postgresql.org; Thu, 17 Jun 2021 20:18:48 +0000 Received: from localhost ([127.0.0.1] helo=malur.postgresql.org) by malur.postgresql.org with esmtp (Exim 4.92) (envelope-from ) id 1ltyTO-0007xv-Le for pgsql-hackers@arkaria.postgresql.org; Thu, 17 Jun 2021 20:18:46 +0000 Received: from makus.postgresql.org ([2001:4800:3e1:1::229]) by malur.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_256_GCM_SHA384:256) (Exim 4.92) (envelope-from ) id 1ltyTN-0007xO-Ud for pgsql-hackers@lists.postgresql.org; Thu, 17 Jun 2021 20:18:46 +0000 Received: from mail-pj1-x1036.google.com ([2607:f8b0:4864:20::1036]) by makus.postgresql.org with esmtps (TLS1.3:ECDHE_RSA_AES_128_GCM_SHA256:128) (Exim 4.92) (envelope-from ) id 1ltyTK-0006Yw-A2 for pgsql-hackers@lists.postgresql.org; Thu, 17 Jun 2021 20:18:44 +0000 Received: by mail-pj1-x1036.google.com with SMTP id m15-20020a17090a5a4fb029016f385ffad0so1720951pji.0 for ; Thu, 17 Jun 2021 13:18:42 -0700 (PDT) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=enterprisedb-com.20150623.gappssmtp.com; s=20150623; h=from:mime-version:subject:message-id:date:to; bh=AKxms8z++r5gbfEnLzKWRgdrCIhGlxwsQZ+slyhu0kI=; b=knnfqdEkNdPTN2dThZ8Dr2aWgxI0gzXEJ+5FsGvshUYWCZRnLRrRardelykzmnlJWG +Rf/zpxN6MQqEHtcXox/GewuRbgnNxibSgNRGmOS0OPUmVKUVe3FcR04Y4/ReSW004Wv OM6M0rzDdHrp7luAG/8q9CVsnOCLfesnT+YQo3UgAAarClHMXLR+P8NQ/kXpL4er6MAM behU+7byBlIsv2F93PZj7eXGNmvaY9cD+u77+5FhNsy8Mp8dMDGEiqiS3uXfu7uGUuFs GJJDB9AbT12u+6Xn7vJbZMjNSLtSAKf/yRqHj8yXJUWhDzw3zu13Pa92iDm4tLTh2xFg V7NQ== X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20161025; h=x-gm-message-state:from:mime-version:subject:message-id:date:to; bh=AKxms8z++r5gbfEnLzKWRgdrCIhGlxwsQZ+slyhu0kI=; b=ufVSdcZzYKJJj4+O5MZg2Jhn1RA+HmWetlmjqC9p47/9/80i6zLoRWiWbS6iCsb5DT sTyMePUeXqd3G7cceALBegMV3uS6UkENaGn4SipTNJEBZFqdm6r9xFRy8s2RfOHDR1Tj hjphlr8UtPNJCW1RfcFBjv9i5Qi2riOMyq6mWdvNQmWTmyoq14RSiW6UaO+po1RaTYDp lgJS8lHdr80SHkZgqgLcplH4bGUI7Jpw8yMGY5BiBxQqWsp2nC1QU3T/6CLpK8VW9JQZ Zego6ZK8Pz4MKubSnNKPMW/UyDziJJS0cPsoPF9TgYEcRBxjXJ+TquJdSwQ+0SWOBnYI /XJw== X-Gm-Message-State: AOAM53055RzqSeH7B/FebuQBS1zwjBoY1G4+QAgiA0C6jbEDQXDbrRAn e7MIcEsTo26TOOXQS06H6xUdIXSBVwJqGZoa39FVOaWx14D0R3nbdtfsdpBF6bo3NfAYUNXVXu8 16+sn2YXNb37SLWJgX0RSwa+Hot3mteECNChKCQBpFqRkLdtooRN7glIFYW/uT1PPLUg2Jk69AS BuAWx3oXpwg9L/U2pbzKogpHvUi1W4v5FwPwCZGKT138qv9nbvhnM03SPYhSyenK0+fFOXLar0T 2fhC5c17dDcLysHNTHZVMzTq+rOCoXzErOB0WW9dRqT2mFM32SN X-Google-Smtp-Source: ABdhPJwmjApxt5P8aH3lguIpLlBWxRANqbb12OXHEXVtSXT28uDV0+qf2zEGe0HRsOtK05NGasl80Q== X-Received: by 2002:a17:902:720c:b029:11e:787d:407e with SMTP id ba12-20020a170902720cb029011e787d407emr1518658plb.31.1623961120456; Thu, 17 Jun 2021 13:18:40 -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 h8sm6498126pgr.43.2021.06.17.13.18.39 for (version=TLS1_2 cipher=ECDHE-ECDSA-AES128-GCM-SHA256 bits=128/128); Thu, 17 Jun 2021 13:18:39 -0700 (PDT) From: Mark Dilger Content-Type: multipart/mixed; boundary="Apple-Mail=_7684FE32-4253-4C1C-BAA6-171FE7A77CC1" Mime-Version: 1.0 (Mac OS X Mail 13.4 \(3608.120.23.2.6\)) Subject: Optionally automatically disable logical replication subscriptions on error Message-Id: Date: Thu, 17 Jun 2021 13:18:38 -0700 To: PostgreSQL Hackers 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=_7684FE32-4253-4C1C-BAA6-171FE7A77CC1 Content-Transfer-Encoding: quoted-printable Content-Type: text/plain; charset=us-ascii Hackers, 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. As things currently stand, only superusers can create subscriptions. = Ongoing work to delegate superuser tasks to non-superusers creates the = potential for even more errors to be triggered, specifically, errors = where the apply worker does not have permission to make changes to the = target table. The attached patch makes it possible to create a subscription using a = new subscription_parameter, "disable_on_error", such that rather than = going into an infinite loop, the apply worker will catch errors and = automatically disable the subscription, breaking the loop. The new = parameter defaults to false. When false, the PG_TRY/PG_CATCH overhead = is avoided, so for subscriptions which do not use the feature, there = shouldn't be any change. Users can manually clear the error after = fixing the underlying issue with an ALTER SUBSCRIPTION .. ENABLE = command.=20 =20 In addition to helping on production systems, this makes writing TAP = tests involving error conditions simpler. I originally ran into the = motivation to write this patch when frustrated that TAP tests needed to = parse the apply worker log file to determine whether permission failures = were occurring and what they were. It was also obnoxiously easy to have = a test get stuck waiting for a permanently stuck subscription to catch = up. This helps with both issues. I don't think this is quite ready for commit, but I'd like feedback if = folks like this idea or want to suggest design changes. --Apple-Mail=_7684FE32-4253-4C1C-BAA6-171FE7A77CC1 Content-Disposition: attachment; filename=v1-0001-Optionally-disabling-subscriptions-on-error.patch Content-Type: application/octet-stream; x-unix-mode=0644; name="v1-0001-Optionally-disabling-subscriptions-on-error.patch" Content-Transfer-Encoding: quoted-printable =46rom=20e046be4b851ccebddabae47f57cbc0d8f49fadc8=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= v1]=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,=0A= the=20apply=20worker=20catches=20errors=20thrown,=20and=20for=20errors=20= that=20are=0Adeemed=20not=20to=20be=20transient,=20disables=20the=20= subscription=20in=20order=20to=0Abreak=20the=20loop.=20=20New=20columns=20= in=20the=20pg_subscription=20table=20help=0Adiagnose=20the=20situation:=20= =20'disabled_by_error'=20shows=20whether=20this=20has=0Aoccurred=20and=20= 'suberrmsg'=20includes=20the=20messsage=20field=20of=20the=20error.=0A= The=20error=20is=20still=20also=20written=20to=20the=20logs.=0A=0AIn=20= addition=20to=20helping=20on=20production=20systems,=20this=20makes=20= writing=20TAP=0Atests=20involving=20error=20conditions=20simpler.=20=20= Rather=20than=20having=20to=0Aopen=20and=20parse=20the=20apply=20= worker's=20log=20file,=20the=20test=20can=20query=20the=0A= pg_subscription=20table.=20=20It=20also=20helps=20that=20the=20workers=20= don't=20go=20into=0Aan=20infinite=20loop=20during=20the=20test.=0A---=0A=20= doc/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=20src/backend/catalog/pg_subscription.c=20=20= =20=20=20=20=20=20=20|=20=2010=20+=0A=20= src/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=2062=20++++++=0A=20src/backend/replication/logical/launcher.c=20=20=20= =20|=20=20=202=20+=0A=20src/backend/replication/logical/worker.c=20=20=20= =20=20=20|=20176=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=20src/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= 13=20++=0A=20.../subscription/t/022_disable_on_error.pl=20=20=20=20|=20= 125=20+++++++++++++=0A=2010=20files=20changed,=20408=20insertions(+),=20= 5=20deletions(-)=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/src/backend/catalog/pg_subscription.c=20= b/src/backend/catalog/pg_subscription.c=0Aindex=2029fc4218cd..20b51a5f1f=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,8=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->disable_on_error=20=3D=20subform->disable_on_error;=0A+=09= sub->disabled_by_error=20=3D=20subform->disabled_by_error;=0A=20=09= sub->binary=20=3D=20subform->subbinary;=0A=20=09sub->stream=20=3D=20= subform->substream;=0A=20=0A@@=20-95,6=20+97,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..49a02f9353=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,=20= disable_on_error,=0A+=09=09=09=20=20disabled_by_error,=20subbinary,=20= substream,=20subslotname,=0A+=09=09=09=20=20subsynccommit,=20= subpublications)=0A=20=20=20=20=20ON=20pg_subscription=20TO=20public;=0A= diff=20--git=20a/src/backend/commands/subscriptioncmds.c=20= b/src/backend/commands/subscriptioncmds.c=0Aindex=20= 75e195f286..098036f77e=20100644=0A---=20= a/src/backend/commands/subscriptioncmds.c=0A+++=20= b/src/backend/commands/subscriptioncmds.c=0A@@=20-63,6=20+63,8=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*disable_on_error_given,=0A+=09=09=09=09=09=09=20=20=20bool=20= *disable_on_error,=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=20bool=20*copy_data,=0A@@=20= -84,13=20+86,24=20@@=20parse_subscription_options(List=20*options,=0A=20=09= =09*connect=20=3D=20true;=0A=20=09if=20(enabled)=0A=20=09{=0A+=09=09= Assert(enabled_given);=0A+=0A=20=09=09*enabled_given=20=3D=20false;=0A=20= =09=09*enabled=20=3D=20true;=0A=20=09}=0A+=09if=20(disable_on_error)=0A+=09= {=0A+=09=09Assert(disable_on_error_given);=0A+=0A+=09=09= *disable_on_error_given=20=3D=20false;=0A+=09=09*disable_on_error=20=3D=20= false;=0A+=09}=0A=20=09if=20(create_slot)=0A=20=09=09*create_slot=20=3D=20= true;=0A=20=09if=20(slot_name)=0A=20=09{=0A+=09=09= Assert(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+115,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=20false;=0A=20=09=09*binary=20=3D=20= false;=0A=20=09}=0A=20=09if=20(streaming)=0A=20=09{=0A+=09=09= Assert(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+153,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=20= defGetBoolean(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=20disable_on_error)=0A+=09=09{=0A+=09=09=09if=20= (*disable_on_error_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= *disable_on_error_given=20=3D=20true;=0A+=09=09=09*disable_on_error=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+361,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= disable_on_error_given;=0A+=09bool=09=09disable_on_error;=0A=20=09bool=09= =09copy_data;=0A=20=09bool=09=09streaming;=0A=20=09bool=09=09= streaming_given;=0A@@=20-355,6=20+384,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&disable_on_error_given,=0A+=09= =09=09=09=09=09=09=20=20=20&disable_on_error,=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= +458,10=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_disable_on_error=20-=201]=20=3D=0A+=09=09= BoolGetDatum(disable_on_error);=0A+=09= values[Anum_pg_subscription_disabled_by_error=20-=201]=20=3D=0A+=09=09= BoolGetDatum(false);=0A=20=09values[Anum_pg_subscription_subbinary=20-=20= 1]=20=3D=20BoolGetDatum(binary);=0A=20=09= values[Anum_pg_subscription_substream=20-=201]=20=3D=20= BoolGetDatum(streaming);=0A=20=09values[Anum_pg_subscription_subconninfo=20= -=201]=20=3D=0A@@=20-438,6=20+473,8=20@@=20= CreateSubscription(CreateSubscriptionStmt=20*stmt,=20bool=20isTopLevel)=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=09= CStringGetTextDatum(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+836,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=09disable_on_error;=0A+=09=09=09=09bool=09=09= disable_on_error_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+849,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&disable_on_error_given,=0A+=09=09=09= =09=09=09=09=09=09=09=20=20=20&disable_on_error,=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= +859,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(disable_on_error_given)=0A+=09=09= =09=09{=0A+=09=09=09=09=09values[Anum_pg_subscription_disable_on_error=20= -=201]=20=3D=0A+=09=09=09=09=09=09BoolGetDatum(disable_on_error);=0A+=09=09= =09=09=09replaces[Anum_pg_subscription_disable_on_error=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+916,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"disable_on_error"=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,=20= NULL,=09/*=20no=20"slot_name"=20*/=0A=20=09=09=09=09=09=09=09=09=09=09=20= =20=20NULL,=09/*=20no=20"copy_data"=20*/=0A@@=20-885,6=20+935,15=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_disabled_by_error=20-=201]=20=3D=20false;=0A= +=09=09=09=09replaces[Anum_pg_subscription_disabled_by_error=20-=201]=20= =3D=20true;=0A+=09=09=09=09values[Anum_pg_subscription_suberrmsg=20-=20= 1]=20=3D=20CStringGetTextDatum("");=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+971,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= "disable_on_error"=20*/=0A=20=09=09=09=09=09=09=09=09=09=09=20=20=20= NULL,=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@@=20-958,6=20+1018,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= "disable_on_error"=20*/=0A=20=09=09=09=09=09=09=09=09=09=09=20=20=20= NULL,=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=20isadd=20?=20©_data=20:=20NULL,=09/*=20for=20drop,=20= no=0A@@=20-1005,6=20+1066,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= "disable_on_error"=20*/=0A=20=09=09=09=09=09=09=09=09=09=09=20=20=20= NULL,=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,=0Adiff=20--git=20= a/src/backend/replication/logical/launcher.c=20= b/src/backend/replication/logical/launcher.c=0Aindex=20= e3b11daa89..e8a32b13b6=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,8=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=09= sub->disable_on_error=20=3D=20subform->disable_on_error;=0A+=09=09= sub->disabled_by_error=20=3D=20subform->disabled_by_error;=0A=20=09=09= sub->name=20=3D=20pstrdup(NameStr(subform->subname));=0A=20=09=09/*=20We=20= don't=20fill=20fields=20we=20are=20not=20interested=20in.=20*/=0A=20=0A= diff=20--git=20a/src/backend/replication/logical/worker.c=20= b/src/backend/replication/logical/worker.c=0Aindex=20= bbb659dad0..ffd57b7efc=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,162=20@@=20= LogicalRepApplyLoop(XLogRecPtr=20last_received)=0A=20=09= walrcv_endstreaming(LogRepWorkerWalRcvConn,=20&tli);=0A=20}=0A=20=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=09/*=0A+=09=09=20*=20If=20the=20error=20is=20transient,=20network=20= protocol=20related,=20or=20resource=0A+=09=09=20*=20exhaustion=20= related,=20just=20rethrow=20it.=20=20We=20only=20want=20to=20disable=20= the=0A+=09=09=20*=20subscription=20for=20permanent=20errors.=0A+=09=09=20= */=0A+=09=09switch=20(geterrcode())=0A+=09=09{=0A+=09=09=09case=20= ERRCODE_CONNECTION_EXCEPTION:=0A+=09=09=09case=20= ERRCODE_CONNECTION_DOES_NOT_EXIST:=0A+=09=09=09case=20= ERRCODE_CONNECTION_FAILURE:=0A+=09=09=09case=20= ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION:=0A+=09=09=09case=20= ERRCODE_SQLSERVER_REJECTED_ESTABLISHMENT_OF_SQLCONNECTION:=0A+=09=09=09= case=20ERRCODE_TRANSACTION_RESOLUTION_UNKNOWN:=0A+=09=09=09case=20= ERRCODE_PROTOCOL_VIOLATION:=0A+=09=09=09case=20= ERRCODE_INSUFFICIENT_RESOURCES:=0A+=09=09=09case=20ERRCODE_DISK_FULL:=0A= +=09=09=09case=20ERRCODE_OUT_OF_MEMORY:=0A+=09=09=09case=20= ERRCODE_TOO_MANY_CONNECTIONS:=0A+=09=09=09case=20= ERRCODE_CONFIGURATION_LIMIT_EXCEEDED:=0A+=09=09=09case=20= ERRCODE_PROGRAM_LIMIT_EXCEEDED:=0A+=09=09=09case=20= ERRCODE_STATEMENT_TOO_COMPLEX:=0A+=09=09=09case=20= ERRCODE_TOO_MANY_COLUMNS:=0A+=09=09=09case=20ERRCODE_TOO_MANY_ARGUMENTS:=0A= +=09=09=09case=20ERRCODE_OPERATOR_INTERVENTION:=0A+=09=09=09case=20= ERRCODE_QUERY_CANCELED:=0A+=09=09=09case=20ERRCODE_ADMIN_SHUTDOWN:=0A+=09= =09=09case=20ERRCODE_CRASH_SHUTDOWN:=0A+=09=09=09case=20= ERRCODE_CANNOT_CONNECT_NOW:=0A+=09=09=09case=20ERRCODE_DATABASE_DROPPED:=0A= +=09=09=09case=20ERRCODE_IDLE_SESSION_TIMEOUT:=0A+=09=09=09=09= PG_RE_THROW();=0A+=09=09=09default:=0A+=09=09=09=09break;=0A+=09=09}=0A+=0A= +=09=09/*=0A+=09=09=20*=20Record=20that=20we=20had=20an=20error,=20but=20= otherwise=20don't=20do=20anything=20here=20in=20the=0A+=09=09=20*=20= catch=20block,=20as=20anything=20complicated=20we=20do=20might=20itself=20= throw.=0A+=09=09=20*/=0A+=09=09did_error=20=3D=20true;=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.=0A+=09=20*/=0A+=09if=20(did_error)=0A= +=09{=0A+=09=09Relation=09rel;=0A+=09=09bool=09=09= nulls[Natts_pg_subscription];=0A+=09=09bool=09=09= replaces[Natts_pg_subscription];=0A+=09=09Datum=09=09= values[Natts_pg_subscription];=0A+=09=09HeapTuple=09tup;=0A+=09=09Oid=09=09= =09subid;=0A+=09=09Form_pg_subscription=20form;=0A+=09=09ErrorData=20=20= *edata;=0A+=0A+=09=09/*=0A+=09=09=20*=20Clean=20up=20from=20the=20error=20= and=20get=20a=20fresh=20transaction=20in=20which=20to=0A+=09=09=20*=20= disable=20the=20subscription.=0A+=09=09=20*/=0A+=09=09= MemoryContextSwitchTo(mcxt);=0A+=09=09edata=20=3D=20CopyErrorData();=0A+=0A= +=09=09AbortOutOfAnyTransaction();=0A+=09=09FlushErrorState();=0A+=0A+=09= =09StartTransactionCommand();=0A+=0A+=09=09/*=20Look=20up=20our=20= subscription=20in=20the=20catalogs=20*/=0A+=09=09rel=20=3D=20= table_open(SubscriptionRelationId,=20RowExclusiveLock);=0A+=09=09tup=20=3D= =20SearchSysCacheCopy2(SUBSCRIPTIONNAME,=20MyDatabaseId,=0A+=09=09=09=09=09= =09=09=09=20=20CStringGetDatum(MySubscription->name));=0A+=09=09if=20= (!HeapTupleIsValid(tup))=0A+=09=09=09ereport(ERROR,=0A+=09=09=09=09=09= (errcode(ERRCODE_UNDEFINED_OBJECT),=0A+=09=09=09=09=09=20= errmsg("subscription=20\"%s\"=20does=20not=20exist",=0A+=09=09=09=09=09=09= =09MySubscription->name)));=0A+=0A+=09=09form=20=3D=20= (Form_pg_subscription)=20GETSTRUCT(tup);=0A+=09=09subid=20=3D=20= form->oid;=0A+=09=09LockSharedObject(SubscriptionRelationId,=20subid,=20= 0,=20AccessExclusiveLock);=0A+=0A+=09=09/*=0A+=09=09=20*=20We=20would=20= not=20be=20here=20unless=20this=20subscription's=20disable_on_error=0A+=09= =09=20*=20field=20was=20true=20when=20our=20worker=20began=20applying=20= changes,=20but=20check=0A+=09=09=20*=20whether=20that=20field=20has=20= changed=20in=20the=20interim.=0A+=09=09=20*/=0A+=09=09if=20= (!form->disable_on_error)=0A+=09=09=09ReThrowError(edata);=0A+=0A+=09=09= /*=20Form=20a=20new=20tuple.=20*/=0A+=09=09memset(values,=200,=20= sizeof(values));=0A+=09=09memset(nulls,=20false,=20sizeof(nulls));=0A+=09= =09memset(replaces,=20false,=20sizeof(replaces));=0A+=0A+=09=09/*=20Set=20= the=20subscription=20to=20disabled,=20and=20note=20the=20reason.=20*/=0A= +=09=09values[Anum_pg_subscription_subenabled=20-=201]=20=3D=20= BoolGetDatum(false);=0A+=09=09replaces[Anum_pg_subscription_subenabled=20= -=201]=20=3D=20true;=0A+=09=09= values[Anum_pg_subscription_disabled_by_error=20-=201]=20=3D=0A+=09=09=09= BoolGetDatum(true);=0A+=09=09= replaces[Anum_pg_subscription_disabled_by_error=20-=201]=20=3D=20true;=0A= +=09=09values[Anum_pg_subscription_suberrmsg=20-=201]=20=3D=0A+=09=09=09= CStringGetTextDatum(edata->message=20?=20edata->message=20:=20"");=0A+=09= =09replaces[Anum_pg_subscription_suberrmsg=20-=201]=20=3D=20true;=0A+=0A= +=09=09/*=20Update=20the=20catalog=20*/=0A+=09=09tup=20=3D=20= heap_modify_tuple(tup,=20RelationGetDescr(rel),=20values,=20nulls,=0A+=09= =09=09=09=09=09=09=09replaces);=0A+=09=09CatalogTupleUpdate(rel,=20= &tup->t_self,=20tup);=0A+=09=09heap_freetuple(tup);=0A+=0A+=09=09= table_close(rel,=20RowExclusiveLock);=0A+=0A+=09=09ereport(LOG,=0A+=09=09= =09=09(errmsg("edata=20is=20true=20for=20subscription=20'%s':=20message=20= =3D=20\"%s\",=20hint=20=3D=20\"%s\"",=0A+=09=09=09=09=09=09= MySubscription->name,=0A+=09=09=09=09=09=09edata->message=20?=20= edata->message=20:=20"",=0A+=09=09=09=09=09=09edata->hint=20?=20= edata->hint=20:=20"")));=0A+=0A+=09=09ereport(LOG,=0A+=09=09=09=09= (errmsg("logical=20replication=20apply=20worker=20for=20subscription=20= \"%s\"=20will=20"=0A+=09=09=09=09=09=09"stop=20because=20the=20= subscription=20was=20disabled=20due=20to=20error",=0A+=09=09=09=09=09=09= MySubscription->name)));=0A+=0A+=09=09CommitTransactionCommand();=0A+=0A= +=09=09/*=0A+=09=09=20*=20Now=20that=20we've=20successfully=20disabled=20= the=20subscription,=20rethrow=20the=0A+=09=09=20*=20error=20for=20any=20= other=20consumers=20who=20might=20want=20to=20see=20it.=20=20In=0A+=09=09= =20*=20particular,=20we=20expect=20that=20the=20worker=20will=20write=20= the=20error=20to=20the=0A+=09=09=20*=20server=20log=20and=20then=20exit.=0A= +=09=09=20*/=0A+=09=09ReThrowError(edata);=0A+=09}=0A+}=0A+=0A=20/*=0A=20= =20*=20Send=20a=20Standby=20Status=20Update=20message=20to=20server.=0A=20= =20*=0A@@=20-3138,6=20+3295,20=20@@=20ApplyWorkerMain(Datum=20main_arg)=0A= =20=09=09proc_exit(0);=0A=20=09}=0A=20=0A+=09/*=0A+=09=20*=20This=20is=20= probably=20unnecessary,=20as=20an=20error=20which=20sets=20= "disabled_by_error"=20will=20also=0A+=09=20*=20have=20cleared=20= "enabled",=20but=20for=20now,=20see=20if=20this=20ever=20resolves=20to=20= true.=0A+=09=20*/=0A+=09if=20(MySubscription->disabled_by_error)=0A+=09{=0A= +=09=09ereport(LOG,=0A+=09=09=09=09(errmsg("logical=20replication=20= apply=20worker=20for=20subscription=20\"%s\"=20will=20not=20"=0A+=09=09=09= =09=09=09"start=20because=20the=20subscription=20was=20disabled=20by=20= an=20error",=0A+=09=09=09=09=09=09MySubscription->name)));=0A+=0A+=09=09= proc_exit(0);=0A+=09}=0A+=0A=20=09/*=20Setup=20synchronous=20commit=20= according=20to=20the=20user's=20wishes=20*/=0A=20=09= SetConfigOption("synchronous_commit",=20MySubscription->synccommit,=0A=20= =09=09=09=09=09PGC_BACKEND,=20PGC_S_OVERRIDE);=0A@@=20-3241,7=20+3412,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->disable_on_error)=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..7ffa41c26b=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,12=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= disable_on_error;=09/*=20True=20if=20apply=20errors=20should=20= automatically=0A+=09=09=09=09=09=09=09=09=09=20=20=20disable=20the=20= subscription=20*/=0A+=0A+=09bool=09=09disabled_by_error;=09/*=20True=20= if=20an=20error=20caused=20the=20subscription=0A+=09=09=09=09=09=09=09=09= =09=20=20=20to=20be=20disabled=20*/=0A+=0A=20=09bool=09=09subbinary;=09=09= /*=20True=20if=20the=20subscription=20wants=20the=0A=20=09=09=09=09=09=09= =09=09=20*=20publisher=20to=20send=20data=20in=20binary=20*/=0A=20=0A@@=20= -67,6=20+73,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+100,16=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= disable_on_error;=09/*=20Whether=20errors=20automatically=20disable=20*/=0A= +=09bool=09=09disabled_by_error;=09/*=20Whether=20an=20error=20has=20= disabled=20*/=0A=20=09bool=09=09binary;=09=09=09/*=20Indicates=20if=20= the=20subscription=20wants=20data=20in=0A=20=09=09=09=09=09=09=09=09=20*=20= binary=20format=20*/=0A=20=09bool=09=09stream;=09=09=09/*=20Allow=20= streaming=20in-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=20= slot=20*/=0A=20=09char=09=20=20=20*synccommit;=09=09/*=20Synchronous=20= commit=20setting=20for=20worker=20*/=0A+=09char=09=20=20=20*errmsg;=09=09= =09/*=20Message=20from=20error=20which=20disabled=20*/=0A+=09char=09=20=20= =20*errhint;=09=09/*=20Hint=20from=20error=20which=20disabled=20*/=0A=20=09= List=09=20=20=20*publications;=09/*=20List=20of=20publication=20names=20= to=20subscribe=20to=20*/=0A=20}=20Subscription;=0A=20=0Adiff=20--git=20= a/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..b084bbab5a=0A---=20/dev/null=0A+++=20= b/src/test/subscription/t/022_disable_on_error.pl=0A@@=20-0,0=20+1,125=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>=206;=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(allows_streaming=20=3D>=20'logical');=0A= +$node_subscriber->start;=0A+=0A+#=20Create=20identical=20schema,=20= table=20and=20index=20on=20both=20the=20publisher=20and=0A+#=20= subscriber,=20with=20publications=20and=20subscriptions=20linking=20the=20= two.=0A+#=0A+my=20$publisher_connstr=20=3D=20$node_publisher->connstr=20= .=20'=20dbname=3Dpostgres';=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=20= IDENTITY=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+=09#=20= Create=20the=20publication=20for=20this=20table=0A+=09$cmd=20=3D=20qq(=0A= +CREATE=20PUBLICATION=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+#=20Create=20= an=20additional=20unique=20index=20in=20schema=20s1=20on=20the=20= subscriber=20only,=0A+#=20creating=20the=20circumstances=20for=20the=20= replication=20to=20fail=20while=20applying=20changes=0A+#=20to=20= subscription=20"s1"=20while=20succeeding=20for=20subscription=20"s2".=0A= +#=0A+$cmd=20=3D=20qq(CREATE=20UNIQUE=20INDEX=20s1_tbl_unique=20ON=20= s1.tbl=20(i));=0A+$node_subscriber->safe_psql('postgres',=20$cmd);=0A+=0A= +$node_publisher->wait_for_catchup($_)=20for=20(@schemas);=0A+=0A+#=20= Wait=20for=20initial=20sync=20to=20finish=20as=20well=0A+my=20= $synced_query=20=3D=0A+=20=20"SELECT=20count(1)=20=3D=200=20FROM=20= pg_subscription_rel=20WHERE=20srsubstate=20NOT=20IN=20('s',=20'r');";=0A= +$node_subscriber->poll_query_until('postgres',=20$synced_query)=0A+=20=20= or=20die=20"Timed=20out=20while=20waiting=20for=20subscriber=20to=20= synchronize=20data";=0A+=20=0A+#=20Enter=20non-unique=20data=20for=20= schema=20"s1"=20on=20the=20publisher.=20=20This=20should=20cause=0A+#=20= subscription=20"s1"=20on=20the=20subscriber=20to=20fail=20and=20get=20= automatically=20disabled.=0A+#=0A+$cmd=20=3D=20qq(INSERT=20INTO=20s1.tbl=20= (i)=20VALUES=20(1),=20(1),=20(1));=0A= +$node_publisher->safe_psql('postgres',=20$cmd);=0A+=0A+#=20Wait=20for=20= subscription=20"s1"=20to=20be=20automatically=20disabled.=0A+#=0A+my=20= $s1_disabled_query=20=3D=20qq(=0A+SELECT=20count(1)=20=3D=201=0A+=09FROM=20= pg_catalog.pg_subscription=0A+=09WHERE=20subname=20=3D=20's1'=0A+=09=20=20= AND=20disabled_by_error=20IS=20TRUE);=0A= +$node_subscriber->poll_query_until('postgres',=20$s1_disabled_query);=0A= +=0A+#=20Enter=20unique=20data=20for=20both=20schemas=20on=20the=20= publisher.=20=20This=20should=20succeed=20on=0A+#=20the=20publisher=20= node,=20and=20not=20cause=20any=20additional=20problems=20on=20the=20= subscriber=0A+#=20side=20either,=20though=20disabled=20subscription=20= "s1"=20should=20not=20replicate=20anything.=0A+#=0A+for=20$schema=20= (@schemas)=0A+{=0A+=09$cmd=20=3D=20qq(INSERT=20INTO=20$schema.tbl=20(i)=20= VALUES=20(2));=0A+=09$node_publisher->safe_psql('postgres',=20$cmd);=0A= +}=0A+=0A+#=20Wait=20for=20the=20data=20to=20replicate=20for=20= subscription=20"s2".=20=20This=20tests=20that=20the=0A+#=20problems=20= encountered=20by=20subscription=20"s1"=20do=20not=20cause=20subscription=20= "s2"=20to=0A+#=20get=20stuck.=0A= +$node_publisher->wait_for_catchup("s2");=0A= +$node_subscriber->poll_query_until('postgres',=20$synced_query)=0A+=20=20= or=20die=20"Timed=20out=20while=20waiting=20for=20subscriber=20to=20= synchronize=20data";=0A+=0A+#=20Subscription=20"s1"=20should=20have=20= disabled=20itself=20due=20to=20error=0A+#=0A+$cmd=20=3D=20qq(=0A+SELECT=20= disabled_by_error=20FROM=20pg_catalog.pg_subscription=20WHERE=20subname=20= =3D=20's1');=0A+is=20($node_subscriber->safe_psql('postgres',=20$cmd),=0A= +=09"t",=20"subscription=20s1=20automatically=20disabled");=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=20by=20unique=20key=20violation");=0A+=0A= +#=20Subscription=20"s2"=20should=20still=20be=20enabled=20and=20have=20= replicated=20all=20changes=0A+#=0A+$cmd=20=3D=20qq(=0A+SELECT=20= disabled_by_error=20FROM=20pg_catalog.pg_subscription=20WHERE=20subname=20= =3D=20's2');=0A+is=20($node_subscriber->safe_psql('postgres',=20$cmd),=0A= +=09"f",=20"subscription=20s2=20not=20automatically=20disabled");=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=20i=20= FROM=20s2.tbl);=0A+is=20($node_subscriber->safe_psql('postgres',=20= $cmd),=0A+=09"2",=20"subscription=20s2=20replicated=20data");=0A--=20=0A= 2.21.1=20(Apple=20Git-122.3)=0A=0A= --Apple-Mail=_7684FE32-4253-4C1C-BAA6-171FE7A77CC1 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=_7684FE32-4253-4C1C-BAA6-171FE7A77CC1--