From b2be8e767b62542e774968420fef6457714f2762 Mon Sep 17 00:00:00 2001
From: "Paul A. Jungwirth" <pj@illuminatedcomputing.com>
Date: Wed, 16 Oct 2024 11:06:40 -0700
Subject: [PATCH v42 3/7] Fix logical replication for temporal tables

A WITHOUT OVERLAPS primary key or unique constraint should work as a
REPLICA IDENTITY, since it guarantees uniqueness. But it is a GiST
index, not a btree, so logical replication was getting confused. To
determine the equals operator, we can use the stratnum GiST support
function.
---
 src/backend/commands/tablecmds.c              |  11 +-
 src/backend/executor/execReplication.c        |   9 +-
 .../regress/expected/without_overlaps.out     |  11 +-
 src/test/regress/sql/without_overlaps.sql     |   2 +-
 src/test/subscription/t/034_temporal.pl       | 668 ++++++++++++++++++
 5 files changed, 694 insertions(+), 7 deletions(-)
 create mode 100644 src/test/subscription/t/034_temporal.pl

diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index e14bc0c0548..93361249bdb 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -16884,9 +16884,14 @@ ATExecReplicaIdentity(Relation rel, ReplicaIdentityStmt *stmt, LOCKMODE lockmode
 				 errmsg("\"%s\" is not an index for table \"%s\"",
 						RelationGetRelationName(indexRel),
 						RelationGetRelationName(rel))));
-	/* The AM must support uniqueness, and the index must in fact be unique. */
-	if (!indexRel->rd_indam->amcanunique ||
-		!indexRel->rd_index->indisunique)
+	/*
+	 * The AM must support uniqueness, and the index must in fact be unique.
+	 * If we have a WITHOUT OVERLAPS constraint (identified by uniqueness +
+	 * exclusion), we can use that too.
+	 */
+	if ((!indexRel->rd_indam->amcanunique ||
+		!indexRel->rd_index->indisunique) &&
+		!(indexRel->rd_index->indisunique && indexRel->rd_index->indisexclusion))
 		ereport(ERROR,
 				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 				 errmsg("cannot use non-unique index \"%s\" as replica identity",
diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index 54025c9f150..152fe3140a0 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -15,6 +15,7 @@
 #include "postgres.h"
 
 #include "access/genam.h"
+#include "access/gist.h"
 #include "access/relscan.h"
 #include "access/tableam.h"
 #include "access/transam.h"
@@ -77,7 +78,11 @@ get_equal_strategy_number(Oid opclass)
 {
 	Oid			am = get_opclass_method(opclass);
 
-	return get_equal_strategy_number_for_am(am);
+	/* For GiST indexes we need to ask the opclass what strategy number to use. */
+	if (am == GIST_AM_OID)
+		return GistTranslateStratnum(opclass, RTEqualStrategyNumber);
+	else
+		return get_equal_strategy_number_for_am(am);
 }
 
 /*
@@ -134,6 +139,8 @@ build_replindex_scan_key(ScanKey skey, Relation rel, Relation idxrel,
 		optype = get_opclass_input_type(opclass->values[index_attoff]);
 		opfamily = get_opclass_family(opclass->values[index_attoff]);
 		eq_strategy = get_equal_strategy_number(opclass->values[index_attoff]);
+		if (!OidIsValid(eq_strategy))
+			elog(ERROR, "missing equal strategy for opclass %u", opclass->values[index_attoff]);
 
 		operator = get_opfamily_member(opfamily, optype,
 									   optype,
diff --git a/src/test/regress/expected/without_overlaps.out b/src/test/regress/expected/without_overlaps.out
index 86171c994c9..a05b64e2f78 100644
--- a/src/test/regress/expected/without_overlaps.out
+++ b/src/test/regress/expected/without_overlaps.out
@@ -978,9 +978,16 @@ SELECT * FROM tp2 ORDER BY id, valid_at;
 
 DROP TABLE temporal_partitioned;
 -- ALTER TABLE REPLICA IDENTITY
--- (should fail)
 ALTER TABLE temporal_rng REPLICA IDENTITY USING INDEX temporal_rng_pk;
-ERROR:  cannot use non-unique index "temporal_rng_pk" as replica identity
+\d temporal_rng
+              Table "public.temporal_rng"
+  Column  |   Type    | Collation | Nullable | Default 
+----------+-----------+-----------+----------+---------
+ id       | int4range |           | not null | 
+ valid_at | daterange |           | not null | 
+Indexes:
+    "temporal_rng_pk" PRIMARY KEY (id, valid_at WITHOUT OVERLAPS) REPLICA IDENTITY
+
 --
 -- ON CONFLICT: ranges
 --
diff --git a/src/test/regress/sql/without_overlaps.sql b/src/test/regress/sql/without_overlaps.sql
index 943edf3da63..ebee2f2932e 100644
--- a/src/test/regress/sql/without_overlaps.sql
+++ b/src/test/regress/sql/without_overlaps.sql
@@ -691,8 +691,8 @@ SELECT * FROM tp2 ORDER BY id, valid_at;
 DROP TABLE temporal_partitioned;
 
 -- ALTER TABLE REPLICA IDENTITY
--- (should fail)
 ALTER TABLE temporal_rng REPLICA IDENTITY USING INDEX temporal_rng_pk;
+\d temporal_rng
 
 --
 -- ON CONFLICT: ranges
diff --git a/src/test/subscription/t/034_temporal.pl b/src/test/subscription/t/034_temporal.pl
new file mode 100644
index 00000000000..3ef4249a9fe
--- /dev/null
+++ b/src/test/subscription/t/034_temporal.pl
@@ -0,0 +1,668 @@
+
+# Copyright (c) 2021-2024, PostgreSQL Global Development Group
+
+# Logical replication tests for temporal tables
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# setup
+
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->start;
+
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->start;
+
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+
+my ($result, $stdout, $stderr);
+
+# #################################
+# Test with REPLICA IDENTITY DEFAULT:
+# #################################
+
+# create tables on publisher
+
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE temporal_no_key (id int4range, valid_at daterange, a text)"
+);
+
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE temporal_pk (id int4range, valid_at daterange, a text, PRIMARY KEY (id, valid_at WITHOUT OVERLAPS))"
+);
+
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE temporal_unique (id int4range NOT NULL, valid_at daterange NOT NULL, a text, UNIQUE (id, valid_at WITHOUT OVERLAPS))"
+);
+
+# create tables on subscriber
+
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE temporal_no_key (id int4range, valid_at daterange, a text)"
+);
+
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE temporal_pk (id int4range, valid_at daterange, a text, PRIMARY KEY (id, valid_at WITHOUT OVERLAPS))"
+);
+
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE temporal_unique (id int4range NOT NULL, valid_at daterange NOT NULL, a text, UNIQUE (id, valid_at WITHOUT OVERLAPS))"
+);
+
+# sync initial data:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_no_key (id, valid_at, a)
+   VALUES ('[1,2)', '[2000-01-01,2010-01-01)', 'a')");
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_pk (id, valid_at, a)
+   VALUES ('[1,2)', '[2000-01-01,2010-01-01)', 'a')");
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_unique (id, valid_at, a)
+   VALUES ('[1,2)', '[2000-01-01,2010-01-01)', 'a')");
+
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION pub1 FOR ALL TABLES");
+$node_subscriber->safe_psql('postgres',
+	"CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub1"
+);
+$node_subscriber->wait_for_subscription_sync;
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_no_key ORDER BY id, valid_at");
+is( $result,
+	qq{[1,2)|[2000-01-01,2010-01-01)|a},
+	'synced temporal_no_key DEFAULT');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_pk ORDER BY id, valid_at");
+is($result, qq{[1,2)|[2000-01-01,2010-01-01)|a},
+	'synced temporal_pk DEFAULT');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_unique ORDER BY id, valid_at");
+is( $result,
+	qq{[1,2)|[2000-01-01,2010-01-01)|a},
+	'synced temporal_unique DEFAULT');
+
+# replicate with no key:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_no_key (id, valid_at, a)
+   VALUES ('[2,3)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[3,4)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[4,5)', '[2000-01-01,2010-01-01)', 'a')");
+
+($result, $stdout, $stderr) = $node_publisher->psql('postgres',
+	"UPDATE temporal_no_key SET a = 'b' WHERE id = '[2,3)'");
+is( $stderr,
+	qq(psql:<stdin>:1: ERROR:  cannot update table "temporal_no_key" because it does not have a replica identity and publishes updates
+HINT:  To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.),
+	"can't UPDATE temporal_no_key DEFAULT");
+
+($result, $stdout, $stderr) = $node_publisher->psql('postgres',
+	"DELETE FROM temporal_no_key WHERE id = '[3,4)'");
+is( $stderr,
+	qq(psql:<stdin>:1: ERROR:  cannot delete from table "temporal_no_key" because it does not have a replica identity and publishes deletes
+HINT:  To enable deleting from the table, set REPLICA IDENTITY using ALTER TABLE.),
+	"can't DELETE temporal_no_key DEFAULT");
+
+$node_publisher->wait_for_catchup('sub1');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_no_key ORDER BY id, valid_at");
+is( $result, qq{[1,2)|[2000-01-01,2010-01-01)|a
+[2,3)|[2000-01-01,2010-01-01)|a
+[3,4)|[2000-01-01,2010-01-01)|a
+[4,5)|[2000-01-01,2010-01-01)|a}, 'replicated temporal_no_key DEFAULT');
+
+# replicate with a primary key:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_pk (id, valid_at, a)
+   VALUES ('[2,3)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[3,4)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[4,5)', '[2000-01-01,2010-01-01)', 'a')");
+
+$node_publisher->safe_psql('postgres',
+	"UPDATE temporal_pk SET a = 'b' WHERE id = '[2,3)'");
+
+$node_publisher->safe_psql('postgres',
+	"DELETE FROM temporal_pk WHERE id = '[3,4)'");
+
+$node_publisher->wait_for_catchup('sub1');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_pk ORDER BY id, valid_at");
+is( $result, qq{[1,2)|[2000-01-01,2010-01-01)|a
+[2,3)|[2000-01-01,2010-01-01)|b
+[4,5)|[2000-01-01,2010-01-01)|a}, 'replicated temporal_pk DEFAULT');
+
+# replicate with a unique key:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_unique (id, valid_at, a)
+   VALUES ('[2,3)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[3,4)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[4,5)', '[2000-01-01,2010-01-01)', 'a')");
+
+($result, $stdout, $stderr) = $node_publisher->psql('postgres',
+	"UPDATE temporal_unique SET a = 'b' WHERE id = '[2,3)'");
+is( $stderr,
+	qq(psql:<stdin>:1: ERROR:  cannot update table "temporal_unique" because it does not have a replica identity and publishes updates
+HINT:  To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.),
+	"can't UPDATE temporal_unique DEFAULT");
+
+($result, $stdout, $stderr) = $node_publisher->psql('postgres',
+	"DELETE FROM temporal_unique WHERE id = '[3,4)'");
+is( $stderr,
+	qq(psql:<stdin>:1: ERROR:  cannot delete from table "temporal_unique" because it does not have a replica identity and publishes deletes
+HINT:  To enable deleting from the table, set REPLICA IDENTITY using ALTER TABLE.),
+	"can't DELETE temporal_unique DEFAULT");
+
+$node_publisher->wait_for_catchup('sub1');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_unique ORDER BY id, valid_at");
+is( $result, qq{[1,2)|[2000-01-01,2010-01-01)|a
+[2,3)|[2000-01-01,2010-01-01)|a
+[3,4)|[2000-01-01,2010-01-01)|a
+[4,5)|[2000-01-01,2010-01-01)|a}, 'replicated temporal_unique DEFAULT');
+
+# cleanup
+
+$node_publisher->safe_psql('postgres', "DROP TABLE temporal_no_key");
+$node_publisher->safe_psql('postgres', "DROP TABLE temporal_pk");
+$node_publisher->safe_psql('postgres', "DROP TABLE temporal_unique");
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION pub1");
+$node_subscriber->safe_psql('postgres', "DROP TABLE temporal_no_key");
+$node_subscriber->safe_psql('postgres', "DROP TABLE temporal_pk");
+$node_subscriber->safe_psql('postgres', "DROP TABLE temporal_unique");
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION sub1");
+
+
+# #################################
+# Test with REPLICA IDENTITY FULL:
+# #################################
+
+# create tables on publisher
+
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE temporal_no_key (id int4range, valid_at daterange, a text)"
+);
+$node_publisher->safe_psql('postgres',
+	"ALTER TABLE temporal_no_key REPLICA IDENTITY FULL");
+
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE temporal_pk (id int4range, valid_at daterange, a text, PRIMARY KEY (id, valid_at WITHOUT OVERLAPS))"
+);
+$node_publisher->safe_psql('postgres',
+	"ALTER TABLE temporal_pk REPLICA IDENTITY FULL");
+
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE temporal_unique (id int4range NOT NULL, valid_at daterange NOT NULL, a text, UNIQUE (id, valid_at WITHOUT OVERLAPS))"
+);
+$node_publisher->safe_psql('postgres',
+	"ALTER TABLE temporal_unique REPLICA IDENTITY FULL");
+
+# create tables on subscriber
+
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE temporal_no_key (id int4range, valid_at daterange, a text)"
+);
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE temporal_no_key REPLICA IDENTITY FULL");
+
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE temporal_pk (id int4range, valid_at daterange, a text, PRIMARY KEY (id, valid_at WITHOUT OVERLAPS))"
+);
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE temporal_pk REPLICA IDENTITY FULL");
+
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE temporal_unique (id int4range NOT NULL, valid_at daterange NOT NULL, a text, UNIQUE (id, valid_at WITHOUT OVERLAPS))"
+);
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE temporal_unique REPLICA IDENTITY FULL");
+
+# sync initial data:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_no_key (id, valid_at, a)
+   VALUES ('[1,2)', '[2000-01-01,2010-01-01)', 'a')");
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_pk (id, valid_at, a)
+   VALUES ('[1,2)', '[2000-01-01,2010-01-01)', 'a')");
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_unique (id, valid_at, a)
+   VALUES ('[1,2)', '[2000-01-01,2010-01-01)', 'a')");
+
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION pub1 FOR ALL TABLES");
+$node_subscriber->safe_psql('postgres',
+	"CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub1"
+);
+$node_subscriber->wait_for_subscription_sync;
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_no_key ORDER BY id, valid_at");
+is( $result,
+	qq{[1,2)|[2000-01-01,2010-01-01)|a},
+	'synced temporal_no_key FULL');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_pk ORDER BY id, valid_at");
+is($result, qq{[1,2)|[2000-01-01,2010-01-01)|a}, 'synced temporal_pk FULL');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_unique ORDER BY id, valid_at");
+is( $result,
+	qq{[1,2)|[2000-01-01,2010-01-01)|a},
+	'synced temporal_unique FULL');
+
+# replicate with no key:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_no_key (id, valid_at, a)
+   VALUES ('[2,3)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[3,4)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[4,5)', '[2000-01-01,2010-01-01)', 'a')");
+
+$node_publisher->safe_psql('postgres',
+	"UPDATE temporal_no_key SET a = 'b' WHERE id = '[2,3)'");
+
+$node_publisher->safe_psql('postgres',
+	"DELETE FROM temporal_no_key WHERE id = '[3,4)'");
+
+$node_publisher->wait_for_catchup('sub1');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_no_key ORDER BY id, valid_at");
+is( $result, qq{[1,2)|[2000-01-01,2010-01-01)|a
+[2,3)|[2000-01-01,2010-01-01)|b
+[4,5)|[2000-01-01,2010-01-01)|a}, 'replicated temporal_no_key FULL');
+
+# replicate with a primary key:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_pk (id, valid_at, a)
+   VALUES ('[2,3)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[3,4)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[4,5)', '[2000-01-01,2010-01-01)', 'a')");
+
+$node_publisher->safe_psql('postgres',
+	"UPDATE temporal_pk SET a = 'b' WHERE id = '[2,3)'");
+
+$node_publisher->safe_psql('postgres',
+	"DELETE FROM temporal_pk WHERE id = '[3,4)'");
+
+$node_publisher->wait_for_catchup('sub1');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_pk ORDER BY id, valid_at");
+is( $result, qq{[1,2)|[2000-01-01,2010-01-01)|a
+[2,3)|[2000-01-01,2010-01-01)|b
+[4,5)|[2000-01-01,2010-01-01)|a}, 'replicated temporal_pk FULL');
+
+# replicate with a unique key:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_unique (id, valid_at, a)
+   VALUES ('[2,3)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[3,4)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[4,5)', '[2000-01-01,2010-01-01)', 'a')");
+
+$node_publisher->safe_psql('postgres',
+	"UPDATE temporal_unique SET a = 'b' WHERE id = '[2,3)'");
+
+$node_publisher->safe_psql('postgres',
+	"DELETE FROM temporal_unique WHERE id = '[3,4)'");
+
+$node_publisher->wait_for_catchup('sub1');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_unique ORDER BY id, valid_at");
+is( $result, qq{[1,2)|[2000-01-01,2010-01-01)|a
+[2,3)|[2000-01-01,2010-01-01)|b
+[4,5)|[2000-01-01,2010-01-01)|a}, 'replicated temporal_unique FULL');
+
+# cleanup
+
+$node_publisher->safe_psql('postgres', "DROP TABLE temporal_no_key");
+$node_publisher->safe_psql('postgres', "DROP TABLE temporal_pk");
+$node_publisher->safe_psql('postgres', "DROP TABLE temporal_unique");
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION pub1");
+$node_subscriber->safe_psql('postgres', "DROP TABLE temporal_no_key");
+$node_subscriber->safe_psql('postgres', "DROP TABLE temporal_pk");
+$node_subscriber->safe_psql('postgres', "DROP TABLE temporal_unique");
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION sub1");
+
+
+# #################################
+# Test with REPLICA IDENTITY USING INDEX
+# #################################
+
+# create tables on publisher
+
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE temporal_pk (id int4range, valid_at daterange, a text, PRIMARY KEY (id, valid_at WITHOUT OVERLAPS))"
+);
+$node_publisher->safe_psql('postgres',
+	"ALTER TABLE temporal_pk REPLICA IDENTITY USING INDEX temporal_pk_pkey");
+
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE temporal_unique (id int4range NOT NULL, valid_at daterange NOT NULL, a text, UNIQUE (id, valid_at WITHOUT OVERLAPS))"
+);
+$node_publisher->safe_psql('postgres',
+	"ALTER TABLE temporal_unique REPLICA IDENTITY USING INDEX temporal_unique_id_valid_at_key"
+);
+
+# create tables on subscriber
+
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE temporal_pk (id int4range, valid_at daterange, a text, PRIMARY KEY (id, valid_at WITHOUT OVERLAPS))"
+);
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE temporal_pk REPLICA IDENTITY USING INDEX temporal_pk_pkey");
+
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE temporal_unique (id int4range NOT NULL, valid_at daterange NOT NULL, a text, UNIQUE (id, valid_at WITHOUT OVERLAPS))"
+);
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE temporal_unique REPLICA IDENTITY USING INDEX temporal_unique_id_valid_at_key"
+);
+
+# sync initial data:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_pk (id, valid_at, a)
+   VALUES ('[1,2)', '[2000-01-01,2010-01-01)', 'a')");
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_unique (id, valid_at, a)
+   VALUES ('[1,2)', '[2000-01-01,2010-01-01)', 'a')");
+
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION pub1 FOR ALL TABLES");
+$node_subscriber->safe_psql('postgres',
+	"CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub1"
+);
+$node_subscriber->wait_for_subscription_sync;
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_pk ORDER BY id, valid_at");
+is( $result,
+	qq{[1,2)|[2000-01-01,2010-01-01)|a},
+	'synced temporal_pk USING INDEX');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_unique ORDER BY id, valid_at");
+is( $result,
+	qq{[1,2)|[2000-01-01,2010-01-01)|a},
+	'synced temporal_unique USING INDEX');
+
+# replicate with a primary key:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_pk (id, valid_at, a)
+   VALUES ('[2,3)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[3,4)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[4,5)', '[2000-01-01,2010-01-01)', 'a')");
+
+$node_publisher->safe_psql('postgres',
+	"UPDATE temporal_pk SET a = 'b' WHERE id = '[2,3)'");
+
+$node_publisher->safe_psql('postgres',
+	"DELETE FROM temporal_pk WHERE id = '[3,4)'");
+
+$node_publisher->wait_for_catchup('sub1');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_pk ORDER BY id, valid_at");
+is( $result, qq{[1,2)|[2000-01-01,2010-01-01)|a
+[2,3)|[2000-01-01,2010-01-01)|b
+[4,5)|[2000-01-01,2010-01-01)|a}, 'replicated temporal_pk USING INDEX');
+
+# replicate with a unique key:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_unique (id, valid_at, a)
+   VALUES ('[2,3)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[3,4)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[4,5)', '[2000-01-01,2010-01-01)', 'a')");
+
+$node_publisher->safe_psql('postgres',
+	"UPDATE temporal_unique SET a = 'b' WHERE id = '[2,3)'");
+
+$node_publisher->safe_psql('postgres',
+	"DELETE FROM temporal_unique WHERE id = '[3,4)'");
+
+$node_publisher->wait_for_catchup('sub1');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_unique ORDER BY id, valid_at");
+is( $result, qq{[1,2)|[2000-01-01,2010-01-01)|a
+[2,3)|[2000-01-01,2010-01-01)|b
+[4,5)|[2000-01-01,2010-01-01)|a}, 'replicated temporal_unique USING INDEX');
+
+# cleanup
+
+$node_publisher->safe_psql('postgres', "DROP TABLE temporal_pk");
+$node_publisher->safe_psql('postgres', "DROP TABLE temporal_unique");
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION pub1");
+$node_subscriber->safe_psql('postgres', "DROP TABLE temporal_pk");
+$node_subscriber->safe_psql('postgres', "DROP TABLE temporal_unique");
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION sub1");
+
+
+# #################################
+# Test with REPLICA IDENTITY NOTHING
+# #################################
+
+# create tables on publisher
+
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE temporal_no_key (id int4range, valid_at daterange, a text, PRIMARY KEY (id, valid_at WITHOUT OVERLAPS))"
+);
+$node_publisher->safe_psql('postgres',
+	"ALTER TABLE temporal_no_key REPLICA IDENTITY NOTHING");
+
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE temporal_pk (id int4range, valid_at daterange, a text, PRIMARY KEY (id, valid_at WITHOUT OVERLAPS))"
+);
+$node_publisher->safe_psql('postgres',
+	"ALTER TABLE temporal_pk REPLICA IDENTITY NOTHING");
+
+$node_publisher->safe_psql('postgres',
+	"CREATE TABLE temporal_unique (id int4range, valid_at daterange, a text, UNIQUE (id, valid_at WITHOUT OVERLAPS))"
+);
+$node_publisher->safe_psql('postgres',
+	"ALTER TABLE temporal_unique REPLICA IDENTITY NOTHING");
+
+# create tables on subscriber
+
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE temporal_no_key (id int4range, valid_at daterange, a text)"
+);
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE temporal_no_key REPLICA IDENTITY NOTHING");
+
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE temporal_pk (id int4range, valid_at daterange, a text, PRIMARY KEY (id, valid_at WITHOUT OVERLAPS))"
+);
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE temporal_pk REPLICA IDENTITY NOTHING");
+
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE temporal_unique (id int4range, valid_at daterange, a text, UNIQUE (id, valid_at WITHOUT OVERLAPS))"
+);
+$node_subscriber->safe_psql('postgres',
+	"ALTER TABLE temporal_unique REPLICA IDENTITY NOTHING");
+
+# sync initial data:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_no_key (id, valid_at, a)
+   VALUES ('[1,2)', '[2000-01-01,2010-01-01)', 'a')");
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_pk (id, valid_at, a)
+   VALUES ('[1,2)', '[2000-01-01,2010-01-01)', 'a')");
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_unique (id, valid_at, a)
+   VALUES ('[1,2)', '[2000-01-01,2010-01-01)', 'a')");
+
+$node_publisher->safe_psql('postgres',
+	"CREATE PUBLICATION pub1 FOR ALL TABLES");
+$node_subscriber->safe_psql('postgres',
+	"CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub1"
+);
+$node_subscriber->wait_for_subscription_sync;
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_no_key ORDER BY id, valid_at");
+is( $result,
+	qq{[1,2)|[2000-01-01,2010-01-01)|a},
+	'synced temporal_no_key NOTHING');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_pk ORDER BY id, valid_at");
+is($result, qq{[1,2)|[2000-01-01,2010-01-01)|a},
+	'synced temporal_pk NOTHING');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_unique ORDER BY id, valid_at");
+is( $result,
+	qq{[1,2)|[2000-01-01,2010-01-01)|a},
+	'synced temporal_unique NOTHING');
+
+# replicate with no key:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_no_key (id, valid_at, a)
+   VALUES ('[2,3)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[3,4)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[4,5)', '[2000-01-01,2010-01-01)', 'a')");
+
+($result, $stdout, $stderr) = $node_publisher->psql('postgres',
+	"UPDATE temporal_no_key SET a = 'b' WHERE id = '[2,3)'");
+is( $stderr,
+	qq(psql:<stdin>:1: ERROR:  cannot update table "temporal_no_key" because it does not have a replica identity and publishes updates
+HINT:  To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.),
+	"can't UPDATE temporal_no_key NOTHING");
+
+($result, $stdout, $stderr) = $node_publisher->psql('postgres',
+	"DELETE FROM temporal_no_key WHERE id = '[3,4)'");
+is( $stderr,
+	qq(psql:<stdin>:1: ERROR:  cannot delete from table "temporal_no_key" because it does not have a replica identity and publishes deletes
+HINT:  To enable deleting from the table, set REPLICA IDENTITY using ALTER TABLE.),
+	"can't DELETE temporal_no_key NOTHING");
+
+$node_publisher->wait_for_catchup('sub1');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_no_key ORDER BY id, valid_at");
+is( $result, qq{[1,2)|[2000-01-01,2010-01-01)|a
+[2,3)|[2000-01-01,2010-01-01)|a
+[3,4)|[2000-01-01,2010-01-01)|a
+[4,5)|[2000-01-01,2010-01-01)|a}, 'replicated temporal_no_key NOTHING');
+
+# replicate with a primary key:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_pk (id, valid_at, a)
+   VALUES ('[2,3)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[3,4)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[4,5)', '[2000-01-01,2010-01-01)', 'a')");
+
+($result, $stdout, $stderr) = $node_publisher->psql('postgres',
+	"UPDATE temporal_pk SET a = 'b' WHERE id = '[2,3)'");
+is( $stderr,
+	qq(psql:<stdin>:1: ERROR:  cannot update table "temporal_pk" because it does not have a replica identity and publishes updates
+HINT:  To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.),
+	"can't UPDATE temporal_pk NOTHING");
+
+($result, $stdout, $stderr) = $node_publisher->psql('postgres',
+	"DELETE FROM temporal_pk WHERE id = '[3,4)'");
+is( $stderr,
+	qq(psql:<stdin>:1: ERROR:  cannot delete from table "temporal_pk" because it does not have a replica identity and publishes deletes
+HINT:  To enable deleting from the table, set REPLICA IDENTITY using ALTER TABLE.),
+	"can't DELETE temporal_pk NOTHING");
+
+$node_publisher->wait_for_catchup('sub1');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_pk ORDER BY id, valid_at");
+is( $result, qq{[1,2)|[2000-01-01,2010-01-01)|a
+[2,3)|[2000-01-01,2010-01-01)|a
+[3,4)|[2000-01-01,2010-01-01)|a
+[4,5)|[2000-01-01,2010-01-01)|a}, 'replicated temporal_pk NOTHING');
+
+# replicate with a unique key:
+
+$node_publisher->safe_psql(
+	'postgres',
+	"INSERT INTO temporal_unique (id, valid_at, a)
+   VALUES ('[2,3)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[3,4)', '[2000-01-01,2010-01-01)', 'a'),
+          ('[4,5)', '[2000-01-01,2010-01-01)', 'a')");
+
+($result, $stdout, $stderr) = $node_publisher->psql('postgres',
+	"UPDATE temporal_unique SET a = 'b' WHERE id = '[2,3)'");
+is( $stderr,
+	qq(psql:<stdin>:1: ERROR:  cannot update table "temporal_unique" because it does not have a replica identity and publishes updates
+HINT:  To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.),
+	"can't UPDATE temporal_unique NOTHING");
+
+($result, $stdout, $stderr) = $node_publisher->psql('postgres',
+	"DELETE FROM temporal_unique WHERE id = '[3,4)'");
+is( $stderr,
+	qq(psql:<stdin>:1: ERROR:  cannot delete from table "temporal_unique" because it does not have a replica identity and publishes deletes
+HINT:  To enable deleting from the table, set REPLICA IDENTITY using ALTER TABLE.),
+	"can't DELETE temporal_unique NOTHING");
+
+$node_publisher->wait_for_catchup('sub1');
+
+$result = $node_subscriber->safe_psql('postgres',
+	"SELECT * FROM temporal_unique ORDER BY id, valid_at");
+is( $result, qq{[1,2)|[2000-01-01,2010-01-01)|a
+[2,3)|[2000-01-01,2010-01-01)|a
+[3,4)|[2000-01-01,2010-01-01)|a
+[4,5)|[2000-01-01,2010-01-01)|a}, 'replicated temporal_unique NOTHING');
+
+# cleanup
+
+$node_publisher->safe_psql('postgres', "DROP TABLE temporal_pk");
+$node_publisher->safe_psql('postgres', "DROP TABLE temporal_unique");
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION pub1");
+$node_subscriber->safe_psql('postgres', "DROP TABLE temporal_pk");
+$node_subscriber->safe_psql('postgres', "DROP TABLE temporal_unique");
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION sub1");
+
+
+done_testing();
+
-- 
2.42.0