public inbox for [email protected]
help / color / mirror / Atom feedFrom: Zhijie Hou (Fujitsu) <[email protected]>
To: pgsql-hackers <[email protected]>
Cc: Nisha Moond <[email protected]>
Cc: Dilip Kumar <[email protected]>
Cc: shveta malik <[email protected]>
Cc: Jan Wieck <[email protected]>
Cc: Tomas Vondra <[email protected]>
Cc: Amit Kapila <[email protected]>
Cc: Ashutosh Bapat <[email protected]>
Subject: Conflict detection and logging in logical replication
Date: Fri, 21 Jun 2024 07:47:20 +0000
Message-ID: <OS0PR01MB5716352552DFADB8E9AD1D8994C92@OS0PR01MB5716.jpnprd01.prod.outlook.com> (raw)
Hi hackers,
Cc people involved in the original thread[1].
I am starting a new thread to share and discuss the implementation of
conflict detection and logging in logical replication, as well as the
collection of statistics related to these conflicts.
In the original conflict resolution thread[1], we have decided to
split this work into multiple patches to facilitate incremental progress
towards supporting conflict resolution in logical replication. This phased
approach will allow us to address simpler tasks first. The overall work
plan involves: 1. conflict detection (detect and log conflicts like
'insert_exists', 'update_differ', 'update_missing', and 'delete_missing')
2. implement simple built-in resolution strategies like
'apply(remote_apply)' and 'skip(keep_local)'. 3. monitor capability for
conflicts and resolutions in statistics or history table.
Following the feedback received from PGconf.dev and discussions in the
conflict resolution thread, features 1 and 3 are important independently.
So, we start a separate thread for them.
Here are the basic designs for the detection and statistics:
- The detail of the conflict detection
We add a new parameter detect_conflict for CREATE and ALTER subscription
commands. This new parameter will decide if subscription will go for
confict detection. By default, conflict detection will be off for a
subscription.
When conflict detection is enabled, additional logging is triggered in the
following conflict scenarios:
insert_exists: Inserting a row that violates a NOT DEFERRABLE unique constraint.
update_differ: updating a row that was previously modified by another origin.
update_missing: The tuple to be updated is missing.
delete_missing: The tuple to be deleted is missing.
For insert_exists conflict, the log can include origin and commit
timestamp details of the conflicting key with track_commit_timestamp
enabled. And update_differ conflict can only be detected when
track_commit_timestamp is enabled.
Regarding insert_exists conflicts, the current design is to pass
noDupErr=true in ExecInsertIndexTuples() to prevent immediate error
handling on duplicate key violation. After calling
ExecInsertIndexTuples(), if there was any potential conflict in the
unique indexes, we report an ERROR for the insert_exists conflict along
with additional information (origin, committs, key value) for the
conflicting row. Another way for this is to conduct a pre-check for
duplicate key violation before applying the INSERT operation, but this
could introduce overhead for each INSERT even in the absence of conflicts.
We welcome any alternative viewpoints on this matter.
- The detail of statistics collection
We add columns(insert_exists_count, update_differ_count,
update_missing_count, delete_missing_count) in view
pg_stat_subscription_workers to shows information about the conflict which
occur during the application of logical replication changes.
The conflicts will be tracked when track_conflict option of the
subscription is enabled. Additionally, update_differ can be detected only
when track_commit_timestamp is enabled.
The patches for above features are attached.
Suggestions and comments are highly appreciated.
[1] https://www.postgresql.org/message-id/CAA4eK1LgPyzPr_Vrvvr4syrde4hyT%3DQQnGjdRUNP-tz3eYa%3DGQ%40mail...
Best Regards,
Hou Zhijie
Attachments:
[application/octet-stream] v1-0002-Collect-statistics-about-conflicts-in-logical-rep.patch (19.2K, ../OS0PR01MB5716352552DFADB8E9AD1D8994C92@OS0PR01MB5716.jpnprd01.prod.outlook.com/2-v1-0002-Collect-statistics-about-conflicts-in-logical-rep.patch)
download | inline diff:
From 239842a83728dd75c2e91f37bc6596f26c4c72da Mon Sep 17 00:00:00 2001
From: Hou Zhijie <[email protected]>
Date: Fri, 21 Jun 2024 10:41:49 +0800
Subject: [PATCH v1 2/2] Collect statistics about conflicts in logical
replication
This commit adds columns in view pg_stat_subscription_workers to shows
information about the conflict which occur during the application of
logical replication changes. Currently, the following columns are added.
insert_exists_counts:
Number of times inserting a row that iolates a NOT DEFERRABLE unique constraint.
update_differ_counts:
Number of times updating a row that was previously modified by another origin.
update_missing_counts:
Number of times that the tuple to be updated is missing.
delete_missing_counts:
Number of times that the tuple to be deleted is missing.
The conflicts will be tracked only when track_conflict option of the
subscription is enabled. Additionally, update_differ can be detected only
when track_commit_timestamp is enabled.
---
doc/src/sgml/monitoring.sgml | 52 ++++++++++++-
doc/src/sgml/ref/create_subscription.sgml | 4 +-
src/backend/catalog/system_views.sql | 4 +
src/backend/replication/logical/conflict.c | 4 +
.../utils/activity/pgstat_subscription.c | 17 ++++
src/backend/utils/adt/pgstatfuncs.c | 20 ++++-
src/include/catalog/pg_proc.dat | 6 +-
src/include/pgstat.h | 4 +
src/include/replication/conflict.h | 2 +
src/test/regress/expected/rules.out | 6 +-
src/test/subscription/t/026_stats.pl | 77 +++++++++++++++++--
11 files changed, 178 insertions(+), 18 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index b2ad9b446f..0ceb71f214 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -507,7 +507,7 @@ postgres 27093 0.0 0.0 30096 2752 ? Ss 11:34 0:00 postgres: ser
<row>
<entry><structname>pg_stat_subscription_stats</structname><indexterm><primary>pg_stat_subscription_stats</primary></indexterm></entry>
- <entry>One row per subscription, showing statistics about errors.
+ <entry>One row per subscription, showing statistics about errors and conflicts.
See <link linkend="monitoring-pg-stat-subscription-stats">
<structname>pg_stat_subscription_stats</structname></link> for details.
</entry>
@@ -2163,6 +2163,56 @@ description | Waiting for a newly initialized WAL file to reach durable storage
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>insert_exists_count</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of times inserting a row that violates a
+ <literal>NOT DEFERRABLE</literal> unique constraint while applying
+ changes. This conflict is counted only if
+ <link linkend="sql-createsubscription-params-with-detect-conflict"><literal>detect_conflict</literal></link>
+ is enabled
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>update_differ_count</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of times updating a row that was previously modified by another
+ source while applying changes. This conflict is counted only when the
+ <link linkend="sql-createsubscription-params-with-detect-conflict"><literal>detect_conflict</literal></link>
+ option of the subscription and <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ are enabled
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>update_missing_count</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of times that the tuple to be updated is not found while applying
+ changes. This conflict is counted only if
+ <link linkend="sql-createsubscription-params-with-detect-conflict"><literal>detect_conflict</literal></link>
+ is enabled
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>delete_missing_count</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of times that the tuple to be deleted is not found while applying
+ changes. This conflict is counted only if
+ <link linkend="sql-createsubscription-params-with-detect-conflict"><literal>detect_conflict</literal></link>
+ is enabled
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>stats_reset</structfield> <type>timestamp with time zone</type>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index ce37fa6490..06bea458a6 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -437,8 +437,8 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
The default is <literal>false</literal>.
</para>
<para>
- When conflict detection is enabled, additional logging is triggered
- in the following scenarios:
+ When conflict detection is enabled, additional logging is triggered and
+ the conflict statistics are collected in the following scenarios:
<variablelist>
<varlistentry>
<term><literal>insert_exists</literal></term>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 30393e6d67..182836ac82 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1370,6 +1370,10 @@ CREATE VIEW pg_stat_subscription_stats AS
s.subname,
ss.apply_error_count,
ss.sync_error_count,
+ ss.insert_exists_count,
+ ss.update_differ_count,
+ ss.update_missing_count,
+ ss.delete_missing_count,
ss.stats_reset
FROM pg_subscription as s,
pg_stat_get_subscription_stats(s.oid) as ss;
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index c8d2446033..9167ebee5b 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -15,8 +15,10 @@
#include "postgres.h"
#include "access/commit_ts.h"
+#include "pgstat.h"
#include "replication/conflict.h"
#include "replication/origin.h"
+#include "replication/worker_internal.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
@@ -75,6 +77,8 @@ ReportApplyConflict(int elevel, ConflictType type, Relation localrel,
RepOriginId localorigin, TimestampTz localts,
TupleTableSlot *conflictslot)
{
+ pgstat_report_subscription_conflict(MySubscription->oid, type);
+
ereport(elevel,
errcode(ERRCODE_INTEGRITY_CONSTRAINT_VIOLATION),
errmsg("conflict %s detected on relation \"%s.%s\"",
diff --git a/src/backend/utils/activity/pgstat_subscription.c b/src/backend/utils/activity/pgstat_subscription.c
index d9af8de658..e06c92727e 100644
--- a/src/backend/utils/activity/pgstat_subscription.c
+++ b/src/backend/utils/activity/pgstat_subscription.c
@@ -39,6 +39,21 @@ pgstat_report_subscription_error(Oid subid, bool is_apply_error)
pending->sync_error_count++;
}
+/*
+ * Report a subscription conflict.
+ */
+void
+pgstat_report_subscription_conflict(Oid subid, ConflictType type)
+{
+ PgStat_EntryRef *entry_ref;
+ PgStat_BackendSubEntry *pending;
+
+ entry_ref = pgstat_prep_pending_entry(PGSTAT_KIND_SUBSCRIPTION,
+ InvalidOid, subid, NULL);
+ pending = entry_ref->pending;
+ pending->conflict_count[type]++;
+}
+
/*
* Report creating the subscription.
*/
@@ -101,6 +116,8 @@ pgstat_subscription_flush_cb(PgStat_EntryRef *entry_ref, bool nowait)
#define SUB_ACC(fld) shsubent->stats.fld += localent->fld
SUB_ACC(apply_error_count);
SUB_ACC(sync_error_count);
+ for (int i = 0; i < CONFLICT_NUM_TYPES; i++)
+ SUB_ACC(conflict_count[i]);
#undef SUB_ACC
pgstat_unlock_entry(entry_ref);
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index 3876339ee1..4cdf4b2aff 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -1966,7 +1966,7 @@ pg_stat_get_replication_slot(PG_FUNCTION_ARGS)
Datum
pg_stat_get_subscription_stats(PG_FUNCTION_ARGS)
{
-#define PG_STAT_GET_SUBSCRIPTION_STATS_COLS 4
+#define PG_STAT_GET_SUBSCRIPTION_STATS_COLS 8
Oid subid = PG_GETARG_OID(0);
TupleDesc tupdesc;
Datum values[PG_STAT_GET_SUBSCRIPTION_STATS_COLS] = {0};
@@ -1985,7 +1985,15 @@ pg_stat_get_subscription_stats(PG_FUNCTION_ARGS)
INT8OID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber) 3, "sync_error_count",
INT8OID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 4, "stats_reset",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 4, "insert_exists_count",
+ INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, (AttrNumber) 5, "update_differ_count",
+ INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, (AttrNumber) 6, "update_missing_count",
+ INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, (AttrNumber) 7, "delete_missing_count",
+ INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, (AttrNumber) 8, "stats_reset",
TIMESTAMPTZOID, -1, 0);
BlessTupleDesc(tupdesc);
@@ -2005,11 +2013,15 @@ pg_stat_get_subscription_stats(PG_FUNCTION_ARGS)
/* sync_error_count */
values[2] = Int64GetDatum(subentry->sync_error_count);
+ /* conflict count */
+ for (int i = 0 ; i < CONFLICT_NUM_TYPES; i++)
+ values[3 + i] = Int64GetDatum(subentry->conflict_count[i]);
+
/* stats_reset */
if (subentry->stat_reset_timestamp == 0)
- nulls[3] = true;
+ nulls[7] = true;
else
- values[3] = TimestampTzGetDatum(subentry->stat_reset_timestamp);
+ values[7] = TimestampTzGetDatum(subentry->stat_reset_timestamp);
/* Returns the record as Datum */
PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 6a5476d3c4..08bc966a2f 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5505,9 +5505,9 @@
{ oid => '6231', descr => 'statistics: information about subscription stats',
proname => 'pg_stat_get_subscription_stats', provolatile => 's',
proparallel => 'r', prorettype => 'record', proargtypes => 'oid',
- proallargtypes => '{oid,oid,int8,int8,timestamptz}',
- proargmodes => '{i,o,o,o,o}',
- proargnames => '{subid,subid,apply_error_count,sync_error_count,stats_reset}',
+ proallargtypes => '{oid,oid,int8,int8,int8,int8,int8,int8,timestamptz}',
+ proargmodes => '{i,o,o,o,o,o,o,o,o}',
+ proargnames => '{subid,subid,apply_error_count,sync_error_count,insert_exists_count,update_differ_count,update_missing_count,delete_missing_count,stats_reset}',
prosrc => 'pg_stat_get_subscription_stats' },
{ oid => '6118', descr => 'statistics: information about subscription',
proname => 'pg_stat_get_subscription', prorows => '10', proisstrict => 'f',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 2136239710..b957e7ad36 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -14,6 +14,7 @@
#include "datatype/timestamp.h"
#include "portability/instr_time.h"
#include "postmaster/pgarch.h" /* for MAX_XFN_CHARS */
+#include "replication/conflict.h"
#include "utils/backend_progress.h" /* for backward compatibility */
#include "utils/backend_status.h" /* for backward compatibility */
#include "utils/relcache.h"
@@ -135,6 +136,7 @@ typedef struct PgStat_BackendSubEntry
{
PgStat_Counter apply_error_count;
PgStat_Counter sync_error_count;
+ PgStat_Counter conflict_count[CONFLICT_NUM_TYPES];
} PgStat_BackendSubEntry;
/* ----------
@@ -393,6 +395,7 @@ typedef struct PgStat_StatSubEntry
{
PgStat_Counter apply_error_count;
PgStat_Counter sync_error_count;
+ PgStat_Counter conflict_count[CONFLICT_NUM_TYPES];
TimestampTz stat_reset_timestamp;
} PgStat_StatSubEntry;
@@ -695,6 +698,7 @@ extern PgStat_SLRUStats *pgstat_fetch_slru(void);
*/
extern void pgstat_report_subscription_error(Oid subid, bool is_apply_error);
+extern void pgstat_report_subscription_conflict(Oid subid, ConflictType conflict);
extern void pgstat_create_subscription(Oid subid);
extern void pgstat_drop_subscription(Oid subid);
extern PgStat_StatSubEntry *pgstat_fetch_stat_subscription(Oid subid);
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index 4a6743b24b..7b9a3d755c 100644
--- a/src/include/replication/conflict.h
+++ b/src/include/replication/conflict.h
@@ -33,6 +33,8 @@ typedef enum
CT_DELETE_MISSING,
} ConflictType;
+#define CONFLICT_NUM_TYPES (CT_DELETE_MISSING + 1)
+
extern bool GetTupleCommitTs(TupleTableSlot *localslot, TransactionId *xmin,
RepOriginId *localorigin, TimestampTz *localts);
extern void ReportApplyConflict(int elevel, ConflictType type,
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 13178e2b3d..80a6857b00 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2141,9 +2141,13 @@ pg_stat_subscription_stats| SELECT ss.subid,
s.subname,
ss.apply_error_count,
ss.sync_error_count,
+ ss.insert_exists_count,
+ ss.update_differ_count,
+ ss.update_missing_count,
+ ss.delete_missing_count,
ss.stats_reset
FROM pg_subscription s,
- LATERAL pg_stat_get_subscription_stats(s.oid) ss(subid, apply_error_count, sync_error_count, stats_reset);
+ LATERAL pg_stat_get_subscription_stats(s.oid) ss(subid, apply_error_count, sync_error_count, insert_exists_count, update_differ_count, update_missing_count, delete_missing_count, stats_reset);
pg_stat_sys_indexes| SELECT relid,
indexrelid,
schemaname,
diff --git a/src/test/subscription/t/026_stats.pl b/src/test/subscription/t/026_stats.pl
index fb3e5629b3..9c00f2a243 100644
--- a/src/test/subscription/t/026_stats.pl
+++ b/src/test/subscription/t/026_stats.pl
@@ -16,6 +16,7 @@ $node_publisher->start;
# Create subscriber node.
my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf', 'track_commit_timestamp = on');
$node_subscriber->start;
@@ -30,6 +31,7 @@ sub create_sub_pub_w_errors
qq[
BEGIN;
CREATE TABLE $table_name(a int);
+ ALTER TABLE $table_name REPLICA IDENTITY FULL;
INSERT INTO $table_name VALUES (1);
COMMIT;
]);
@@ -53,7 +55,7 @@ sub create_sub_pub_w_errors
# infinite error loop due to violating the unique constraint.
my $sub_name = $table_name . '_sub';
$node_subscriber->safe_psql($db,
- qq(CREATE SUBSCRIPTION $sub_name CONNECTION '$publisher_connstr' PUBLICATION $pub_name)
+ qq(CREATE SUBSCRIPTION $sub_name CONNECTION '$publisher_connstr' PUBLICATION $pub_name WITH (detect_conflict = on))
);
$node_publisher->wait_for_catchup($sub_name);
@@ -95,7 +97,7 @@ sub create_sub_pub_w_errors
$node_subscriber->poll_query_until(
$db,
qq[
- SELECT apply_error_count > 0
+ SELECT apply_error_count > 0 AND insert_exists_count > 0
FROM pg_stat_subscription_stats
WHERE subname = '$sub_name'
])
@@ -105,6 +107,47 @@ sub create_sub_pub_w_errors
# Truncate test table so that apply worker can continue.
$node_subscriber->safe_psql($db, qq(TRUNCATE $table_name));
+ # Update and delete data to test table on the publisher, the update
+ # and delete will be skipped an error on the subscriber as there is no
+ # data in the test table.
+ $node_publisher->safe_psql($db, qq(
+ UPDATE $table_name SET a = 2;
+ DELETE FROM $table_name;
+ ));
+
+ # Wait for the tuple miss to be reported.
+ $node_subscriber->poll_query_until(
+ $db,
+ qq[
+ SELECT update_missing_count > 0 AND delete_missing_count > 0
+ FROM pg_stat_subscription_stats
+ WHERE subname = '$sub_name'
+ ])
+ or die
+ qq(Timed out while waiting for tuple missing conflict for subscription '$sub_name');
+
+ # Prepare data for further tests.
+ $node_publisher->safe_psql($db, qq(INSERT INTO $table_name VALUES (1)));
+ $node_publisher->wait_for_catchup($sub_name);
+ $node_subscriber->safe_psql($db, qq(
+ TRUNCATE $table_name;
+ INSERT INTO $table_name VALUES (1);
+ ));
+
+ # Update and delete data to test table on the publisher, the update
+ # should update a row on the subscriber that was modified by a different origin.
+ $node_publisher->safe_psql($db, qq(UPDATE $table_name SET a = 2;));
+
+ $node_subscriber->poll_query_until(
+ $db,
+ qq[
+ SELECT update_differ_count > 0
+ FROM pg_stat_subscription_stats
+ WHERE subname = '$sub_name'
+ ])
+ or die
+ qq(Timed out while waiting for update_differ conflict for subscription '$sub_name');
+
return ($pub_name, $sub_name);
}
@@ -128,11 +171,15 @@ is( $node_subscriber->safe_psql(
$db,
qq(SELECT apply_error_count > 0,
sync_error_count > 0,
+ insert_exists_count > 0,
+ update_differ_count > 0,
+ update_missing_count > 0,
+ delete_missing_count > 0,
stats_reset IS NULL
FROM pg_stat_subscription_stats
WHERE subname = '$sub1_name')
),
- qq(t|t|t),
+ qq(t|t|t|t|t|t|t),
qq(Check that apply errors and sync errors are both > 0 and stats_reset is NULL for subscription '$sub1_name'.)
);
@@ -146,11 +193,15 @@ is( $node_subscriber->safe_psql(
$db,
qq(SELECT apply_error_count = 0,
sync_error_count = 0,
+ insert_exists_count = 0,
+ update_differ_count = 0,
+ update_missing_count = 0,
+ delete_missing_count = 0,
stats_reset IS NOT NULL
FROM pg_stat_subscription_stats
WHERE subname = '$sub1_name')
),
- qq(t|t|t),
+ qq(t|t|t|t|t|t|t),
qq(Confirm that apply errors and sync errors are both 0 and stats_reset is not NULL after reset for subscription '$sub1_name'.)
);
@@ -186,11 +237,15 @@ is( $node_subscriber->safe_psql(
$db,
qq(SELECT apply_error_count > 0,
sync_error_count > 0,
+ insert_exists_count > 0,
+ update_differ_count > 0,
+ update_missing_count > 0,
+ delete_missing_count > 0,
stats_reset IS NULL
FROM pg_stat_subscription_stats
WHERE subname = '$sub2_name')
),
- qq(t|t|t),
+ qq(t|t|t|t|t|t|t),
qq(Confirm that apply errors and sync errors are both > 0 and stats_reset is NULL for sub '$sub2_name'.)
);
@@ -203,11 +258,15 @@ is( $node_subscriber->safe_psql(
$db,
qq(SELECT apply_error_count = 0,
sync_error_count = 0,
+ insert_exists_count = 0,
+ update_differ_count = 0,
+ update_missing_count = 0,
+ delete_missing_count = 0,
stats_reset IS NOT NULL
FROM pg_stat_subscription_stats
WHERE subname = '$sub1_name')
),
- qq(t|t|t),
+ qq(t|t|t|t|t|t|t),
qq(Confirm that apply errors and sync errors are both 0 and stats_reset is not NULL for sub '$sub1_name' after reset.)
);
@@ -215,11 +274,15 @@ is( $node_subscriber->safe_psql(
$db,
qq(SELECT apply_error_count = 0,
sync_error_count = 0,
+ insert_exists_count = 0,
+ update_differ_count = 0,
+ update_missing_count = 0,
+ delete_missing_count = 0,
stats_reset IS NOT NULL
FROM pg_stat_subscription_stats
WHERE subname = '$sub2_name')
),
- qq(t|t|t),
+ qq(t|t|t|t|t|t|t),
qq(Confirm that apply errors and sync errors are both 0 and stats_reset is not NULL for sub '$sub2_name' after reset.)
);
--
2.30.0.windows.2
[application/octet-stream] v1-0001-Detect-and-log-conflicts-in-logical-replication.patch (88.4K, ../OS0PR01MB5716352552DFADB8E9AD1D8994C92@OS0PR01MB5716.jpnprd01.prod.outlook.com/3-v1-0001-Detect-and-log-conflicts-in-logical-replication.patch)
download | inline diff:
From 9ae275a2af640e3c2b0c38ae2623e441768a44d4 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <[email protected]>
Date: Fri, 31 May 2024 14:20:45 +0530
Subject: [PATCH v1 1/2] Detect and log conflicts in logical replication
This patch adds a new parameter detect_conflict for CREATE and ALTER
subscription commands. This new parameter will decide if subscription will
go for confict detection. By default, conflict detection will be off for a
subscription.
When conflict detection is enabled, additional logging is triggered in the
following conflict scenarios:
insert_exists: Inserting a row that iolates a NOT DEFERRABLE unique constraint.
update_differ: updating a row that was previously modified by another origin.
update_missing: The tuple to be updated is missing.
delete_missing: The tuple to be deleted is missing.
For insert_exists conflict, the log can include origin and commit
timestamp details of the conflicting key with track_commit_timestamp
enabled.
update_differ conflict can only be detected when track_commit_timestamp is
enabled.
---
doc/src/sgml/catalogs.sgml | 9 +
doc/src/sgml/ref/alter_subscription.sgml | 5 +-
doc/src/sgml/ref/create_subscription.sgml | 55 ++++++
src/backend/catalog/pg_subscription.c | 1 +
src/backend/catalog/system_views.sql | 3 +-
src/backend/commands/subscriptioncmds.c | 31 +++-
src/backend/executor/execIndexing.c | 5 +-
src/backend/executor/execReplication.c | 132 +++++++++++++-
src/backend/replication/logical/Makefile | 1 +
src/backend/replication/logical/conflict.c | 188 ++++++++++++++++++++
src/backend/replication/logical/meson.build | 1 +
src/backend/replication/logical/worker.c | 50 ++++--
src/bin/pg_dump/pg_dump.c | 17 +-
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/psql/describe.c | 6 +-
src/bin/psql/tab-complete.c | 14 +-
src/include/catalog/pg_subscription.h | 4 +
src/include/replication/conflict.h | 44 +++++
src/test/regress/expected/subscription.out | 176 ++++++++++--------
src/test/regress/sql/subscription.sql | 15 ++
src/test/subscription/t/001_rep_changes.pl | 15 +-
src/test/subscription/t/013_partition.pl | 48 ++---
src/test/subscription/t/029_on_error.pl | 5 +-
src/test/subscription/t/030_origin.pl | 26 +++
src/tools/pgindent/typedefs.list | 1 +
25 files changed, 698 insertions(+), 155 deletions(-)
create mode 100644 src/backend/replication/logical/conflict.c
create mode 100644 src/include/replication/conflict.h
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index a63cc71efa..a9b6f293ea 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -8034,6 +8034,15 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>subdetectconflict</structfield> <type>bool</type>
+ </para>
+ <para>
+ If true, the subscription is enabled for conflict detection.
+ </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/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 476f195622..5f6b83e415 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -228,8 +228,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-disable-on-error"><literal>disable_on_error</literal></link>,
<link linkend="sql-createsubscription-params-with-password-required"><literal>password_required</literal></link>,
<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>, and
- <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>.
+ <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>,
+ <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
+ <link linkend="sql-createsubscription-params-with-detect-conflict"><literal>detect_conflict</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 740b7d9421..ce37fa6490 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -428,6 +428,61 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</para>
</listitem>
</varlistentry>
+
+ <varlistentry id="sql-createsubscription-params-with-detect-conflict">
+ <term><literal>detect_conflict</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ Specifies whether the subscription is enabled for conflict detection.
+ The default is <literal>false</literal>.
+ </para>
+ <para>
+ When conflict detection is enabled, additional logging is triggered
+ in the following scenarios:
+ <variablelist>
+ <varlistentry>
+ <term><literal>insert_exists</literal></term>
+ <listitem>
+ <para>
+ Inserting a row that violates a <literal>NOT DEFERRABLE</literal>
+ unique constraint. Note that to obtain the origin and commit
+ timestamp details of the conflicting key in the log, ensure that
+ <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ is enabled.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry>
+ <term><literal>update_differ</literal></term>
+ <listitem>
+ <para>
+ Updating a row that was previously modified by another origin. Note that this
+ conflict can only be detected when
+ <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ is enabled.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry>
+ <term><literal>update_missing</literal></term>
+ <listitem>
+ <para>
+ The tuple to be updated is not found.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry>
+ <term><literal>delete_missing</literal></term>
+ <listitem>
+ <para>
+ The tuple to be deleted is not found.
+ </para>
+ </listitem>
+ </varlistentry>
+ </variablelist>
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist></para>
</listitem>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 9efc9159f2..5a423f4fb0 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -72,6 +72,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->passwordrequired = subform->subpasswordrequired;
sub->runasowner = subform->subrunasowner;
sub->failover = subform->subfailover;
+ sub->detectconflict = subform->subdetectconflict;
/* Get conninfo */
datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index efb29adeb3..30393e6d67 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1360,7 +1360,8 @@ REVOKE ALL ON pg_subscription FROM public;
GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
subbinary, substream, subtwophasestate, subdisableonerr,
subpasswordrequired, subrunasowner, subfailover,
- subslotname, subsynccommit, subpublications, suborigin)
+ subdetectconflict, 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 e407428dbc..e670d72708 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -70,8 +70,9 @@
#define SUBOPT_PASSWORD_REQUIRED 0x00000800
#define SUBOPT_RUN_AS_OWNER 0x00001000
#define SUBOPT_FAILOVER 0x00002000
-#define SUBOPT_LSN 0x00004000
-#define SUBOPT_ORIGIN 0x00008000
+#define SUBOPT_DETECT_CONFLICT 0x00004000
+#define SUBOPT_LSN 0x00008000
+#define SUBOPT_ORIGIN 0x00010000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -97,6 +98,7 @@ typedef struct SubOpts
bool passwordrequired;
bool runasowner;
bool failover;
+ bool detectconflict;
char *origin;
XLogRecPtr lsn;
} SubOpts;
@@ -159,6 +161,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->runasowner = false;
if (IsSet(supported_opts, SUBOPT_FAILOVER))
opts->failover = false;
+ if (IsSet(supported_opts, SUBOPT_DETECT_CONFLICT))
+ opts->detectconflict = false;
if (IsSet(supported_opts, SUBOPT_ORIGIN))
opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
@@ -316,6 +320,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->specified_opts |= SUBOPT_FAILOVER;
opts->failover = defGetBoolean(defel);
}
+ else if (IsSet(supported_opts, SUBOPT_DETECT_CONFLICT) &&
+ strcmp(defel->defname, "detect_conflict") == 0)
+ {
+ if (IsSet(opts->specified_opts, SUBOPT_DETECT_CONFLICT))
+ errorConflictingDefElem(defel, pstate);
+
+ opts->specified_opts |= SUBOPT_DETECT_CONFLICT;
+ opts->detectconflict = defGetBoolean(defel);
+ }
else if (IsSet(supported_opts, SUBOPT_ORIGIN) &&
strcmp(defel->defname, "origin") == 0)
{
@@ -603,7 +616,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED |
- SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN);
+ SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
+ SUBOPT_DETECT_CONFLICT | SUBOPT_ORIGIN);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/*
@@ -710,6 +724,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
values[Anum_pg_subscription_subpasswordrequired - 1] = BoolGetDatum(opts.passwordrequired);
values[Anum_pg_subscription_subrunasowner - 1] = BoolGetDatum(opts.runasowner);
values[Anum_pg_subscription_subfailover - 1] = BoolGetDatum(opts.failover);
+ values[Anum_pg_subscription_subdetectconflict - 1] =
+ BoolGetDatum(opts.detectconflict);
values[Anum_pg_subscription_subconninfo - 1] =
CStringGetTextDatum(conninfo);
if (opts.slot_name)
@@ -1146,7 +1162,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR |
SUBOPT_PASSWORD_REQUIRED |
SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
- SUBOPT_ORIGIN);
+ SUBOPT_DETECT_CONFLICT | SUBOPT_ORIGIN);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1256,6 +1272,13 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
replaces[Anum_pg_subscription_subfailover - 1] = true;
}
+ if (IsSet(opts.specified_opts, SUBOPT_DETECT_CONFLICT))
+ {
+ values[Anum_pg_subscription_subdetectconflict - 1] =
+ BoolGetDatum(opts.detectconflict);
+ replaces[Anum_pg_subscription_subdetectconflict - 1] = true;
+ }
+
if (IsSet(opts.specified_opts, SUBOPT_ORIGIN))
{
values[Anum_pg_subscription_suborigin - 1] =
diff --git a/src/backend/executor/execIndexing.c b/src/backend/executor/execIndexing.c
index 9f05b3654c..d3e428e2f1 100644
--- a/src/backend/executor/execIndexing.c
+++ b/src/backend/executor/execIndexing.c
@@ -207,8 +207,9 @@ ExecOpenIndices(ResultRelInfo *resultRelInfo, bool speculative)
ii = BuildIndexInfo(indexDesc);
/*
- * If the indexes are to be used for speculative insertion, add extra
- * information required by unique index entries.
+ * If the indexes are to be used for speculative insertion or conflict
+ * detection in logical replication, add extra information required by
+ * unique index entries.
*/
if (speculative && ii->ii_Unique)
BuildSpeculativeIndexInfo(indexDesc, ii);
diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index d0a89cd577..2fd6d37cf1 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -23,6 +23,7 @@
#include "commands/trigger.h"
#include "executor/executor.h"
#include "executor/nodeModifyTable.h"
+#include "replication/conflict.h"
#include "replication/logicalrelation.h"
#include "storage/lmgr.h"
#include "utils/builtins.h"
@@ -480,6 +481,77 @@ retry:
return found;
}
+/*
+ * Find the tuple that violates the passed in unique index constraint
+ * (conflictindex).
+ *
+ * Return false if there is no conflict. Otherwise return false, and the
+ * conflicting tuple is locked and returned in *conflictslot.
+ */
+static bool
+FindConflictTuple(ResultRelInfo *resultRelInfo, EState *estate,
+ Oid conflictindex, TupleTableSlot *slot,
+ TupleTableSlot **conflictslot)
+{
+ Relation rel = resultRelInfo->ri_RelationDesc;
+ ItemPointerData conflictTid;
+ TM_FailureData tmfd;
+ TM_Result res;
+
+ Assert(OidIsValid(conflictindex));
+
+retry:
+
+ if (ExecCheckIndexConstraints(resultRelInfo, slot, estate,
+ &conflictTid, list_make1_oid(conflictindex)))
+ return false;
+
+ *conflictslot = table_slot_create(rel, NULL);
+
+ PushActiveSnapshot(GetLatestSnapshot());
+
+ res = table_tuple_lock(rel, &conflictTid, GetLatestSnapshot(),
+ *conflictslot,
+ GetCurrentCommandId(false),
+ LockTupleShare,
+ LockWaitBlock,
+ 0 /* don't follow updates */ ,
+ &tmfd);
+
+ PopActiveSnapshot();
+
+ switch (res)
+ {
+ case TM_Ok:
+ break;
+ case TM_Updated:
+ /* XXX: Improve handling here */
+ if (ItemPointerIndicatesMovedPartitions(&tmfd.ctid))
+ ereport(LOG,
+ (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+ errmsg("tuple to be locked was already moved to another partition due to concurrent update, retrying")));
+ else
+ ereport(LOG,
+ (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+ errmsg("concurrent update, retrying")));
+ goto retry;
+ case TM_Deleted:
+ /* XXX: Improve handling here */
+ ereport(LOG,
+ (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+ errmsg("concurrent delete, retrying")));
+ goto retry;
+ case TM_Invisible:
+ elog(ERROR, "attempted to lock invisible tuple");
+ break;
+ default:
+ elog(ERROR, "unexpected table_tuple_lock status: %u", res);
+ break;
+ }
+
+ return true;
+}
+
/*
* Insert tuple represented in the slot to the relation, update the indexes,
* and execute any constraints and per-row triggers.
@@ -509,6 +581,13 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
if (!skip_tuple)
{
List *recheckIndexes = NIL;
+ List *conflictindexes;
+ bool conflict = false;
+ RepOriginId origin;
+ TimestampTz committs;
+ TransactionId xmin;
+ TupleTableSlot *conflictslot;
+ Oid conflictidx = InvalidOid;
/* Compute stored generated columns */
if (rel->rd_att->constr &&
@@ -525,22 +604,57 @@ ExecSimpleRelationInsert(ResultRelInfo *resultRelInfo,
/* OK, store the tuple and create index entries for it */
simple_table_tuple_insert(resultRelInfo->ri_RelationDesc, slot);
+ conflictindexes = resultRelInfo->ri_onConflictArbiterIndexes;
+
if (resultRelInfo->ri_NumIndices > 0)
recheckIndexes = ExecInsertIndexTuples(resultRelInfo,
- slot, estate, false, false,
- NULL, NIL, false);
+ slot, estate, false,
+ conflictindexes, &conflict,
+ conflictindexes, false);
+
+ if (!conflict)
+ {
+ /* AFTER ROW INSERT Triggers */
+ ExecARInsertTriggers(estate, resultRelInfo, slot,
+ recheckIndexes, NULL);
+
+ /*
+ * XXX we should in theory pass a TransitionCaptureState object to
+ * the above to capture transition tuples, but after statement
+ * triggers don't actually get fired by replication yet anyway
+ */
+ list_free(recheckIndexes);
- /* AFTER ROW INSERT Triggers */
- ExecARInsertTriggers(estate, resultRelInfo, slot,
- recheckIndexes, NULL);
+ return;
+ }
+
+ /* Get a unique index that had potential conflicts */
+ foreach_oid(uniqueidx, conflictindexes)
+ if (list_member_oid(recheckIndexes, uniqueidx))
+ {
+ conflictidx = uniqueidx;
+ break;
+ }
/*
- * XXX we should in theory pass a TransitionCaptureState object to the
- * above to capture transition tuples, but after statement triggers
- * don't actually get fired by replication yet anyway
+ * Reports the conflict.
+ *
+ * Here, we attempt to find the conflict tuple. This operation may
+ * seem redundant with the unique violation check of indexam, but
+ * since we perform this only when we are detecting conflict in
+ * logical replication and encountering potential conflicts with any
+ * unique index constraints (which should not be frequent), so it's
+ * ok. Moreover, we are going to report an ERROR and restart the
+ * logical replication anyway, so the additional cost of finding the
+ * tuple should be acceptable.
*/
+ if (FindConflictTuple(resultRelInfo, estate, conflictidx, slot, &conflictslot))
+ GetTupleCommitTs(conflictslot, &xmin, &origin, &committs);
+ else
+ conflictslot = NULL;
- list_free(recheckIndexes);
+ ReportApplyConflict(ERROR, CT_INSERT_EXISTS, rel, conflictidx, xmin,
+ origin, committs, conflictslot);
}
}
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index ba03eeff1c..1e08bbbd4e 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -16,6 +16,7 @@ override CPPFLAGS := -I$(srcdir) $(CPPFLAGS)
OBJS = \
applyparallelworker.o \
+ conflict.o \
decode.o \
launcher.o \
logical.o \
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
new file mode 100644
index 0000000000..c8d2446033
--- /dev/null
+++ b/src/backend/replication/logical/conflict.c
@@ -0,0 +1,188 @@
+/*-------------------------------------------------------------------------
+ * conflict.c
+ * Functionality for detecting and logging conflicts.
+ *
+ * Copyright (c) 2024, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/backend/replication/logical/conflict.c
+ *
+ * This file contains the code for detecting and logging conflicts on
+ * the subscriber during logical replication.
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/commit_ts.h"
+#include "replication/conflict.h"
+#include "replication/origin.h"
+#include "utils/lsyscache.h"
+#include "utils/rel.h"
+
+const char *const ConflictTypeNames[] = {
+ [CT_INSERT_EXISTS] = "insert_exists",
+ [CT_UPDATE_DIFFER] = "update_differ",
+ [CT_UPDATE_MISSING] = "update_missing",
+ [CT_DELETE_MISSING] = "delete_missing"
+};
+
+static char *build_index_value_desc(Oid indexoid, TupleTableSlot *conflictslot);
+static int errdetail_apply_conflict(ConflictType type, Oid conflictidx,
+ TransactionId localxmin,
+ RepOriginId localorigin,
+ TimestampTz localts,
+ TupleTableSlot *conflictslot);
+
+/*
+ * Get the xmin and commit timestamp data (origin and timestamp) associated
+ * with the provided local tuple.
+ *
+ * Return true if the commit timestamp data was found, false otherwise.
+ */
+bool
+GetTupleCommitTs(TupleTableSlot *localslot, TransactionId *xmin,
+ RepOriginId *localorigin, TimestampTz *localts)
+{
+ Datum xminDatum;
+ bool isnull;
+
+ xminDatum = slot_getsysattr(localslot, MinTransactionIdAttributeNumber,
+ &isnull);
+ *xmin = DatumGetTransactionId(xminDatum);
+ Assert(!isnull);
+
+ /*
+ * The commit timestamp data is not available if track_commit_timestamp is
+ * disabled.
+ */
+ if (!track_commit_timestamp)
+ {
+ *localorigin = InvalidRepOriginId;
+ *localts = 0;
+ return false;
+ }
+
+ return TransactionIdGetCommitTsData(*xmin, localts, localorigin);
+}
+
+/*
+ * Report a conflict when applying remote changes.
+ */
+void
+ReportApplyConflict(int elevel, ConflictType type, Relation localrel,
+ Oid conflictidx, TransactionId localxmin,
+ RepOriginId localorigin, TimestampTz localts,
+ TupleTableSlot *conflictslot)
+{
+ ereport(elevel,
+ errcode(ERRCODE_INTEGRITY_CONSTRAINT_VIOLATION),
+ errmsg("conflict %s detected on relation \"%s.%s\"",
+ ConflictTypeNames[type],
+ get_namespace_name(RelationGetNamespace(localrel)),
+ RelationGetRelationName(localrel)),
+ errdetail_apply_conflict(type, conflictidx, localxmin, localorigin,
+ localts, conflictslot));
+}
+
+/*
+ * Find all unique indexes to check for a conflict and store them into
+ * ResultRelInfo.
+ */
+void
+InitConflictIndexes(ResultRelInfo *relInfo)
+{
+ List *uniqueIndexes = NIL;
+
+ for (int i = 0; i < relInfo->ri_NumIndices; i++)
+ {
+ Relation indexRelation = relInfo->ri_IndexRelationDescs[i];
+
+ if (indexRelation == NULL)
+ continue;
+
+ /* Detect conflict only for unique indexes */
+ if (!relInfo->ri_IndexRelationInfo[i]->ii_Unique)
+ continue;
+
+ /* Don't support conflict detection for deferrable index */
+ if (!indexRelation->rd_index->indimmediate)
+ continue;
+
+ uniqueIndexes = lappend_oid(uniqueIndexes,
+ RelationGetRelid(indexRelation));
+ }
+
+ relInfo->ri_onConflictArbiterIndexes = uniqueIndexes;
+}
+
+/*
+ * Add an errdetail() line showing conflict detail.
+ */
+static int
+errdetail_apply_conflict(ConflictType type, Oid conflictidx,
+ TransactionId localxmin, RepOriginId localorigin,
+ TimestampTz localts, TupleTableSlot *conflictslot)
+{
+ switch (type)
+ {
+ case CT_INSERT_EXISTS:
+ {
+ /*
+ * Bulid the index value string. If the return value is NULL,
+ * it indicates that either the conflict slot is null or the
+ * current user lacks permissions to view all the columns
+ * involved.
+ */
+ char *index_value = build_index_value_desc(conflictidx,
+ conflictslot);
+
+ if (index_value && localts)
+ return errdetail("Key %s already exists in unique index \"%s\", which was modified by origin %u in transaction %u at %s.",
+ index_value, get_rel_name(conflictidx), localorigin,
+ localxmin, timestamptz_to_str(localts));
+ else if (index_value && !localts)
+ return errdetail("Key %s already exists in unique index \"%s\", which was modified in transaction %u.",
+ index_value, get_rel_name(conflictidx), localxmin);
+ else
+ return errdetail("Key already exists in unique index \"%s\".",
+ get_rel_name(conflictidx));
+ }
+ case CT_UPDATE_DIFFER:
+ return errdetail("Updating a row that was modified by a different origin %u in transaction %u at %s.",
+ localorigin, localxmin, timestamptz_to_str(localts));
+ case CT_UPDATE_MISSING:
+ return errdetail("Did not find the row to be updated.");
+ case CT_DELETE_MISSING:
+ return errdetail("Did not find the row to be deleted.");
+ }
+
+ return 0; /* silence compiler warning */
+}
+
+/*
+ * Helper functions to construct a string describing the contents of an index
+ * entry. See BuildIndexValueDescription for details.
+ */
+static char *
+build_index_value_desc(Oid indexoid, TupleTableSlot *conflictslot)
+{
+ char *conflict_row;
+ Relation indexDesc;
+
+ if (!conflictslot)
+ return NULL;
+
+ /* Assume the index has been locked */
+ indexDesc = index_open(indexoid, NoLock);
+
+ slot_getallattrs(conflictslot);
+
+ conflict_row = BuildIndexValueDescription(indexDesc,
+ conflictslot->tts_values,
+ conflictslot->tts_isnull);
+
+ index_close(indexDesc, NoLock);
+
+ return conflict_row;
+}
diff --git a/src/backend/replication/logical/meson.build b/src/backend/replication/logical/meson.build
index 3dec36a6de..3d36249d8a 100644
--- a/src/backend/replication/logical/meson.build
+++ b/src/backend/replication/logical/meson.build
@@ -2,6 +2,7 @@
backend_sources += files(
'applyparallelworker.c',
+ 'conflict.c',
'decode.c',
'launcher.c',
'logical.c',
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index b5a80fe3e8..af73e09b01 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -167,6 +167,7 @@
#include "postmaster/bgworker.h"
#include "postmaster/interrupt.h"
#include "postmaster/walwriter.h"
+#include "replication/conflict.h"
#include "replication/logicallauncher.h"
#include "replication/logicalproto.h"
#include "replication/logicalrelation.h"
@@ -2461,7 +2462,10 @@ apply_handle_insert_internal(ApplyExecutionData *edata,
EState *estate = edata->estate;
/* We must open indexes here. */
- ExecOpenIndices(relinfo, false);
+ ExecOpenIndices(relinfo, MySubscription->detectconflict);
+
+ if (MySubscription->detectconflict)
+ InitConflictIndexes(relinfo);
/* Do the insert. */
TargetPrivilegesCheck(relinfo->ri_RelationDesc, ACL_INSERT);
@@ -2664,6 +2668,20 @@ apply_handle_update_internal(ApplyExecutionData *edata,
*/
if (found)
{
+ RepOriginId localorigin;
+ TransactionId localxmin;
+ TimestampTz localts;
+
+ /*
+ * If conflict detection is enabled, check whether the local tuple was
+ * modified by a different origin. If detected, report the conflict.
+ */
+ if (MySubscription->detectconflict &&
+ GetTupleCommitTs(localslot, &localxmin, &localorigin, &localts) &&
+ localorigin != replorigin_session_origin)
+ ReportApplyConflict(LOG, CT_UPDATE_DIFFER, localrel, InvalidOid,
+ localxmin, localorigin, localts, NULL);
+
/* Process and store remote tuple in the slot */
oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
slot_modify_data(remoteslot, localslot, relmapentry, newtup);
@@ -2681,13 +2699,10 @@ apply_handle_update_internal(ApplyExecutionData *edata,
/*
* The tuple to be updated could not be found. Do nothing except for
* emitting a log message.
- *
- * XXX should this be promoted to ereport(LOG) perhaps?
*/
- elog(DEBUG1,
- "logical replication did not find row to be updated "
- "in replication target relation \"%s\"",
- RelationGetRelationName(localrel));
+ if (MySubscription->detectconflict)
+ ReportApplyConflict(LOG, CT_UPDATE_MISSING, localrel, InvalidOid,
+ InvalidTransactionId, InvalidRepOriginId, 0, NULL);
}
/* Cleanup. */
@@ -2821,13 +2836,10 @@ apply_handle_delete_internal(ApplyExecutionData *edata,
/*
* The tuple to be deleted could not be found. Do nothing except for
* emitting a log message.
- *
- * XXX should this be promoted to ereport(LOG) perhaps?
*/
- elog(DEBUG1,
- "logical replication did not find row to be deleted "
- "in replication target relation \"%s\"",
- RelationGetRelationName(localrel));
+ if (MySubscription->detectconflict)
+ ReportApplyConflict(LOG, CT_DELETE_MISSING, localrel, InvalidOid,
+ InvalidTransactionId, InvalidRepOriginId, 0, NULL);
}
/* Cleanup. */
@@ -3005,13 +3017,13 @@ apply_handle_tuple_routing(ApplyExecutionData *edata,
/*
* The tuple to be updated could not be found. Do nothing
* except for emitting a log message.
- *
- * XXX should this be promoted to ereport(LOG) perhaps?
*/
- elog(DEBUG1,
- "logical replication did not find row to be updated "
- "in replication target relation's partition \"%s\"",
- RelationGetRelationName(partrel));
+ if (MySubscription->detectconflict)
+ ReportApplyConflict(LOG, CT_UPDATE_MISSING,
+ partrel, InvalidOid,
+ InvalidTransactionId,
+ InvalidRepOriginId, 0, NULL);
+
return;
}
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index e324070828..c6b67c692d 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4739,6 +4739,7 @@ getSubscriptions(Archive *fout)
int i_suboriginremotelsn;
int i_subenabled;
int i_subfailover;
+ int i_subdetectconflict;
int i,
ntups;
@@ -4811,11 +4812,17 @@ getSubscriptions(Archive *fout)
if (fout->remoteVersion >= 170000)
appendPQExpBufferStr(query,
- " s.subfailover\n");
+ " s.subfailover,\n");
else
appendPQExpBuffer(query,
- " false AS subfailover\n");
+ " false AS subfailover,\n");
+ if (fout->remoteVersion >= 170000)
+ appendPQExpBufferStr(query,
+ " s.subdetectconflict\n");
+ else
+ appendPQExpBuffer(query,
+ " false AS subdetectconflict\n");
appendPQExpBufferStr(query,
"FROM pg_subscription s\n");
@@ -4854,6 +4861,7 @@ getSubscriptions(Archive *fout)
i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn");
i_subenabled = PQfnumber(res, "subenabled");
i_subfailover = PQfnumber(res, "subfailover");
+ i_subdetectconflict = PQfnumber(res, "subdetectconflict");
subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo));
@@ -4900,6 +4908,8 @@ getSubscriptions(Archive *fout)
pg_strdup(PQgetvalue(res, i, i_subenabled));
subinfo[i].subfailover =
pg_strdup(PQgetvalue(res, i, i_subfailover));
+ subinfo[i].subdetectconflict =
+ pg_strdup(PQgetvalue(res, i, i_subdetectconflict));
/* Decide whether we want to dump it */
selectDumpableObject(&(subinfo[i].dobj), fout);
@@ -5140,6 +5150,9 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
if (strcmp(subinfo->subfailover, "t") == 0)
appendPQExpBufferStr(query, ", failover = true");
+ if (strcmp(subinfo->subdetectconflict, "t") == 0)
+ appendPQExpBufferStr(query, ", detect_conflict = true");
+
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 865823868f..02aa4a6f32 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -671,6 +671,7 @@ typedef struct _SubscriptionInfo
char *suborigin;
char *suboriginremotelsn;
char *subfailover;
+ char *subdetectconflict;
} SubscriptionInfo;
/*
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index f67bf0b892..0472fe2e87 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6529,7 +6529,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};
if (pset.sversion < 100000)
{
@@ -6597,6 +6597,10 @@ describeSubscriptions(const char *pattern, bool verbose)
appendPQExpBuffer(&buf,
", subfailover AS \"%s\"\n",
gettext_noop("Failover"));
+ if (pset.sversion >= 170000)
+ appendPQExpBuffer(&buf,
+ ", subdetectconflict AS \"%s\"\n",
+ gettext_noop("Detect conflict"));
appendPQExpBuffer(&buf,
", subsynccommit AS \"%s\"\n"
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index d453e224d9..219fac7e71 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1946,9 +1946,10 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH("(", "PUBLICATION");
/* ALTER SUBSCRIPTION <name> SET ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
- COMPLETE_WITH("binary", "disable_on_error", "failover", "origin",
- "password_required", "run_as_owner", "slot_name",
- "streaming", "synchronous_commit");
+ COMPLETE_WITH("binary", "detect_conflict", "disable_on_error",
+ "failover", "origin", "password_required",
+ "run_as_owner", "slot_name", "streaming",
+ "synchronous_commit");
/* ALTER SUBSCRIPTION <name> SKIP ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SKIP", "("))
COMPLETE_WITH("lsn");
@@ -3363,9 +3364,10 @@ psql_completion(const char *text, int start, int end)
/* Complete "CREATE SUBSCRIPTION <name> ... WITH ( <opt>" */
else if (HeadMatches("CREATE", "SUBSCRIPTION") && TailMatches("WITH", "("))
COMPLETE_WITH("binary", "connect", "copy_data", "create_slot",
- "disable_on_error", "enabled", "failover", "origin",
- "password_required", "run_as_owner", "slot_name",
- "streaming", "synchronous_commit", "two_phase");
+ "detect_conflict", "disable_on_error", "enabled",
+ "failover", "origin", "password_required",
+ "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_subscription.h b/src/include/catalog/pg_subscription.h
index 0aa14ec4a2..17daf11dc7 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -98,6 +98,9 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
* slots) in the upstream database are enabled
* to be synchronized to the standbys. */
+ bool subdetectconflict; /* True if replication should perform
+ * conflict detection */
+
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* Connection string to the publisher */
text subconninfo BKI_FORCE_NOT_NULL;
@@ -151,6 +154,7 @@ typedef struct Subscription
* (i.e. the main slot and the table sync
* slots) in the upstream database are enabled
* to be synchronized to the standbys. */
+ bool detectconflict; /* True if conflict detection is enabled */
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/conflict.h b/src/include/replication/conflict.h
new file mode 100644
index 0000000000..4a6743b24b
--- /dev/null
+++ b/src/include/replication/conflict.h
@@ -0,0 +1,44 @@
+/*-------------------------------------------------------------------------
+ * logical.h
+ * Exports for conflict detection and log
+ *
+ * Copyright (c) 2012-2024, PostgreSQL Global Development Group
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef CONFLICT_H
+#define CONFLICT_H
+
+#include "access/xlogdefs.h"
+#include "executor/tuptable.h"
+#include "nodes/execnodes.h"
+#include "utils/relcache.h"
+#include "utils/timestamp.h"
+
+/*
+ * Conflict types that could be encountered when applying remote changes.
+ */
+typedef enum
+{
+ /* The row to be inserted violates unique constraint */
+ CT_INSERT_EXISTS,
+
+ /* The row to be updated was modified by a different origin */
+ CT_UPDATE_DIFFER,
+
+ /* The row to be updated is missing */
+ CT_UPDATE_MISSING,
+
+ /* The row to be deleted is missing */
+ CT_DELETE_MISSING,
+} ConflictType;
+
+extern bool GetTupleCommitTs(TupleTableSlot *localslot, TransactionId *xmin,
+ RepOriginId *localorigin, TimestampTz *localts);
+extern void ReportApplyConflict(int elevel, ConflictType type,
+ Relation localrel, Oid conflictidx,
+ TransactionId localxmin, RepOriginId localorigin,
+ TimestampTz localts, TupleTableSlot *conflictslot);
+extern void InitConflictIndexes(ResultRelInfo *relInfo);
+
+#endif
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 0f2a25cdc1..a8b0086dd9 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 | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | t | 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 | Detect conflict | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | t | f | f | f | 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 | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | 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 | Detect conflict | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | f | 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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | 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 | Detect conflict | 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
(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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | t | 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 | Detect conflict | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-----------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | t | f | f | 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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | 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 | Detect conflict | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-----------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | f | 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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | 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 | Detect conflict | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-----------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | f | 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 | Synchronous commit | Conninfo | Skip LSN
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | 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 | Detect conflict | Synchronous commit | Conninfo | Skip LSN
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-----------------+--------------------+------------------------------+----------
+ regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | f | 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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | t | 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 | Detect conflict | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | t | f | f | f | 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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | 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 | Detect conflict | 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
(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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | 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 | Detect conflict | 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
(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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | 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 | Detect conflict | 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
(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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | 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 | Detect conflict | 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
(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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | 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 | Detect conflict | 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
(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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | 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 | Detect conflict | 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
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -371,10 +371,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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | 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 | Detect conflict | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
(1 row)
--fail - alter of two_phase option not supported.
@@ -383,10 +383,10 @@ ERROR: unrecognized subscription parameter: "two_phase"
-- but can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | 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 | Detect conflict | 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
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -396,10 +396,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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | 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 | Detect conflict | 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
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -412,18 +412,42 @@ 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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | 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 | Detect conflict | 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
(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 | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | t | 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 | Detect conflict | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | t | f | f | f | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+-- fail - detect_conflict must be boolean
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, detect_conflict = foo);
+ERROR: detect_conflict requires a Boolean value
+-- now it works
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, detect_conflict = true);
+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 | Detect conflict | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-----------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | t | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+ALTER SUBSCRIPTION regress_testsub SET (detect_conflict = false);
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Detect conflict | 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
(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 3e5ba4cb8c..a77b196704 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -290,6 +290,21 @@ ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
+-- fail - detect_conflict must be boolean
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, detect_conflict = foo);
+
+-- now it works
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, detect_conflict = true);
+
+\dRs+
+
+ALTER SUBSCRIPTION regress_testsub SET (detect_conflict = false);
+
+\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;
diff --git a/src/test/subscription/t/001_rep_changes.pl b/src/test/subscription/t/001_rep_changes.pl
index 471e981962..78c0307165 100644
--- a/src/test/subscription/t/001_rep_changes.pl
+++ b/src/test/subscription/t/001_rep_changes.pl
@@ -331,11 +331,12 @@ is( $result, qq(1|bar
2|baz),
'update works with REPLICA IDENTITY FULL and a primary key');
-# Check that subscriber handles cases where update/delete target tuple
-# is missing. We have to look for the DEBUG1 log messages about that,
-# so temporarily bump up the log verbosity.
-$node_subscriber->append_conf('postgresql.conf', "log_min_messages = debug1");
-$node_subscriber->reload;
+# To check that subscriber handles cases where update/delete target tuple
+# is missing, detect_conflict is temporarily enabled to log conflicts
+# related to missing tuples.
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub SET (detect_conflict = true)"
+);
$node_subscriber->safe_psql('postgres', "DELETE FROM tab_full_pk");
@@ -352,10 +353,10 @@ $node_publisher->wait_for_catchup('tap_sub');
my $logfile = slurp_file($node_subscriber->logfile, $log_location);
ok( $logfile =~
- qr/logical replication did not find row to be updated in replication target relation "tab_full_pk"/,
+ qr/conflict update_missing detected on relation "public.tab_full_pk".*\n.*DETAIL:.* Did not find the row to be updated./m,
'update target row is missing');
ok( $logfile =~
- qr/logical replication did not find row to be deleted in replication target relation "tab_full_pk"/,
+ qr/conflict delete_missing detected on relation "public.tab_full_pk".*\n.*DETAIL:.* Did not find the row to be deleted./m,
'delete target row is missing');
$node_subscriber->append_conf('postgresql.conf',
diff --git a/src/test/subscription/t/013_partition.pl b/src/test/subscription/t/013_partition.pl
index 29580525a9..c64f17211d 100644
--- a/src/test/subscription/t/013_partition.pl
+++ b/src/test/subscription/t/013_partition.pl
@@ -343,12 +343,12 @@ $result =
$node_subscriber2->safe_psql('postgres', "SELECT a FROM tab1 ORDER BY 1");
is($result, qq(), 'truncate of tab1 replicated');
-# Check that subscriber handles cases where update/delete target tuple
-# is missing. We have to look for the DEBUG1 log messages about that,
-# so temporarily bump up the log verbosity.
-$node_subscriber1->append_conf('postgresql.conf',
- "log_min_messages = debug1");
-$node_subscriber1->reload;
+# To check that subscriber handles cases where update/delete target tuple
+# is missing, detect_conflict is temporarily enabled to log conflicts
+# related to missing tuples.
+$node_subscriber1->safe_psql('postgres',
+ "ALTER SUBSCRIPTION sub1 SET (detect_conflict = true)"
+);
$node_publisher->safe_psql('postgres',
"INSERT INTO tab1 VALUES (1, 'foo'), (4, 'bar'), (10, 'baz')");
@@ -372,21 +372,21 @@ $node_publisher->wait_for_catchup('sub2');
my $logfile = slurp_file($node_subscriber1->logfile(), $log_location);
ok( $logfile =~
- qr/logical replication did not find row to be updated in replication target relation's partition "tab1_2_2"/,
+ qr/conflict update_missing detected on relation "public.tab1_2_2".*\n.*DETAIL:.* Did not find the row to be updated./,
'update target row is missing in tab1_2_2');
ok( $logfile =~
- qr/logical replication did not find row to be deleted in replication target relation "tab1_1"/,
+ qr/conflict delete_missing detected on relation "public.tab1_1".*\n.*DETAIL:.* Did not find the row to be deleted./,
'delete target row is missing in tab1_1');
ok( $logfile =~
- qr/logical replication did not find row to be deleted in replication target relation "tab1_2_2"/,
+ qr/conflict delete_missing detected on relation "public.tab1_2_2".*\n.*DETAIL:.* Did not find the row to be deleted./,
'delete target row is missing in tab1_2_2');
ok( $logfile =~
- qr/logical replication did not find row to be deleted in replication target relation "tab1_def"/,
+ qr/conflict delete_missing detected on relation "public.tab1_def".*\n.*DETAIL:.* Did not find the row to be deleted./,
'delete target row is missing in tab1_def');
-$node_subscriber1->append_conf('postgresql.conf',
- "log_min_messages = warning");
-$node_subscriber1->reload;
+$node_subscriber1->safe_psql('postgres',
+ "ALTER SUBSCRIPTION sub1 SET (detect_conflict = false)"
+);
# Tests for replication using root table identity and schema
@@ -773,12 +773,12 @@ pub_tab2|3|yyy
pub_tab2|5|zzz
xxx_c|6|aaa), 'inserts into tab2 replicated');
-# Check that subscriber handles cases where update/delete target tuple
-# is missing. We have to look for the DEBUG1 log messages about that,
-# so temporarily bump up the log verbosity.
-$node_subscriber1->append_conf('postgresql.conf',
- "log_min_messages = debug1");
-$node_subscriber1->reload;
+# To check that subscriber handles cases where update/delete target tuple
+# is missing, detect_conflict is temporarily enabled to log conflicts
+# related to missing tuples.
+$node_subscriber1->safe_psql('postgres',
+ "ALTER SUBSCRIPTION sub_viaroot SET (detect_conflict = true)"
+);
$node_subscriber1->safe_psql('postgres', "DELETE FROM tab2");
@@ -796,15 +796,15 @@ $node_publisher->wait_for_catchup('sub2');
$logfile = slurp_file($node_subscriber1->logfile(), $log_location);
ok( $logfile =~
- qr/logical replication did not find row to be updated in replication target relation's partition "tab2_1"/,
+ qr/conflict update_missing detected on relation "public.tab2_1".*\n.*DETAIL:.* Did not find the row to be updated./,
'update target row is missing in tab2_1');
ok( $logfile =~
- qr/logical replication did not find row to be deleted in replication target relation "tab2_1"/,
+ qr/conflict delete_missing detected on relation "public.tab2_1".*\n.*DETAIL:.* Did not find the row to be deleted./,
'delete target row is missing in tab2_1');
-$node_subscriber1->append_conf('postgresql.conf',
- "log_min_messages = warning");
-$node_subscriber1->reload;
+$node_subscriber1->safe_psql('postgres',
+ "ALTER SUBSCRIPTION sub_viaroot SET (detect_conflict = false)"
+);
# Test that replication continues to work correctly after altering the
# partition of a partitioned target table.
diff --git a/src/test/subscription/t/029_on_error.pl b/src/test/subscription/t/029_on_error.pl
index 0ab57a4b5b..496a3c6cd9 100644
--- a/src/test/subscription/t/029_on_error.pl
+++ b/src/test/subscription/t/029_on_error.pl
@@ -30,7 +30,7 @@ sub test_skip_lsn
# ERROR with its CONTEXT when retrieving this information.
my $contents = slurp_file($node_subscriber->logfile, $offset);
$contents =~
- qr/duplicate key value violates unique constraint "tbl_pkey".*\n.*DETAIL:.*\n.*CONTEXT:.* for replication target relation "public.tbl" in transaction \d+, finished at ([[:xdigit:]]+\/[[:xdigit:]]+)/m
+ qr/conflict insert_exists detected on relation "public.tbl".*\n.*DETAIL:.* Key \(i\)=\(1\) already exists in unique index "tbl_pkey", which was modified by origin \d+ in transaction \d+ at .*\n.*CONTEXT:.* for replication target relation "public.tbl" in transaction \d+, finished at ([[:xdigit:]]+\/[[:xdigit:]]+)/m
or die "could not get error-LSN";
my $lsn = $1;
@@ -83,6 +83,7 @@ $node_subscriber->append_conf(
'postgresql.conf',
qq[
max_prepared_transactions = 10
+track_commit_timestamp = on
]);
$node_subscriber->start;
@@ -109,7 +110,7 @@ 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, streaming = on, two_phase = on)"
+ "CREATE SUBSCRIPTION sub CONNECTION '$publisher_connstr' PUBLICATION pub WITH (disable_on_error = true, streaming = on, two_phase = on, detect_conflict = on)"
);
# Initial synchronization failure causes the subscription to be disabled.
diff --git a/src/test/subscription/t/030_origin.pl b/src/test/subscription/t/030_origin.pl
index 056561f008..03dabfeb72 100644
--- a/src/test/subscription/t/030_origin.pl
+++ b/src/test/subscription/t/030_origin.pl
@@ -26,7 +26,12 @@ my $stderr;
# node_A
my $node_A = PostgreSQL::Test::Cluster->new('node_A');
$node_A->init(allows_streaming => 'logical');
+
+# Enable the track_commit_timestamp to detect the conflict when attempting to
+# update a row that was previously modified by a different origin.
+$node_A->append_conf('postgresql.conf', 'track_commit_timestamp = on');
$node_A->start;
+
# node_B
my $node_B = PostgreSQL::Test::Cluster->new('node_B');
$node_B->init(allows_streaming => 'logical');
@@ -89,11 +94,32 @@ is( $result, qq(11
'Inserted successfully without leading to infinite recursion in bidirectional replication setup'
);
+###############################################################################
+# Check that the conflict can be detected when attempting to update a row that
+# was previously modified by a different source.
+###############################################################################
+
+$node_A->safe_psql('postgres',
+ "ALTER SUBSCRIPTION $subname_AB SET (detect_conflict = true);");
+
+$node_B->safe_psql('postgres', "UPDATE tab SET a = 10 WHERE a = 11;");
+
+$node_A->wait_for_log(
+ qr/Updating a row that was modified by a different origin [0-9]+ in transaction [0-9]+ at .*/
+);
+
$node_A->safe_psql('postgres', "DELETE FROM tab;");
$node_A->wait_for_catchup($subname_BA);
$node_B->wait_for_catchup($subname_AB);
+# The remaining tests no longer test conflict detection.
+$node_A->safe_psql('postgres',
+ "ALTER SUBSCRIPTION $subname_AB SET (detect_conflict = false);");
+
+$node_A->append_conf('postgresql.conf', 'track_commit_timestamp = off');
+$node_A->restart;
+
###############################################################################
# Check that remote data of node_B (that originated from node_C) is not
# published to node_A.
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 61ad417cde..f42efe12b7 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -465,6 +465,7 @@ ConditionVariableMinimallyPadded
ConditionalStack
ConfigData
ConfigVariable
+ConflictType
ConnCacheEntry
ConnCacheKey
ConnParams
--
2.30.0.windows.2
view thread (2+ messages)
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: Conflict detection and logging in logical replication
In-Reply-To: <OS0PR01MB5716352552DFADB8E9AD1D8994C92@OS0PR01MB5716.jpnprd01.prod.outlook.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox