open-dsi
/
postgres
mirror of https://github.com/postgres/postgres
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Optionally disable subscriptions on error.

Browse Source 
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 the error written to the
subscription server log, and restarting.

To partially remedy the situation, this patch adds a new subscription
option named 'disable_on_error'. To be consistent with old behavior, this
option defaults to false. When true, both the tablesync worker and apply
worker catch any errors thrown and disable the subscription in order to
break the loop. The error is still also written in the logs.

Once the subscription is disabled, users can either manually resolve the
conflict/error or skip the conflicting transaction by using
pg_replication_origin_advance() function. After resolving the conflict,
users need to enable the subscription to allow apply process to proceed.

Author: Osumi Takamichi and Mark Dilger
Reviewed-by: Greg Nancarrow, Vignesh C, Amit Kapila, Wang wei, Tang Haiying, Peter Smith, Masahiko Sawada, Shi Yu
Discussion : https://postgr.es/m/DB35438F-9356-4841-89A0-412709EBD3AB%40enterprisedb.com
pull/81/head
Amit Kapila 3 years ago
parent 369398ed88
commit 705e20f855
17 changed files with 421 additions and 108 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 10
      doc/src/sgml/catalogs.sgml
  2. 5
      doc/src/sgml/logical-replication.sgml
  3. 4
      doc/src/sgml/ref/alter_subscription.sgml
  4. 12
      doc/src/sgml/ref/create_subscription.sgml
  5. 40
      src/backend/catalog/pg_subscription.c
  6. 3
      src/backend/catalog/system_views.sql
  7. 27
      src/backend/commands/subscriptioncmds.c
  8. 162
      src/backend/replication/logical/worker.c
  9. 14
      src/bin/pg_dump/pg_dump.c
  10. 1
      src/bin/pg_dump/pg_dump.h
  11. 10
      src/bin/psql/describe.c
  12. 4
      src/bin/psql/tab-complete.c
  13. 2
      src/include/catalog/catversion.h
  14. 7
      src/include/catalog/pg_subscription.h
  15. 119
      src/test/regress/expected/subscription.out
  16. 15
      src/test/regress/sql/subscription.sql
  17. 94
      src/test/subscription/t/029_disable_on_error.pl

10
doc/src/sgml/catalogs.sgml
Unescape View File

@ -7769,6 +7769,16 @@ SCRAM-SHA-256$<replaceable>&lt;iteration count&gt;</replaceable>:<replaceable>&l
</para></entry> </para></entry>
</row> </row>
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>subdisableonerr</structfield> <type>bool</type>
</para>
<para>
If true, the subscription will be disabled if one of its workers
detects an error
</para></entry>
</row>
<row> <row>
<entry role="catalog_table_entry"><para role="column_definition"> <entry role="catalog_table_entry"><para role="column_definition">
<structfield>subconninfo</structfield> <type>text</type> <structfield>subconninfo</structfield> <type>text</type>

5
doc/src/sgml/logical-replication.sgml
Unescape View File

@ -364,8 +364,9 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
the replication origin name can be found from the server log (LSN 0/14C0378 and the replication origin name can be found from the server log (LSN 0/14C0378 and
replication origin <literal>pg_16395</literal> in the above case). To skip the replication origin <literal>pg_16395</literal> in the above case). To skip the
transaction, the subscription needs to be disabled temporarily by transaction, the subscription needs to be disabled temporarily by
<command>ALTER SUBSCRIPTION ... DISABLE</command> first. Then, the transaction <command>ALTER SUBSCRIPTION ... DISABLE</command> first or alternatively, the
can be skipped by calling the subscription can be used with the <literal>disable_on_error</literal> option.
Then, the transaction can be skipped by calling the
<link linkend="pg-replication-origin-advance"> <link linkend="pg-replication-origin-advance">
<function>pg_replication_origin_advance()</function></link> function with <function>pg_replication_origin_advance()</function></link> function with
the <parameter>node_name</parameter> (i.e., <literal>pg_16395</literal>) and the the <parameter>node_name</parameter> (i.e., <literal>pg_16395</literal>) and the

4
doc/src/sgml/ref/alter_subscription.sgml
Unescape View File

@ -204,8 +204,8 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
information. The parameters that can be altered information. The parameters that can be altered
are <literal>slot_name</literal>, are <literal>slot_name</literal>,
<literal>synchronous_commit</literal>, <literal>synchronous_commit</literal>,
<literal>binary</literal>, and <literal>binary</literal>, <literal>streaming</literal>, and
<literal>streaming</literal>. <literal>disable_on_error</literal>.
</para> </para>
</listitem> </listitem>
</varlistentry> </varlistentry>

12
doc/src/sgml/ref/create_subscription.sgml
Unescape View File

