v27-0004-add-a-max_conflict_retention_duration-subscripti.patch
application/octet-stream
Filename: v27-0004-add-a-max_conflict_retention_duration-subscripti.patch
Type: application/octet-stream
Part: 3
Patch
Format: format-patch
Series: patch v27-0004
Subject: add a max_conflict_retention_duration subscription option
| File | + | − |
|---|---|---|
| doc/src/sgml/catalogs.sgml | 10 | 0 |
| doc/src/sgml/monitoring.sgml | 13 | 0 |
| doc/src/sgml/ref/alter_subscription.sgml | 3 | 2 |
| doc/src/sgml/ref/create_subscription.sgml | 40 | 0 |
| doc/src/sgml/system-views.sgml | 11 | 0 |
| src/backend/catalog/pg_subscription.c | 1 | 0 |
| src/backend/catalog/system_views.sql | 4 | 3 |
| src/backend/commands/subscriptioncmds.c | 28 | 4 |
| src/backend/replication/logical/launcher.c | 71 | 8 |
| src/backend/replication/logical/worker.c | 130 | 14 |
| src/backend/replication/slot.c | 12 | 1 |
| src/bin/pg_dump/pg_dump.c | 17 | 2 |
| src/bin/pg_dump/pg_dump.h | 1 | 0 |
| src/bin/psql/describe.c | 7 | 1 |
| src/bin/psql/tab-complete.in.c | 8 | 8 |
| src/include/catalog/pg_proc.dat | 3 | 3 |
| src/include/catalog/pg_subscription.h | 10 | 0 |
| src/include/replication/slot.h | 2 | 0 |
| src/include/replication/worker_internal.h | 6 | 0 |
| src/test/regress/expected/rules.out | 3 | 2 |
| src/test/regress/expected/subscription.out | 109 | 84 |
| src/test/regress/sql/subscription.sql | 16 | 0 |
From 7cf1ebb2603f0cc47ea7e29a99c9f26c17dfe0a2 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <houzj.fnst@cn.fujitsu.com>
Date: Wed, 15 Jan 2025 16:46:15 +0800
Subject: [PATCH v27 4/7] add a max_conflict_retention_duration subscription
option
This commit introduces the max_conflict_retention_duration subscription option,
designed to prevent excessive accumulation of dead tuples when
retain_conflict_info is enabled and the the apply worker cannot catch up with
the publisher's workload.
If the time spent advancing non-removable transaction ID surpasses the
max_conflict_retention_duration threshold, the apply worker would stop
retaining information for conflict detection. The replication slot
pg_conflict_detection will be invalidated if all apply workers associated with
the subscription, where retain_conflict_info is enabled, confirm that the
retention duration exceeded the max_conflict_retention_duration.
In this patch, a replication slot will not be automatically re-created if it
becomes invalidated. Users can disable retain_conflict_info and re-enable it
after confirming that the replication slot has been dropped. An upcoming patch
will include support for automatic slot recreation once at least one apply
worker confirms that the retention duration is within the
max_conflict_retention_duration limit.
To monitor worker's conflict retention status, this patch also introduces a new
column 'retain_conflict_info' in the pg_stat_subscription view. This column
indicates whether the apply worker is effectively retaining conflict
information. The value is set to true only if retain_conflict_info is enabled
for the associated subscription, and the retention duration for conflict
detection by the apply worker has not exceeded max_conflict_retention_duration.
---
doc/src/sgml/catalogs.sgml | 10 ++
doc/src/sgml/monitoring.sgml | 13 ++
doc/src/sgml/ref/alter_subscription.sgml | 5 +-
doc/src/sgml/ref/create_subscription.sgml | 40 +++++
doc/src/sgml/system-views.sgml | 11 ++
src/backend/catalog/pg_subscription.c | 1 +
src/backend/catalog/system_views.sql | 7 +-
src/backend/commands/subscriptioncmds.c | 32 +++-
src/backend/replication/logical/launcher.c | 79 ++++++++-
src/backend/replication/logical/worker.c | 144 +++++++++++++--
src/backend/replication/slot.c | 13 +-
src/bin/pg_dump/pg_dump.c | 19 +-
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/psql/describe.c | 8 +-
src/bin/psql/tab-complete.in.c | 16 +-
src/include/catalog/pg_proc.dat | 6 +-
src/include/catalog/pg_subscription.h | 10 ++
src/include/replication/slot.h | 2 +
src/include/replication/worker_internal.h | 6 +
src/test/regress/expected/rules.out | 5 +-
src/test/regress/expected/subscription.out | 193 ++++++++++++---------
src/test/regress/sql/subscription.sql | 16 ++
22 files changed, 505 insertions(+), 132 deletions(-)
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 3abb86cf8c2..d609285dcae 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -8077,6 +8077,16 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>submaxconflictretentionduration</structfield> <type>interger</type>
+ </para>
+ <para>
+ If not zero, the maximum duration (in milliseconds) for which conflict
+ information can be retained for conflict detection by the apply worker.
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>subconninfo</structfield> <type>text</type>
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index edc2470bcf9..0375c15f76c 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -2101,6 +2101,19 @@ description | Waiting for a newly initialized WAL file to reach durable storage
sender; NULL for parallel apply workers
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>retain_conflict_info</structfield> <type>boolean</type>
+ </para>
+ <para>
+ True if <link linkend="sql-createsubscription-params-with-retain-conflict-info"><literal>retain_conflict_info</literal></link>
+ is enabled and the duration for which conflict information is
+ retained for conflict detection by this apply worker does not exceed
+ <link linkend="sql-createsubscription-params-with-max-conflict-retention-duration"><literal>max_conflict_retention_duration</literal></link>; NULL for
+ parallel apply workers and table synchronization workers.
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index e5415c3150d..b772cefe075 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -236,8 +236,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-run-as-owner"><literal>run_as_owner</literal></link>,
<link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>,
<link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>,
- <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>, and
- <link linkend="sql-createsubscription-params-with-retain-conflict-info"><literal>retain_conflict_info</literal></link>.
+ <link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>,
+ <link linkend="sql-createsubscription-params-with-retain-conflict-info"><literal>retain_conflict_info</literal></link>, and
+ <link linkend="sql-createsubscription-params-with-max-conflict-retention-duration"><literal>max_conflict_retention_duration</literal></link>.
Only a superuser can set <literal>password_required = false</literal>.
</para>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 61374b74d1c..fef6c0757e3 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -466,6 +466,46 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</para>
</listitem>
</varlistentry>
+
+ <varlistentry id="sql-createsubscription-params-with-max-conflict-retention-duration">
+ <term><literal>max_conflict_retention_duration</literal> (<type>integer</type>)</term>
+ <listitem>
+ <para>
+ Specifies the maximum duration (in milliseconds) for which conflict
+ information can be retained for conflict detection by the apply worker.
+ The default value is <literal>0</literal>, indicating that conflict
+ information is retained until it is no longer needed for detection
+ purposes.
+ </para>
+
+ <para>
+ The replication slot
+ <quote><literal>pg_conflict_detection</literal></quote> that used to
+ retain conflict information will be invalidated if all apply workers
+ associated with the subscription, where
+ <literal>retain_conflict_info</literal> is enabled, confirm that the
+ retention duration exceeded the
+ <literal>max_conflict_retention_duration</literal>. If the replication
+ slot is invalidated, you can disable
+ <literal>retain_conflict_info</literal> and re-enable it after
+ confirming this replication slot has been dropped.
+ </para>
+
+ <para>
+ This option is effective only when
+ <literal>retain_conflict_info</literal> is enabled and the apply
+ worker associated with the subscription is active.
+ </para>
+
+ <warning>
+ <para>
+ Note that setting a non-zero value for this option could lead to
+ conflict information being removed prematurely, potentially missing
+ some conflict detections.
+ </para>
+ </warning>
+ </listitem>
+ </varlistentry>
</variablelist></para>
</listitem>
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index be81c2b51d2..c795eb21921 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2621,6 +2621,17 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
perform logical decoding. It is set only for logical slots.
</para>
</listitem>
+ <listitem>
+ <para>
+ <literal>conflict_retention_exceeds_max_duration</literal> means that
+ the duration for retaining conflict information, which is used
+ in logical replication conflict detection, has exceeded the maximum
+ allowable limit. It is set only for the slot
+ <literal>pg_conflict_detection</literal>, which is created when
+ <link linkend="sql-createsubscription-params-with-retain-conflict-info"><literal>retain_conflict_info</literal></link>
+ is enabled.
+ </para>
+ </listitem>
</itemizedlist>
</para></entry>
</row>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 39cfae43d6f..d3d9cc1779c 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -104,6 +104,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->runasowner = subform->subrunasowner;
sub->failover = subform->subfailover;
sub->retainconflictinfo = subform->subretainconflictinfo;
+ sub->maxconflictretentionduration = subform->submaxconflictretentionduration;
/* Get conninfo */
datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 7659bf8373f..930013c5984 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -980,7 +980,8 @@ CREATE VIEW pg_stat_subscription AS
st.last_msg_send_time,
st.last_msg_receipt_time,
st.latest_end_lsn,
- st.latest_end_time
+ st.latest_end_time,
+ st.retain_conflict_info
FROM pg_subscription su
LEFT JOIN pg_stat_get_subscription(NULL) st
ON (st.subid = su.oid);
@@ -1371,8 +1372,8 @@ REVOKE ALL ON pg_subscription FROM public;
GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
subbinary, substream, subtwophasestate, subdisableonerr,
subpasswordrequired, subrunasowner, subfailover,
- subretainconflictinfo, subslotname, subsynccommit,
- subpublications, suborigin)
+ subretainconflictinfo, submaxconflictretentionduration, subslotname,
+ subsynccommit, subpublications, suborigin)
ON pg_subscription TO public;
CREATE VIEW pg_stat_subscription_stats AS
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 51bd52e48a3..13199740320 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -73,8 +73,9 @@
#define SUBOPT_RUN_AS_OWNER 0x00001000
#define SUBOPT_FAILOVER 0x00002000
#define SUBOPT_RETAIN_CONFLICT_INFO 0x00004000
-#define SUBOPT_LSN 0x00008000
-#define SUBOPT_ORIGIN 0x00010000
+#define SUBOPT_MAX_CONFLICT_RETENTION_DURATION 0x00008000
+#define SUBOPT_LSN 0x00010000
+#define SUBOPT_ORIGIN 0x00020000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -101,6 +102,7 @@ typedef struct SubOpts
bool runasowner;
bool failover;
bool retainconflictinfo;
+ int32 maxconflictretentionduration;
char *origin;
XLogRecPtr lsn;
} SubOpts;
@@ -169,6 +171,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->failover = false;
if (IsSet(supported_opts, SUBOPT_RETAIN_CONFLICT_INFO))
opts->retainconflictinfo = false;
+ if (IsSet(supported_opts, SUBOPT_MAX_CONFLICT_RETENTION_DURATION))
+ opts->maxconflictretentionduration = 0;
if (IsSet(supported_opts, SUBOPT_ORIGIN))
opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
@@ -323,6 +327,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->specified_opts |= SUBOPT_RETAIN_CONFLICT_INFO;
opts->retainconflictinfo = defGetBoolean(defel);
}
+ else if (IsSet(supported_opts, SUBOPT_MAX_CONFLICT_RETENTION_DURATION) &&
+ strcmp(defel->defname, "max_conflict_retention_duration") == 0)
+ {
+ if (IsSet(opts->specified_opts, SUBOPT_MAX_CONFLICT_RETENTION_DURATION))
+ errorConflictingDefElem(defel, pstate);
+
+ opts->specified_opts |= SUBOPT_MAX_CONFLICT_RETENTION_DURATION;
+ opts->maxconflictretentionduration = defGetInt32(defel);
+ }
else if (IsSet(supported_opts, SUBOPT_ORIGIN) &&
strcmp(defel->defname, "origin") == 0)
{
@@ -580,7 +593,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED |
SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
- SUBOPT_RETAIN_CONFLICT_INFO | SUBOPT_ORIGIN);
+ SUBOPT_RETAIN_CONFLICT_INFO | SUBOPT_MAX_CONFLICT_RETENTION_DURATION |
+ SUBOPT_ORIGIN);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/*
@@ -695,6 +709,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
values[Anum_pg_subscription_subfailover - 1] = BoolGetDatum(opts.failover);
values[Anum_pg_subscription_subretainconflictinfo - 1] =
BoolGetDatum(opts.retainconflictinfo);
+ values[Anum_pg_subscription_submaxconflictretentionduration - 1] =
+ Int32GetDatum(opts.maxconflictretentionduration);
values[Anum_pg_subscription_subconninfo - 1] =
CStringGetTextDatum(conninfo);
if (opts.slot_name)
@@ -1193,7 +1209,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
SUBOPT_DISABLE_ON_ERR |
SUBOPT_PASSWORD_REQUIRED |
SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
- SUBOPT_RETAIN_CONFLICT_INFO | SUBOPT_ORIGIN);
+ SUBOPT_RETAIN_CONFLICT_INFO |
+ SUBOPT_MAX_CONFLICT_RETENTION_DURATION | SUBOPT_ORIGIN);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1376,6 +1393,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
retain_conflict_info = opts.retainconflictinfo;
}
+ if (IsSet(opts.specified_opts, SUBOPT_MAX_CONFLICT_RETENTION_DURATION))
+ {
+ values[Anum_pg_subscription_submaxconflictretentionduration - 1] =
+ Int32GetDatum(opts.maxconflictretentionduration);
+ replaces[Anum_pg_subscription_submaxconflictretentionduration - 1] = true;
+ }
+
if (IsSet(opts.specified_opts, SUBOPT_ORIGIN))
{
values[Anum_pg_subscription_suborigin - 1] =
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index c176bc0d87d..0229a331d5d 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -23,6 +23,7 @@
#include "access/tableam.h"
#include "access/xact.h"
#include "catalog/pg_subscription.h"
+#include "catalog/pg_subscription_d.h"
#include "catalog/pg_subscription_rel.h"
#include "funcapi.h"
#include "lib/dshash.h"
@@ -42,6 +43,7 @@
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/snapmgr.h"
+#include "utils/syscache.h"
/* max sleep time between cycles (3min) */
#define DEFAULT_NAPTIME_PER_CYCLE 180000L
@@ -458,6 +460,7 @@ retry:
worker->leader_pid = is_parallel_apply_worker ? MyProcPid : InvalidPid;
worker->parallel_apply = is_parallel_apply_worker;
worker->oldest_nonremovable_xid = InvalidFullTransactionId;
+ worker->stop_conflict_retention = false;
worker->last_lsn = InvalidXLogRecPtr;
TIMESTAMP_NOBEGIN(worker->last_send_time);
TIMESTAMP_NOBEGIN(worker->last_recv_time);
@@ -1170,7 +1173,8 @@ ApplyLauncherMain(Datum main_arg)
MemoryContext oldctx;
long wait_time = DEFAULT_NAPTIME_PER_CYCLE;
bool can_advance_xmin = true;
- bool retain_conflict_info = false;
+ int nretain_conflict_info = 0;
+ int nstop_retention = 0;
FullTransactionId xmin = InvalidFullTransactionId;
CHECK_FOR_INTERRUPTS();
@@ -1199,7 +1203,7 @@ ApplyLauncherMain(Datum main_arg)
*/
if (sub->retainconflictinfo)
{
- retain_conflict_info = true;
+ nretain_conflict_info++;
can_advance_xmin &= sub->enabled;
/*
@@ -1225,22 +1229,32 @@ ApplyLauncherMain(Datum main_arg)
* the new xmin for advancing the replication slot used in
* conflict detection.
*/
- if (sub->retainconflictinfo && can_advance_xmin)
+ if (sub->retainconflictinfo)
{
FullTransactionId nonremovable_xid;
+ bool stop_conflict_retention;
SpinLockAcquire(&w->relmutex);
nonremovable_xid = w->oldest_nonremovable_xid;
+ stop_conflict_retention = w->stop_conflict_retention;
SpinLockRelease(&w->relmutex);
+ /*
+ * Skip collecting oldest_nonremovable_xid for workers
+ * that have stopped conflict retention.
+ */
+ if (stop_conflict_retention)
+ nstop_retention++;
+
/*
* Stop advancing xmin if an invalid non-removable
* transaction ID is found, otherwise update xmin.
*/
- if (!FullTransactionIdIsValid(nonremovable_xid))
+ else if (!FullTransactionIdIsValid(nonremovable_xid))
can_advance_xmin = false;
- else if (!FullTransactionIdIsValid(xmin) ||
- FullTransactionIdPrecedes(nonremovable_xid, xmin))
+ else if (can_advance_xmin &&
+ (!FullTransactionIdIsValid(xmin) ||
+ FullTransactionIdPrecedes(nonremovable_xid, xmin)))
xmin = nonremovable_xid;
}
@@ -1285,12 +1299,35 @@ ApplyLauncherMain(Datum main_arg)
}
}
+ /*
+ * Do nothing if the replication slot is invalidated due to conflict
+ * retention duration.
+ */
+ if (nretain_conflict_info &&
+ MyReplicationSlot->data.invalidated != RS_INVAL_NONE)
+ {
+ Assert(MyReplicationSlot->data.invalidated == RS_INVAL_CONFLICT_RETENTION_DURATION);
+ }
+
+ /*
+ * Invalidate the conflict slot if all workers with
+ * retain_conflict_info enabled have stopped further conflict
+ * retention.
+ */
+ else if (nstop_retention && nretain_conflict_info == nstop_retention)
+ {
+ ReplicationSlotRelease();
+ InvalidateObsoleteReplicationSlots(RS_INVAL_CONFLICT_RETENTION_DURATION,
+ InvalidXLogRecPtr, InvalidOid,
+ InvalidTransactionId);
+ }
+
/*
* Maintain the xmin value of the replication slot for conflict
* detection if needed, and update the sleep time before the next
* attempt.
*/
- if (retain_conflict_info)
+ else if (nretain_conflict_info)
{
bool updated = false;
@@ -1518,7 +1555,7 @@ GetLeaderApplyWorkerPid(pid_t pid)
Datum
pg_stat_get_subscription(PG_FUNCTION_ARGS)
{
-#define PG_STAT_GET_SUBSCRIPTION_COLS 10
+#define PG_STAT_GET_SUBSCRIPTION_COLS 11
Oid subid = PG_ARGISNULL(0) ? InvalidOid : PG_GETARG_OID(0);
int i;
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
@@ -1595,6 +1632,32 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS)
elog(ERROR, "unknown worker type");
}
+ /*
+ * Only the leader apply worker manages conflict retention (see
+ * maybe_advance_nonremovable_xid() for details).
+ */
+ if (!isParallelApplyWorker(&worker) && !isTablesyncWorker(&worker))
+ {
+ HeapTuple tup;
+ Form_pg_subscription subform;
+
+ tup = SearchSysCache1(SUBSCRIPTIONOID,
+ ObjectIdGetDatum(worker.subid));
+
+ if (!HeapTupleIsValid(tup))
+ elog(ERROR, "cache lookup failed for subscription %u",
+ worker.subid);
+
+ subform = (Form_pg_subscription) GETSTRUCT(tup);
+
+ values[10] = subform->subretainconflictinfo &&
+ !worker.stop_conflict_retention;
+
+ ReleaseSysCache(tup);
+ }
+ else
+ nulls[10] = true;
+
tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
values, nulls);
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 44ff7796b82..22f2b794f62 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -173,6 +173,7 @@
#include "replication/logicalrelation.h"
#include "replication/logicalworker.h"
#include "replication/origin.h"
+#include "replication/slot.h"
#include "replication/walreceiver.h"
#include "replication/worker_internal.h"
#include "rewrite/rewriteHandler.h"
@@ -453,6 +454,7 @@ static void wait_for_publisher_status(RetainConflictInfoData *data,
static void wait_for_local_flush(RetainConflictInfoData *data);
static void adjust_xid_advance_interval(RetainConflictInfoData *data,
bool new_xid_found);
+static void update_conflict_retention_status(void);
static void apply_handle_commit_internal(LogicalRepCommitData *commit_data);
static void apply_handle_insert_internal(ApplyExecutionData *edata,
@@ -3853,7 +3855,8 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
* Ensure to wake up when it's possible to attempt to advance the
* non-removable transaction ID.
*/
- if (data.phase == RCI_GET_CANDIDATE_XID && data.xid_advance_interval)
+ if (!MyLogicalRepWorker->stop_conflict_retention &&
+ data.phase == RCI_GET_CANDIDATE_XID && data.xid_advance_interval)
wait_time = Min(wait_time, data.xid_advance_interval);
rc = WaitLatchOrSocket(MyLatch,
@@ -4110,6 +4113,10 @@ maybe_advance_nonremovable_xid(RetainConflictInfoData *data,
if (!am_leader_apply_worker())
return;
+ /* Exit early if we have already stopped retaining */
+ if (MyLogicalRepWorker->stop_conflict_retention)
+ return;
+
switch (data->phase)
{
case RCI_GET_CANDIDATE_XID:
@@ -4272,6 +4279,8 @@ wait_for_publisher_status(RetainConflictInfoData *data, bool status_received)
static void
wait_for_local_flush(RetainConflictInfoData *data)
{
+ bool stop_conflict_retention = false;
+
Assert(!XLogRecPtrIsInvalid(data->remote_lsn) &&
FullTransactionIdIsValid(data->candidate_xid));
@@ -4333,22 +4342,72 @@ wait_for_local_flush(RetainConflictInfoData *data)
data->flushpos_update_time = data->last_recv_time;
}
- /* Return to wait for the changes to be applied */
+ /* Check if changes up to the remote_lsn have been applied and flushed */
if (last_flushpos < data->remote_lsn)
- return;
+ {
+ TimestampTz now;
- /*
- * Reaching here means the remote WAL position has been received, and all
- * transactions up to that position on the publisher have been applied and
- * flushed locally. So, we can advance the non-removable transaction ID.
- */
- SpinLockAcquire(&MyLogicalRepWorker->relmutex);
- MyLogicalRepWorker->oldest_nonremovable_xid = data->candidate_xid;
- SpinLockRelease(&MyLogicalRepWorker->relmutex);
+ /*
+ * Use last_recv_time when applying changes in the loop to avoid
+ * unnecessary system time retrieval. If last_recv_time is not
+ * available, obtain the current timestamp.
+ */
+ now = data->last_recv_time ? data->last_recv_time : GetCurrentTimestamp();
+
+ /*
+ * If the wait time has not exceeded the maximum limit
+ * (max_conflict_retention_duration), continue waiting for the changes
+ * to be applied. Otherwise, stop tracking the non-removable
+ * transaction ID by this apply worker.
+ */
+ if (!MySubscription->maxconflictretentionduration ||
+ !TimestampDifferenceExceeds(data->candidate_xid_time, now,
+ MySubscription->maxconflictretentionduration))
+ return;
- elog(DEBUG2, "confirmed remote flush up to %X/%X: new oldest_nonremovable_xid %u",
- LSN_FORMAT_ARGS(data->remote_lsn),
- XidFromFullTransactionId(data->candidate_xid));
+ stop_conflict_retention = true;
+ }
+
+ if (!stop_conflict_retention)
+ {
+ /*
+ * Reaching here means the remote WAL position has been received, and
+ * all transactions up to that position on the publisher have been
+ * applied and flushed locally. So, we can advance the non-removable
+ * transaction ID.
+ */
+ SpinLockAcquire(&MyLogicalRepWorker->relmutex);
+ MyLogicalRepWorker->oldest_nonremovable_xid = data->candidate_xid;
+ SpinLockRelease(&MyLogicalRepWorker->relmutex);
+
+ elog(DEBUG2, "confirmed remote flush up to %X/%X: new oldest_nonremovable_xid %u",
+ LSN_FORMAT_ARGS(data->remote_lsn),
+ XidFromFullTransactionId(data->candidate_xid));
+ }
+ else
+ {
+ /*
+ * Reaching here means the time spent applying changes up to the
+ * remote_lsn has exceeded the maximum allowed limit
+ * (max_conflict_retention_duration). So, we will stop retaining
+ * conflict information.
+ *
+ * Currently, the retention will not resume automatically unless user
+ * manually disable retain_conflict_info and re-enable it after
+ * confirming that the replication slot has been dropped.
+ */
+ SpinLockAcquire(&MyLogicalRepWorker->relmutex);
+ MyLogicalRepWorker->oldest_nonremovable_xid = InvalidFullTransactionId;
+ MyLogicalRepWorker->stop_conflict_retention = true;
+ SpinLockRelease(&MyLogicalRepWorker->relmutex);
+
+ ereport(LOG,
+ errmsg("logical replication worker for subscription \"%s\" will stop retaining conflict information",
+ MySubscription->name),
+ errdetail("The time spent applying changes up to LSN %X/%X has exceeded the maximum limit of %u ms.",
+ LSN_FORMAT_ARGS(data->remote_lsn),
+ MySubscription->maxconflictretentionduration));
+ }
/* Notify launcher to update the xmin of the conflict slot */
ApplyLauncherWakeup();
@@ -4412,6 +4471,51 @@ adjust_xid_advance_interval(RetainConflictInfoData *data, bool new_xid_found)
}
}
+/*
+ * Update the conflict retention status for the current apply worker. It checks
+ * whether the worker should stop retaining conflict information due to
+ * invalidation of the replication slot ("pg_conflict_detection").
+ *
+ * Currently, the replication slot is invalidated only if the duration for
+ * retaining conflict information exceeds the allowed maximum.
+ */
+static void
+update_conflict_retention_status(void)
+{
+ ReplicationSlotInvalidationCause cause = RS_INVAL_NONE;
+ ReplicationSlot *slot;
+
+ /* Exit early if retaining conflict information is not required */
+ if (!MySubscription->retainconflictinfo)
+ return;
+
+ /*
+ * Only the leader apply worker manages conflict retention (see
+ * maybe_advance_nonremovable_xid() for details).
+ */
+ if (!am_leader_apply_worker())
+ return;
+
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+
+ slot = SearchNamedReplicationSlot(CONFLICT_DETECTION_SLOT, false);
+
+ if (slot)
+ {
+ SpinLockAcquire(&slot->mutex);
+ cause = slot->data.invalidated;
+ SpinLockRelease(&slot->mutex);
+
+ Assert(cause == RS_INVAL_NONE || cause == RS_INVAL_CONFLICT_RETENTION_DURATION);
+ }
+
+ LWLockRelease(ReplicationSlotControlLock);
+
+ SpinLockAcquire(&MyLogicalRepWorker->relmutex);
+ MyLogicalRepWorker->stop_conflict_retention = cause != RS_INVAL_NONE;
+ SpinLockRelease(&MyLogicalRepWorker->relmutex);
+}
+
/*
* Exit routine for apply workers due to subscription parameter changes.
*/
@@ -4583,6 +4687,16 @@ maybe_reread_subscription(void)
CommitTransactionCommand();
MySubscriptionValid = true;
+
+ /*
+ * Update worker status to avoid unnecessary conflict retention if the
+ * replication slot ("pg_conflict_detection") was invalidated prior to
+ * enabling the retain_conflict_info option. This is also necessary to
+ * restart conflict retention if the user has disabled and subsequently
+ * re-enabled the retain_conflict_info option, resulting in the
+ * replication slot being recreated.
+ */
+ update_conflict_retention_status();
}
/*
@@ -5221,6 +5335,8 @@ InitializeLogRepWorker(void)
MySubscription->name)));
CommitTransactionCommand();
+
+ update_conflict_retention_status();
}
/*
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 38c3b3f9c3c..9f6d520ea7a 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -108,10 +108,11 @@ const char *const SlotInvalidationCauses[] = {
[RS_INVAL_WAL_REMOVED] = "wal_removed",
[RS_INVAL_HORIZON] = "rows_removed",
[RS_INVAL_WAL_LEVEL] = "wal_level_insufficient",
+ [RS_INVAL_CONFLICT_RETENTION_DURATION] = "conflict_retention_exceeds_max_duration",
};
/* Maximum number of invalidation causes */
-#define RS_INVAL_MAX_CAUSES RS_INVAL_WAL_LEVEL
+#define RS_INVAL_MAX_CAUSES RS_INVAL_CONFLICT_RETENTION_DURATION
StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1),
"array length mismatch");
@@ -1572,6 +1573,11 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause,
case RS_INVAL_WAL_LEVEL:
appendStringInfoString(&err_detail, _("Logical decoding on standby requires \"wal_level\" >= \"logical\" on the primary server."));
break;
+
+ case RS_INVAL_CONFLICT_RETENTION_DURATION:
+ appendStringInfo(&err_detail, _("The duration for retaining conflict information exceeds the maximum limit defined by the subscription's max_conflict_retention_duration setting"));
+ break;
+
case RS_INVAL_NONE:
pg_unreachable();
}
@@ -1685,6 +1691,10 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
if (SlotIsLogical(s))
invalidation_cause = cause;
break;
+ case RS_INVAL_CONFLICT_RETENTION_DURATION:
+ if (IsReservedSlotName(NameStr(s->data.name)))
+ invalidation_cause = cause;
+ break;
case RS_INVAL_NONE:
pg_unreachable();
}
@@ -1838,6 +1848,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
* - RS_INVAL_HORIZON: requires a snapshot <= the given horizon in the given
* db; dboid may be InvalidOid for shared relations
* - RS_INVAL_WAL_LEVEL: is logical
+ * - RS_INVAL_CONFLICT_RETENTION_DURATION: is "pg_conflict_detection"
*
* NB - this runs as part of checkpoint, so avoid raising errors if possible.
*/
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 52b94b6c018..9d1cfe35c65 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4869,6 +4869,7 @@ getSubscriptions(Archive *fout)
int i_subenabled;
int i_subfailover;
int i_subretainconflictinfo;
+ int i_submaxconflictretentionduration;
int i,
ntups;
@@ -4948,10 +4949,18 @@ getSubscriptions(Archive *fout)
if (fout->remoteVersion >= 180000)
appendPQExpBufferStr(query,
- " s.subretainconflictinfo\n");
+ " s.subretainconflictinfo,\n");
else
appendPQExpBuffer(query,
- " false AS subretainconflictinfo\n");
+ " false AS subretainconflictinfo,\n");
+
+ if (fout->remoteVersion >= 180000)
+ appendPQExpBufferStr(query,
+ " s.submaxconflictretentionduration\n");
+ else
+ appendPQExpBuffer(query,
+ " 0 AS submaxconflictretentionduration\n");
+
appendPQExpBufferStr(query,
"FROM pg_subscription s\n");
@@ -4985,6 +4994,7 @@ getSubscriptions(Archive *fout)
i_subrunasowner = PQfnumber(res, "subrunasowner");
i_subfailover = PQfnumber(res, "subfailover");
i_subretainconflictinfo = PQfnumber(res, "subretainconflictinfo");
+ i_submaxconflictretentionduration = PQfnumber(res, "submaxconflictretentionduration");
i_subconninfo = PQfnumber(res, "subconninfo");
i_subslotname = PQfnumber(res, "subslotname");
i_subsynccommit = PQfnumber(res, "subsynccommit");
@@ -5020,6 +5030,8 @@ getSubscriptions(Archive *fout)
(strcmp(PQgetvalue(res, i, i_subfailover), "t") == 0);
subinfo[i].subretainconflictinfo =
(strcmp(PQgetvalue(res, i, i_subretainconflictinfo), "t") == 0);
+ subinfo[i].submaxconflictretentionduration =
+ atoi(PQgetvalue(res, i, i_submaxconflictretentionduration));
subinfo[i].subconninfo =
pg_strdup(PQgetvalue(res, i, i_subconninfo));
if (PQgetisnull(res, i, i_subslotname))
@@ -5281,6 +5293,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
if (subinfo->subretainconflictinfo)
appendPQExpBufferStr(query, ", retain_conflict_info = true");
+ if (subinfo->submaxconflictretentionduration)
+ appendPQExpBuffer(query, ", max_conflict_retention_duration = %d", subinfo->submaxconflictretentionduration);
+
if (strcmp(subinfo->subsynccommit, "off") != 0)
appendPQExpBuffer(query, ", synchronous_commit = %s", fmtId(subinfo->subsynccommit));
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 117d5a81f1e..56d9cc8747c 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -682,6 +682,7 @@ typedef struct _SubscriptionInfo
bool subrunasowner;
bool subfailover;
bool subretainconflictinfo;
+ int submaxconflictretentionduration;
char *subconninfo;
char *subslotname;
char *subsynccommit;
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 2cde9e76ca1..7bbc2bf6b92 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6688,7 +6688,7 @@ describeSubscriptions(const char *pattern, bool verbose)
printQueryOpt myopt = pset.popt;
static const bool translate_columns[] = {false, false, false, false,
false, false, false, false, false, false, false, false, false, false,
- false, false};
+ false, false, false};
if (pset.sversion < 100000)
{
@@ -6757,10 +6757,16 @@ describeSubscriptions(const char *pattern, bool verbose)
", subfailover AS \"%s\"\n",
gettext_noop("Failover"));
if (pset.sversion >= 180000)
+ {
appendPQExpBuffer(&buf,
", subretainconflictinfo AS \"%s\"\n",
gettext_noop("Retain conflict info"));
+ appendPQExpBuffer(&buf,
+ ", submaxconflictretentionduration AS \"%s\"\n",
+ gettext_noop("Max conflict retention duration"));
+ }
+
appendPQExpBuffer(&buf,
", subsynccommit AS \"%s\"\n"
", subconninfo AS \"%s\"\n",
diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c
index e0be3c6fd63..d899428e5d5 100644
--- a/src/bin/psql/tab-complete.in.c
+++ b/src/bin/psql/tab-complete.in.c
@@ -2278,10 +2278,10 @@ match_previous_words(int pattern_id,
COMPLETE_WITH("(", "PUBLICATION");
/* ALTER SUBSCRIPTION <name> SET ( */
else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "SET", "("))
- COMPLETE_WITH("binary", "disable_on_error", "failover", "origin",
- "password_required", "retain_conflict_info",
- "run_as_owner", "slot_name", "streaming",
- "synchronous_commit", "two_phase");
+ COMPLETE_WITH("binary", "disable_on_error", "failover",
+ "max_conflict_retention_duration", "origin", "password_required",
+ "retain_conflict_info", "run_as_owner", "slot_name",
+ "streaming", "synchronous_commit", "two_phase");
/* ALTER SUBSCRIPTION <name> SKIP ( */
else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "SKIP", "("))
COMPLETE_WITH("lsn");
@@ -3689,10 +3689,10 @@ match_previous_words(int pattern_id,
/* Complete "CREATE SUBSCRIPTION <name> ... WITH ( <opt>" */
else if (Matches("CREATE", "SUBSCRIPTION", MatchAnyN, "WITH", "("))
COMPLETE_WITH("binary", "connect", "copy_data", "create_slot",
- "disable_on_error", "enabled", "failover", "origin",
- "password_required", "retain_conflict_info",
- "run_as_owner", "slot_name", "streaming",
- "synchronous_commit", "two_phase");
+ "disable_on_error", "enabled", "failover",
+ "max_conflict_retention_duration", "origin", "password_required",
+ "retain_conflict_info", "run_as_owner", "slot_name",
+ "streaming", "synchronous_commit", "two_phase");
/* CREATE TRIGGER --- is allowed inside CREATE SCHEMA, so use TailMatches */
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 5b8c2ad2a54..d76799866c1 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5627,9 +5627,9 @@
proname => 'pg_stat_get_subscription', prorows => '10', proisstrict => 'f',
proretset => 't', provolatile => 's', proparallel => 'r',
prorettype => 'record', proargtypes => 'oid',
- proallargtypes => '{oid,oid,oid,int4,int4,pg_lsn,timestamptz,timestamptz,pg_lsn,timestamptz,text}',
- proargmodes => '{i,o,o,o,o,o,o,o,o,o,o}',
- proargnames => '{subid,subid,relid,pid,leader_pid,received_lsn,last_msg_send_time,last_msg_receipt_time,latest_end_lsn,latest_end_time,worker_type}',
+ proallargtypes => '{oid,oid,oid,int4,int4,pg_lsn,timestamptz,timestamptz,pg_lsn,timestamptz,text,bool}',
+ proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o}',
+ proargnames => '{subid,subid,relid,pid,leader_pid,received_lsn,last_msg_send_time,last_msg_receipt_time,latest_end_lsn,latest_end_time,worker_type,retain_conflict_info}',
prosrc => 'pg_stat_get_subscription' },
{ oid => '2026', descr => 'statistics: current backend PID',
proname => 'pg_backend_pid', provolatile => 's', proparallel => 'r',
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 0ac7c0b120c..f5677b53fad 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -81,6 +81,12 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
bool subretainconflictinfo; /* True if information useful for
* conflict detection is retained */
+ int32 submaxconflictretentionduration; /* The maximum duration
+ * (in milliseconds) for
+ * which conflict
+ * information can be
+ * retained */
+
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* Connection string to the publisher */
text subconninfo BKI_FORCE_NOT_NULL;
@@ -136,6 +142,10 @@ typedef struct Subscription
* to be synchronized to the standbys. */
bool retainconflictinfo; /* True if information useful for conflict
* detection is retained */
+ int32 maxconflictretentionduration; /* The maximum duration (in
+ * milliseconds) for which
+ * conflict information can be
+ * retained */
char *conninfo; /* Connection string to the publisher */
char *slotname; /* Name of the replication slot */
char *synccommit; /* Synchronous commit setting for worker */
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 34683299620..86a3e6a279b 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -63,6 +63,8 @@ typedef enum ReplicationSlotInvalidationCause
RS_INVAL_HORIZON,
/* wal_level insufficient for slot */
RS_INVAL_WAL_LEVEL,
+ /* duration of conflict info retention exceeds the maximum limit */
+ RS_INVAL_CONFLICT_RETENTION_DURATION,
} ReplicationSlotInvalidationCause;
extern PGDLLIMPORT const char *const SlotInvalidationCauses[];
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index b09486017f4..40469f2df28 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -103,6 +103,12 @@ typedef struct LogicalRepWorker
*/
FullTransactionId oldest_nonremovable_xid;
+ /*
+ * Indicates whether the apply worker has stopped retaining conflict
+ * information. This is used only when retain_conflict_info is enabled.
+ */
+ bool stop_conflict_retention;
+
/* Stats. */
XLogRecPtr last_lsn;
TimestampTz last_send_time;
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 3361f6a69c9..ed2bb26b3eb 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2142,9 +2142,10 @@ pg_stat_subscription| SELECT su.oid AS subid,
st.last_msg_send_time,
st.last_msg_receipt_time,
st.latest_end_lsn,
- st.latest_end_time
+ st.latest_end_time,
+ st.retain_conflict_info
FROM (pg_subscription su
- LEFT JOIN pg_stat_get_subscription(NULL::oid) st(subid, relid, pid, leader_pid, received_lsn, last_msg_send_time, last_msg_receipt_time, latest_end_lsn, latest_end_time, worker_type) ON ((st.subid = su.oid)));
+ LEFT JOIN pg_stat_get_subscription(NULL::oid) st(subid, relid, pid, leader_pid, received_lsn, last_msg_send_time, last_msg_receipt_time, latest_end_lsn, latest_end_time, worker_type, retain_conflict_info) ON ((st.subid = su.oid)));
pg_stat_subscription_stats| SELECT ss.subid,
s.subname,
ss.apply_error_count,
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index bff4cc051db..577493fe17b 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -116,18 +116,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub3;
@@ -145,10 +145,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -157,10 +157,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname');
ALTER SUBSCRIPTION regress_testsub SET (password_required = false);
ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | f | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | f | 0 | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
@@ -176,10 +176,10 @@ ERROR: unrecognized subscription parameter: "create_slot"
-- ok
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist2 | 0/12345
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist2 | 0/12345
(1 row)
-- ok - with lsn = NONE
@@ -188,10 +188,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
ERROR: invalid WAL location (LSN): 0/0
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
BEGIN;
@@ -223,10 +223,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
ERROR: invalid value for parameter "synchronous_commit": "foobar"
HINT: Available values: local, remote_write, remote_apply, on, off.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+------------------------------+----------
- regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | f | local | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+------------------------------+----------
+ regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | f | 0 | local | dbname=regress_doesnotexist2 | 0/0
(1 row)
-- rename back to keep the rest simple
@@ -255,19 +255,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (binary = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -279,27 +279,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication already exists
@@ -314,10 +314,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
ERROR: publication "testpub1" is already in subscription "regress_testsub"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication used more than once
@@ -332,10 +332,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub"
-- ok - delete publications
ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -371,19 +371,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- we can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -393,10 +393,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -409,18 +409,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(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 | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -434,19 +434,44 @@ WARNING: information for detecting conflicts cannot be fully retained when "tra
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | t | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | t | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- ok
ALTER SUBSCRIPTION regress_testsub SET (retain_conflict_info = false);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+-- fail - max_conflict_retention_duration must be integer
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, max_conflict_retention_duration = foo);
+ERROR: max_conflict_retention_duration requires an integer value
+-- ok
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, max_conflict_retention_duration = 1000);
+WARNING: subscription was created, but is not connected
+HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 1000 | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+-- ok
+ALTER SUBSCRIPTION regress_testsub SET (max_conflict_retention_duration = 0);
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Retain conflict info | Max conflict retention duration | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+----------------------+---------------------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index c65397e5ac6..17dd6aa26a9 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -303,6 +303,22 @@ ALTER SUBSCRIPTION regress_testsub SET (retain_conflict_info = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
+-- fail - max_conflict_retention_duration must be integer
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, max_conflict_retention_duration = foo);
+
+-- ok
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, max_conflict_retention_duration = 1000);
+
+\dRs+
+
+-- ok
+ALTER SUBSCRIPTION regress_testsub SET (max_conflict_retention_duration = 0);
+
+\dRs+
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+
-- let's do some tests with pg_create_subscription rather than superuser
SET SESSION AUTHORIZATION regress_subscription_user3;
--
2.30.0.windows.2