@ -290,6 +290,18 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</listitem> </listitem>
</varlistentry> </varlistentry>
<varlistentry>
<term><literal>disable_on_error</literal> (<type>boolean</type>)</term>
<listitem>
<para>
Specifies whether the subscription should be automatically disabled
if any errors are detected by subscription workers during data
replication from the publisher. The default is
<literal>false</literal>.
</para>
</listitem>
</varlistentry>
</variablelist> </variablelist>
</para> </para>

40
src/backend/catalog/pg_subscription.c
Unescape View File

@ -69,6 +69,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->binary = subform->subbinary; sub->binary = subform->subbinary;
sub->stream = subform->substream; sub->stream = subform->substream;
sub->twophasestate = subform->subtwophasestate; sub->twophasestate = subform->subtwophasestate;
sub->disableonerr = subform->subdisableonerr;
/* Get conninfo */ /* Get conninfo */
datum = SysCacheGetAttr(SUBSCRIPTIONOID, datum = SysCacheGetAttr(SUBSCRIPTIONOID,
@ -156,6 +157,45 @@ FreeSubscription(Subscription *sub)
pfree(sub); pfree(sub);
} }
/*
* Disable the given subscription.
*/
void
DisableSubscription(Oid subid)
{
Relation rel;
bool nulls[Natts_pg_subscription];
bool replaces[Natts_pg_subscription];
Datum values[Natts_pg_subscription];
HeapTuple tup;
/* Look up the subscription in the catalog */
rel = table_open(SubscriptionRelationId, RowExclusiveLock);
tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
if (!HeapTupleIsValid(tup))
elog(ERROR, "cache lookup failed for subscription %u", subid);
LockSharedObject(SubscriptionRelationId, subid, 0, AccessShareLock);
/* Form a new tuple. */
memset(values, 0, sizeof(values));
memset(nulls, false, sizeof(nulls));
memset(replaces, false, sizeof(replaces));
/* Set the subscription to disabled. */
values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(false);
replaces[Anum_pg_subscription_subenabled - 1] = true;
/* Update the catalog */
tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
replaces);
CatalogTupleUpdate(rel, &tup->t_self, tup);
heap_freetuple(tup);
table_close(rel, NoLock);
}
/* /*
* get_subscription_oid - given a subscription name, look up the OID * get_subscription_oid - given a subscription name, look up the OID
* *

3
src/backend/catalog/system_views.sql
Unescape View File

@ -1261,7 +1261,8 @@ REVOKE ALL ON pg_replication_origin_status FROM public;
-- All columns of pg_subscription except subconninfo are publicly readable. -- All columns of pg_subscription except subconninfo are publicly readable.
REVOKE ALL ON pg_subscription FROM public; REVOKE ALL ON pg_subscription FROM public;
GRANT SELECT (oid, subdbid, subname, subowner, subenabled, subbinary, GRANT SELECT (oid, subdbid, subname, subowner, subenabled, subbinary,
substream, subtwophasestate, subslotname, subsynccommit, subpublications) substream, subtwophasestate, subdisableonerr, subslotname,
subsynccommit, subpublications)
ON pg_subscription TO public; ON pg_subscription TO public;
CREATE VIEW pg_stat_subscription_stats AS CREATE VIEW pg_stat_subscription_stats AS

27
src/backend/commands/subscriptioncmds.c
Unescape View File

@ -61,6 +61,7 @@
#define SUBOPT_BINARY 0x00000080 #define SUBOPT_BINARY 0x00000080
#define SUBOPT_STREAMING 0x00000100 #define SUBOPT_STREAMING 0x00000100
#define SUBOPT_TWOPHASE_COMMIT 0x00000200 #define SUBOPT_TWOPHASE_COMMIT 0x00000200
#define SUBOPT_DISABLE_ON_ERR 0x00000400
/* check if the 'val' has 'bits' set */ /* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits)) #define IsSet(val, bits) (((val) & (bits)) == (bits))
@ -82,6 +83,7 @@ typedef struct SubOpts
bool binary; bool binary;
bool streaming; bool streaming;
bool twophase; bool twophase;
bool disableonerr;
} SubOpts; } SubOpts;
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications); static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
@ -130,6 +132,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->streaming = false; opts->streaming = false;
if (IsSet(supported_opts, SUBOPT_TWOPHASE_COMMIT)) if (IsSet(supported_opts, SUBOPT_TWOPHASE_COMMIT))
opts->twophase = false; opts->twophase = false;
if (IsSet(supported_opts, SUBOPT_DISABLE_ON_ERR))
opts->disableonerr = false;
/* Parse options */ /* Parse options */
foreach(lc, stmt_options) foreach(lc, stmt_options)
@ -249,6 +253,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->specified_opts |= SUBOPT_TWOPHASE_COMMIT; opts->specified_opts |= SUBOPT_TWOPHASE_COMMIT;
opts->twophase = defGetBoolean(defel); opts->twophase = defGetBoolean(defel);
} }
else if (IsSet(supported_opts, SUBOPT_DISABLE_ON_ERR) &&
strcmp(defel->defname, "disable_on_error") == 0)
{
if (IsSet(opts->specified_opts, SUBOPT_DISABLE_ON_ERR))
errorConflictingDefElem(defel, pstate);
opts->specified_opts |= SUBOPT_DISABLE_ON_ERR;
opts->disableonerr = defGetBoolean(defel);
}
else else
ereport(ERROR, ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR), (errcode(ERRCODE_SYNTAX_ERROR),
@ -390,7 +403,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
supported_opts = (SUBOPT_CONNECT | SUBOPT_ENABLED | SUBOPT_CREATE_SLOT | supported_opts = (SUBOPT_CONNECT | SUBOPT_ENABLED | SUBOPT_CREATE_SLOT |
SUBOPT_SLOT_NAME | SUBOPT_COPY_DATA | SUBOPT_SLOT_NAME | SUBOPT_COPY_DATA |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY | SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT); SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
SUBOPT_DISABLE_ON_ERR);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts); parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/* /*
@ -464,6 +478,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
CharGetDatum(opts.twophase ? CharGetDatum(opts.twophase ?
LOGICALREP_TWOPHASE_STATE_PENDING : LOGICALREP_TWOPHASE_STATE_PENDING :
LOGICALREP_TWOPHASE_STATE_DISABLED); LOGICALREP_TWOPHASE_STATE_DISABLED);
values[Anum_pg_subscription_subdisableonerr - 1] = BoolGetDatum(opts.disableonerr);
values[Anum_pg_subscription_subconninfo - 1] = values[Anum_pg_subscription_subconninfo - 1] =
CStringGetTextDatum(conninfo); CStringGetTextDatum(conninfo);
if (opts.slot_name) if (opts.slot_name)
@ -864,7 +879,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
{ {
supported_opts = (SUBOPT_SLOT_NAME | supported_opts = (SUBOPT_SLOT_NAME |
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY | SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING); SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR);
parse_subscription_options(pstate, stmt->options, parse_subscription_options(pstate, stmt->options,
supported_opts, &opts); supported_opts, &opts);
@ -913,6 +928,14 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
replaces[Anum_pg_subscription_substream - 1] = true; replaces[Anum_pg_subscription_substream - 1] = true;
} }
if (IsSet(opts.specified_opts, SUBOPT_DISABLE_ON_ERR))
{
values[Anum_pg_subscription_subdisableonerr - 1]
= BoolGetDatum(opts.disableonerr);
replaces[Anum_pg_subscription_subdisableonerr - 1]
= true;
}
update_tuple = true; update_tuple = true;
break; break;
} }

162
src/backend/replication/logical/worker.c
Unescape View File

@ -305,6 +305,8 @@ static void store_flush_position(XLogRecPtr remote_lsn);
static void maybe_reread_subscription(void); static void maybe_reread_subscription(void);
static void DisableSubscriptionAndExit(void);
/* prototype needed because of stream_commit */ /* prototype needed because of stream_commit */
static void apply_dispatch(StringInfo s); static void apply_dispatch(StringInfo s);
@ -3374,6 +3376,84 @@ TwoPhaseTransactionGid(Oid subid, TransactionId xid, char *gid, int szgid)
snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid); snprintf(gid, szgid, "pg_gid_%u_%u", subid, xid);
} }
/*
* Execute the initial sync with error handling. Disable the subscription,
* if it's required.
*
* Allocate the slot name in long-lived context on return. Note that we don't
* handle FATAL errors which are probably because of system resource error and
* are not repeatable.
*/
static void
start_table_sync(XLogRecPtr *origin_startpos, char **myslotname)
{
char *syncslotname;
Assert(am_tablesync_worker());
PG_TRY();
{
/* Call initial sync. */
syncslotname = LogicalRepSyncTableStart(origin_startpos);
}
PG_CATCH();
{
if (MySubscription->disableonerr)
DisableSubscriptionAndExit();
else
{
/*
* Report the worker failed during table synchronization. Abort
* the current transaction so that the stats message is sent in an
* idle state.
*/
AbortOutOfAnyTransaction();
pgstat_report_subscription_error(MySubscription->oid, false);
PG_RE_THROW();
}
}
PG_END_TRY();
/* allocate slot name in long-lived context */
*myslotname = MemoryContextStrdup(ApplyContext, syncslotname);
pfree(syncslotname);
}
/*
* Run the apply loop with error handling. Disable the subscription,
* if necessary.
*
* Note that we don't handle FATAL errors which are probably because
* of system resource error and are not repeatable.
*/
static void
start_apply(XLogRecPtr origin_startpos)
{
PG_TRY();
{
LogicalRepApplyLoop(origin_startpos);
}
PG_CATCH();
{
if (MySubscription->disableonerr)
DisableSubscriptionAndExit();
else
{
/*
* Report the worker failed while applying changes. Abort the
* current transaction so that the stats message is sent in an
* idle state.
*/
AbortOutOfAnyTransaction();
pgstat_report_subscription_error(MySubscription->oid, !am_tablesync_worker());
PG_RE_THROW();
}
}
PG_END_TRY();
}
/* Logical Replication Apply worker entry point */ /* Logical Replication Apply worker entry point */
void void
ApplyWorkerMain(Datum main_arg) ApplyWorkerMain(Datum main_arg)
@ -3381,8 +3461,8 @@ ApplyWorkerMain(Datum main_arg)
int worker_slot = DatumGetInt32(main_arg); int worker_slot = DatumGetInt32(main_arg);
MemoryContext oldctx; MemoryContext oldctx;
char originname[NAMEDATALEN]; char originname[NAMEDATALEN];
XLogRecPtr origin_startpos; XLogRecPtr origin_startpos = InvalidXLogRecPtr;
char *myslotname; char *myslotname = NULL;
WalRcvStreamOptions options; WalRcvStreamOptions options;
int server_version; int server_version;
@ -3477,32 +3557,7 @@ ApplyWorkerMain(Datum main_arg)
if (am_tablesync_worker()) if (am_tablesync_worker())
{ {
char *syncslotname; start_table_sync(&origin_startpos, &myslotname);
PG_TRY();
{
/* This is table synchronization worker, call initial sync. */
syncslotname = LogicalRepSyncTableStart(&origin_startpos);
}
PG_CATCH();
{
/*
* Abort the current transaction so that we send the stats message
* in an idle state.
*/
AbortOutOfAnyTransaction();
/* Report the worker failed during table synchronization */
pgstat_report_subscription_error(MySubscription->oid, false);
PG_RE_THROW();
}
PG_END_TRY();
/* allocate slot name in long-lived context */
myslotname = MemoryContextStrdup(ApplyContext, syncslotname);
pfree(syncslotname);
/* /*
* Allocate the origin name in long-lived context for error context * Allocate the origin name in long-lived context for error context
@ -3633,24 +3688,43 @@ ApplyWorkerMain(Datum main_arg)
} }
/* Run the main loop. */ /* Run the main loop. */
PG_TRY(); start_apply(origin_startpos);
{
LogicalRepApplyLoop(origin_startpos);
}
PG_CATCH();
{
/*
* Abort the current transaction so that we send the stats message in
* an idle state.
*/
AbortOutOfAnyTransaction();
/* Report the worker failed while applying changes */ proc_exit(0);
pgstat_report_subscription_error(MySubscription->oid, !am_tablesync_worker()); }
PG_RE_THROW(); /*
} * After error recovery, disable the subscription in a new transaction
PG_END_TRY(); * and exit cleanly.
*/
static void
DisableSubscriptionAndExit(void)
{
/*
* Emit the error message, and recover from the error state to an idle
* state
*/
HOLD_INTERRUPTS();
EmitErrorReport();
AbortOutOfAnyTransaction();
FlushErrorState();
RESUME_INTERRUPTS();
/* Report the worker failed during either table synchronization or apply */
pgstat_report_subscription_error(MyLogicalRepWorker->subid,
!am_tablesync_worker());
/* Disable the subscription */
StartTransactionCommand();
DisableSubscription(MySubscription->oid);
CommitTransactionCommand();
/* Notify the subscription has been disabled and exit */
ereport(LOG,
errmsg("logical replication subscription \"%s\" has been disabled due to an error",
MySubscription->name));
proc_exit(0); proc_exit(0);
} }

14
src/bin/pg_dump/pg_dump.c
Unescape View File

@ -4293,6 +4293,7 @@ getSubscriptions(Archive *fout)
int i_subowner; int i_subowner;
int i_substream; int i_substream;
int i_subtwophasestate; int i_subtwophasestate;
int i_subdisableonerr;
int i_subconninfo; int i_subconninfo;
int i_subslotname; int i_subslotname;
int i_subsynccommit; int i_subsynccommit;
@ -4340,10 +4341,13 @@ getSubscriptions(Archive *fout)
appendPQExpBufferStr(query, " false AS substream,\n"); appendPQExpBufferStr(query, " false AS substream,\n");
if (fout->remoteVersion >= 150000) if (fout->remoteVersion >= 150000)
appendPQExpBufferStr(query, " s.subtwophasestate\n"); appendPQExpBufferStr(query,
" s.subtwophasestate,\n"
" s.subdisableonerr\n");
else else
appendPQExpBuffer(query, appendPQExpBuffer(query,
" '%c' AS subtwophasestate\n", " '%c' AS subtwophasestate,\n"
" false AS subdisableonerr\n",
LOGICALREP_TWOPHASE_STATE_DISABLED); LOGICALREP_TWOPHASE_STATE_DISABLED);
appendPQExpBufferStr(query, appendPQExpBufferStr(query,
@ -4366,6 +4370,7 @@ getSubscriptions(Archive *fout)
i_subbinary = PQfnumber(res, "subbinary"); i_subbinary = PQfnumber(res, "subbinary");
i_substream = PQfnumber(res, "substream"); i_substream = PQfnumber(res, "substream");
i_subtwophasestate = PQfnumber(res, "subtwophasestate"); i_subtwophasestate = PQfnumber(res, "subtwophasestate");
i_subdisableonerr = PQfnumber(res, "subdisableonerr");
subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo)); subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo));
@ -4393,6 +4398,8 @@ getSubscriptions(Archive *fout)
pg_strdup(PQgetvalue(res, i, i_substream)); pg_strdup(PQgetvalue(res, i, i_substream));
subinfo[i].subtwophasestate = subinfo[i].subtwophasestate =
pg_strdup(PQgetvalue(res, i, i_subtwophasestate)); pg_strdup(PQgetvalue(res, i, i_subtwophasestate));
subinfo[i].subdisableonerr =
pg_strdup(PQgetvalue(res, i, i_subdisableonerr));
/* Decide whether we want to dump it */ /* Decide whether we want to dump it */
selectDumpableObject(&(subinfo[i].dobj), fout); selectDumpableObject(&(subinfo[i].dobj), fout);
@ -4463,6 +4470,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
if (strcmp(subinfo->subtwophasestate, two_phase_disabled) != 0) if (strcmp(subinfo->subtwophasestate, two_phase_disabled) != 0)
appendPQExpBufferStr(query, ", two_phase = on"); appendPQExpBufferStr(query, ", two_phase = on");
if (strcmp(subinfo->subdisableonerr, "t") == 0)
appendPQExpBufferStr(query, ", disable_on_error = true");
if (strcmp(subinfo->subsynccommit, "off") != 0) if (strcmp(subinfo->subsynccommit, "off") != 0)
appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit)); appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit));

1
src/bin/pg_dump/pg_dump.h
Unescape View File

@ -657,6 +657,7 @@ typedef struct _SubscriptionInfo
char *subbinary; char *subbinary;
char *substream; char *substream;
char *subtwophasestate; char *subtwophasestate;
char *subdisableonerr;
char *subsynccommit; char *subsynccommit;
char *subpublications; char *subpublications;
} SubscriptionInfo; } SubscriptionInfo;

10
src/bin/psql/describe.c
Unescape View File

@ -6084,7 +6084,7 @@ describeSubscriptions(const char *pattern, bool verbose)
PGresult *res; PGresult *res;
printQueryOpt myopt = pset.popt; printQueryOpt myopt = pset.popt;
static const bool translate_columns[] = {false, false, false, false, static const bool translate_columns[] = {false, false, false, false,
false, false, false, false, false}; false, false, false, false, false, false};
if (pset.sversion < 100000) if (pset.sversion < 100000)
{ {
@ -6118,11 +6118,13 @@ describeSubscriptions(const char *pattern, bool verbose)
gettext_noop("Binary"), gettext_noop("Binary"),
gettext_noop("Streaming")); gettext_noop("Streaming"));
/* Two_phase is only supported in v15 and higher */ /* Two_phase and disable_on_error are only supported in v15 and higher */
if (pset.sversion >= 150000) if (pset.sversion >= 150000)
appendPQExpBuffer(&buf, appendPQExpBuffer(&buf,
", subtwophasestate AS \"%s\"\n", ", subtwophasestate AS \"%s\"\n"
gettext_noop("Two phase commit")); ", subdisableonerr AS \"%s\"\n",
gettext_noop("Two phase commit"),
gettext_noop("Disable on error"));
appendPQExpBuffer(&buf, appendPQExpBuffer(&buf,
", subsynccommit AS \"%s\"\n" ", subsynccommit AS \"%s\"\n"

4
src/bin/psql/tab-complete.c
Unescape View File

@ -1834,7 +1834,7 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH("(", "PUBLICATION"); COMPLETE_WITH("(", "PUBLICATION");
/* ALTER SUBSCRIPTION <name> SET ( */ /* ALTER SUBSCRIPTION <name> SET ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "(")) else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
COMPLETE_WITH("binary", "slot_name", "streaming", "synchronous_commit"); COMPLETE_WITH("binary", "slot_name", "streaming", "synchronous_commit", "disable_on_error");
/* ALTER SUBSCRIPTION <name> SET PUBLICATION */ /* ALTER SUBSCRIPTION <name> SET PUBLICATION */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "PUBLICATION")) else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "PUBLICATION"))
{ {
@ -3104,7 +3104,7 @@ psql_completion(const char *text, int start, int end)
else if (HeadMatches("CREATE", "SUBSCRIPTION") && TailMatches("WITH", "(")) else if (HeadMatches("CREATE", "SUBSCRIPTION") && TailMatches("WITH", "("))
COMPLETE_WITH("binary", "connect", "copy_data", "create_slot", COMPLETE_WITH("binary", "connect", "copy_data", "create_slot",
"enabled", "slot_name", "streaming", "enabled", "slot_name", "streaming",
"synchronous_commit", "two_phase"); "synchronous_commit", "two_phase", "disable_on_error");
/* CREATE TRIGGER --- is allowed inside CREATE SCHEMA, so use TailMatches */ /* CREATE TRIGGER --- is allowed inside CREATE SCHEMA, so use TailMatches */

2
src/include/catalog/catversion.h
Unescape View File

@ -53,6 +53,6 @@
*/ */
/* yyyymmddN */ /* yyyymmddN */
#define CATALOG_VERSION_NO 202203031 #define CATALOG_VERSION_NO 202203141
#endif #endif

7
src/include/catalog/pg_subscription.h
Unescape View File

@ -67,6 +67,9 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
char subtwophasestate; /* Stream two-phase transactions */ char subtwophasestate; /* Stream two-phase transactions */
bool subdisableonerr; /* True if a worker error should cause the
* subscription to be disabled */
#ifdef CATALOG_VARLEN /* variable-length fields start here */ #ifdef CATALOG_VARLEN /* variable-length fields start here */
/* Connection string to the publisher */ /* Connection string to the publisher */
text subconninfo BKI_FORCE_NOT_NULL; text subconninfo BKI_FORCE_NOT_NULL;
@ -103,6 +106,9 @@ typedef struct Subscription
* binary format */ * binary format */
bool stream; /* Allow streaming in-progress transactions. */ bool stream; /* Allow streaming in-progress transactions. */
char twophasestate; /* Allow streaming two-phase transactions */ char twophasestate; /* Allow streaming two-phase transactions */
bool disableonerr; /* Indicates if the subscription should be
* automatically disabled if a worker error
* occurs */
char *conninfo; /* Connection string to the publisher */ char *conninfo; /* Connection string to the publisher */
char *slotname; /* Name of the replication slot */ char *slotname; /* Name of the replication slot */
char *synccommit; /* Synchronous commit setting for worker */ char *synccommit; /* Synchronous commit setting for worker */
@ -111,6 +117,7 @@ typedef struct Subscription
extern Subscription *GetSubscription(Oid subid, bool missing_ok); extern Subscription *GetSubscription(Oid subid, bool missing_ok);
extern void FreeSubscription(Subscription *sub); extern void FreeSubscription(Subscription *sub);
extern void DisableSubscription(Oid subid);
extern Oid get_subscription_oid(const char *subname, bool missing_ok); extern Oid get_subscription_oid(const char *subname, bool missing_ok);
extern char *get_subscription_name(Oid subid, bool missing_ok); extern char *get_subscription_name(Oid subid, bool missing_ok);

119
src/test/regress/expected/subscription.out
Unescape View File

@ -76,10 +76,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
\dRs+ \dRs+
List of subscriptions List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
-----------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+----------------------------- -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | off | dbname=regress_doesnotexist regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist
(1 row) (1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false); ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@ -94,10 +94,10 @@ ERROR: subscription "regress_doesnotexist" does not exist
ALTER SUBSCRIPTION regress_testsub SET (create_slot = false); ALTER SUBSCRIPTION regress_testsub SET (create_slot = false);
ERROR: unrecognized subscription parameter: "create_slot" ERROR: unrecognized subscription parameter: "create_slot"
\dRs+ \dRs+
List of subscriptions List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+--------------------+------------------------------ -----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------------------+------------------------------
regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | f | d | off | dbname=regress_doesnotexist2 regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | f | d | f | off | dbname=regress_doesnotexist2
(1 row) (1 row)
BEGIN; BEGIN;
@ -129,10 +129,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
ERROR: invalid value for parameter "synchronous_commit": "foobar" ERROR: invalid value for parameter "synchronous_commit": "foobar"
HINT: Available values: local, remote_write, remote_apply, on, off. HINT: Available values: local, remote_write, remote_apply, on, off.
\dRs+ \dRs+
List of subscriptions List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+--------------------+------------------------------ ---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------------------+------------------------------
regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | f | d | local | dbname=regress_doesnotexist2 regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | f | d | f | local | dbname=regress_doesnotexist2
(1 row) (1 row)
-- rename back to keep the rest simple -- rename back to keep the rest simple
@ -165,19 +165,19 @@ ERROR: binary requires a Boolean value
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, binary = true); CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, binary = true);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
\dRs+ \dRs+
List of subscriptions List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
-----------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+----------------------------- -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
regress_testsub | regress_subscription_user | f | {testpub} | t | f | d | off | dbname=regress_doesnotexist regress_testsub | regress_subscription_user | f | {testpub} | t | f | d | f | off | dbname=regress_doesnotexist
(1 row) (1 row)
ALTER SUBSCRIPTION regress_testsub SET (binary = false); ALTER SUBSCRIPTION regress_testsub SET (binary = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+ \dRs+
List of subscriptions List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
-----------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+----------------------------- -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | off | dbname=regress_doesnotexist regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist
(1 row) (1 row)
DROP SUBSCRIPTION regress_testsub; DROP SUBSCRIPTION regress_testsub;
@ -188,19 +188,19 @@ ERROR: streaming requires a Boolean value
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, streaming = true); CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, streaming = true);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
\dRs+ \dRs+
List of subscriptions List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
-----------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+----------------------------- -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
regress_testsub | regress_subscription_user | f | {testpub} | f | t | d | off | dbname=regress_doesnotexist regress_testsub | regress_subscription_user | f | {testpub} | f | t | d | f | off | dbname=regress_doesnotexist
(1 row) (1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false); ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+ \dRs+
List of subscriptions List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
-----------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+----------------------------- -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | off | dbname=regress_doesnotexist regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist
(1 row) (1 row)
-- fail - publication already exists -- fail - publication already exists
@ -215,10 +215,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false); ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
ERROR: publication "testpub1" is already in subscription "regress_testsub" ERROR: publication "testpub1" is already in subscription "regress_testsub"
\dRs+ \dRs+
List of subscriptions List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+--------------------+----------------------------- -----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | f | d | off | dbname=regress_doesnotexist regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | f | d | f | off | dbname=regress_doesnotexist
(1 row) (1 row)
-- fail - publication used more then once -- fail - publication used more then once
@ -233,10 +233,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub"
-- ok - delete publications -- ok - delete publications
ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false); ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
\dRs+ \dRs+
List of subscriptions List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
-----------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+----------------------------- -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | off | dbname=regress_doesnotexist regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist
(1 row) (1 row)
DROP SUBSCRIPTION regress_testsub; DROP SUBSCRIPTION regress_testsub;
@ -270,10 +270,10 @@ ERROR: two_phase requires a Boolean value
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true); CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
\dRs+ \dRs+
List of subscriptions List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
-----------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+----------------------------- -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
regress_testsub | regress_subscription_user | f | {testpub} | f | f | p | off | dbname=regress_doesnotexist regress_testsub | regress_subscription_user | f | {testpub} | f | f | p | f | off | dbname=regress_doesnotexist
(1 row) (1 row)
--fail - alter of two_phase option not supported. --fail - alter of two_phase option not supported.
@ -282,10 +282,10 @@ ERROR: unrecognized subscription parameter: "two_phase"
-- but can alter streaming when two_phase enabled -- but can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true); ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+ \dRs+
List of subscriptions List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
-----------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+----------------------------- -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
regress_testsub | regress_subscription_user | f | {testpub} | f | t | p | off | dbname=regress_doesnotexist regress_testsub | regress_subscription_user | f | {testpub} | f | t | p | f | off | dbname=regress_doesnotexist
(1 row) (1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@ -294,10 +294,33 @@ DROP SUBSCRIPTION regress_testsub;
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, streaming = true, two_phase = true); CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, streaming = true, two_phase = true);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
\dRs+ \dRs+
List of subscriptions List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Synchronous commit | Conninfo Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
-----------------+---------------------------+---------+-------------+--------+-----------+------------------+--------------------+----------------------------- -----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
regress_testsub | regress_subscription_user | f | {testpub} | f | t | p | off | dbname=regress_doesnotexist regress_testsub | regress_subscription_user | f | {testpub} | f | t | p | f | off | dbname=regress_doesnotexist
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
-- fail - disable_on_error must be boolean
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, disable_on_error = foo);
ERROR: disable_on_error requires a Boolean value
-- now it works
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, disable_on_error = false);
WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables
\dRs+
List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | f | off | dbname=regress_doesnotexist
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
\dRs+
List of subscriptions
Name | Owner | Enabled | Publication | Binary | Streaming | Two phase commit | Disable on error | Synchronous commit | Conninfo
-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------------------+-----------------------------
regress_testsub | regress_subscription_user | f | {testpub} | f | f | d | t | off | dbname=regress_doesnotexist
(1 row) (1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);

15
src/test/regress/sql/subscription.sql
Unescape View File

@ -228,6 +228,21 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE); ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub; DROP SUBSCRIPTION regress_testsub;
-- fail - disable_on_error must be boolean
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, disable_on_error = foo);
-- now it works
CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, disable_on_error = false);
\dRs+
ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
\dRs+
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
RESET SESSION AUTHORIZATION; RESET SESSION AUTHORIZATION;
DROP ROLE regress_subscription_user; DROP ROLE regress_subscription_user;
DROP ROLE regress_subscription_user2; DROP ROLE regress_subscription_user2;

94
src/test/subscription/t/029_disable_on_error.pl
Unescape View File

@ -0,0 +1,94 @@
# Copyright (c) 2021-2022, PostgreSQL Global Development Group
# Test of logical replication subscription self-disabling feature.
use strict;
use warnings;
use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
# create publisher node
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
$node_publisher->start;
# create subscriber node
my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_subscriber->init;
$node_subscriber->start;
# Create identical table on both nodes.
$node_publisher->safe_psql('postgres', "CREATE TABLE tbl (i INT)");
$node_subscriber->safe_psql('postgres', "CREATE TABLE tbl (i INT)");
# Insert duplicate values on the publisher.
$node_publisher->safe_psql('postgres',
"INSERT INTO tbl (i) VALUES (1), (1), (1)");
# Create an additional unique index on the subscriber.
$node_subscriber->safe_psql('postgres',
"CREATE UNIQUE INDEX tbl_unique ON tbl (i)");
# Create a pub/sub to set up logical replication. This tests that the
# uniqueness violation will cause the subscription to fail during initial
# synchronization and make it disabled.
my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
$node_publisher->safe_psql('postgres',
"CREATE PUBLICATION pub FOR TABLE tbl");
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION sub CONNECTION '$publisher_connstr' PUBLICATION pub WITH (disable_on_error = true)"
);
# Initial synchronization failure causes the subscription to be disabled.
$node_subscriber->poll_query_until('postgres',
"SELECT subenabled = false FROM pg_catalog.pg_subscription WHERE subname = 'sub'"
) or die "Timed out while waiting for subscriber to be disabled";
# Drop the unique index on the subscriber which caused the subscription to be
# disabled.
$node_subscriber->safe_psql('postgres', "DROP INDEX tbl_unique");
# Re-enable the subscription "sub".
$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub ENABLE");
# Wait for the data to replicate.
$node_publisher->wait_for_catchup('sub');
$node_subscriber->poll_query_until('postgres',
"SELECT COUNT(1) = 0 FROM pg_subscription_rel sr WHERE sr.srsubstate NOT IN ('s', 'r') AND sr.srrelid = 'tbl'::regclass"
);
# Confirm that we have finished the table sync.
my $result =
$node_subscriber->safe_psql('postgres', "SELECT MAX(i), COUNT(*) FROM tbl");
is($result, qq(1|3), "subscription sub replicated data");
# Delete the data from the subscriber and recreate the unique index.
$node_subscriber->safe_psql('postgres', "DELETE FROM tbl");
$node_subscriber->safe_psql('postgres',
"CREATE UNIQUE INDEX tbl_unique ON tbl (i)");
# Add more non-unique data to the publisher.
$node_publisher->safe_psql('postgres',
"INSERT INTO tbl (i) VALUES (3), (3), (3)");
# Apply failure causes the subscription to be disabled.
$node_subscriber->poll_query_until('postgres',
"SELECT subenabled = false FROM pg_catalog.pg_subscription WHERE subname = 'sub'"
) or die "Timed out while waiting for subscription sub to be disabled";
# Drop the unique index on the subscriber and re-enabled the subscription. Then
# confirm that the previously failing insert was applied OK.
$node_subscriber->safe_psql('postgres', "DROP INDEX tbl_unique");
$node_subscriber->safe_psql('postgres', "ALTER SUBSCRIPTION sub ENABLE");
$node_publisher->wait_for_catchup('sub');
$result = $node_subscriber->safe_psql('postgres',
"SELECT COUNT(*) FROM tbl WHERE i = 3");
is($result, qq(3), 'check the result of apply');
$node_subscriber->stop;
$node_publisher->stop;
done_testing();
Write Preview
Loading…
Cancel
Save