[PGPRO-4074] Use unique index on not null keys as replica identity index if it

was not specified explicitly and there is no primary key.

Cherry pick of

commit 8b0dad8fa8142fb8c8c3df9b2f6fd094f0b22a61
Author: Konstantin Knizhnik <knizhnik@garret.ru>
Date:   Wed Aug 14 14:57:44 2019 +0300

    [PGPRO-4435]

    Use unique index on not null keys as replica identity index if it was not specified explicitly and there is no primary key

    Tags: multimaster.

And

commit 7ca988b1e304f6ffabfcbf5077cd0a4f20ff09e7
Author: Andrey Lepikhov <a.lepikhov@postgrespro.ru>
Date:   Fri Nov 13 13:09:49 2020 +0500

    [PGPRO-4435] creative port of the patch f0e5b970539 from EE11.
    With changes caused by [PGPRO-4175]. Main idea of this change
    proposed by K.Knizhnik (see the comments on the jira's task.

    Change contrib/test_decoding/expected/ddl.out because of changed, by
    this patch, replication logic.

    Tags: multimaster.

tags: multimaster
(cherry picked from commit 996c9b34ee6240a95c785a87e30979bf3366c9e3)

Author: arssher

contrib/test_decoding/expected/ddl.out

@@ -648,19 +648,19 @@ SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'inc
  table public.table_with_unique_not_null: INSERT: id[integer]:2 data[integer]:2
  COMMIT
  BEGIN
- table public.table_with_unique_not_null: DELETE: id[integer]:1 data[integer]:1
+ table public.table_with_unique_not_null: DELETE: id[integer]:1
  COMMIT
  BEGIN
- table public.table_with_unique_not_null: UPDATE: old-key: id[integer]:2 data[integer]:2 new-tuple: id[integer]:2 data[integer]:3
+ table public.table_with_unique_not_null: UPDATE: id[integer]:2 data[integer]:3
  COMMIT
  BEGIN
- table public.table_with_unique_not_null: UPDATE: old-key: id[integer]:2 data[integer]:3 new-tuple: id[integer]:-2 data[integer]:3
+ table public.table_with_unique_not_null: UPDATE: old-key: id[integer]:2 new-tuple: id[integer]:-2 data[integer]:3
  COMMIT
  BEGIN
- table public.table_with_unique_not_null: UPDATE: old-key: id[integer]:-2 data[integer]:3 new-tuple: id[integer]:2 data[integer]:3
+ table public.table_with_unique_not_null: UPDATE: old-key: id[integer]:-2 new-tuple: id[integer]:2 data[integer]:3
  COMMIT
  BEGIN
- table public.table_with_unique_not_null: DELETE: id[integer]:2 data[integer]:3
+ table public.table_with_unique_not_null: DELETE: id[integer]:2
  COMMIT
  BEGIN
  table public.table_with_unique_not_null: INSERT: id[integer]:3 data[integer]:1

src/backend/utils/cache/relcache.c

@@ -4521,6 +4521,7 @@ RelationGetIndexList(Relation relation)
 	char		replident = relation->rd_rel->relreplident;
 	Oid			pkeyIndex = InvalidOid;
 	Oid			candidateIndex = InvalidOid;
+	Oid			uniqueIndex = InvalidOid;
 	MemoryContext oldcxt;
 
 	/* Quick exit if we already computed the list. */
@@ -4578,6 +4579,36 @@ RelationGetIndexList(Relation relation)
 		/* remember explicitly chosen replica index */
 		if (index->indisreplident)
 			candidateIndex = index->indexrelid;
+
+		/*
+		 * If there is no primary key and explicitly specified replica identity index,
+		 * then try to use any unique index with not-null keys as replica identity index.
+		 */
+		if (!OidIsValid(candidateIndex) &&
+			!OidIsValid(pkeyIndex) &&
+			!OidIsValid(uniqueIndex) &&
+			heap_attisnull(htup, Anum_pg_index_indexprs, NULL))
+		{
+			int i;
+			for (i = 0; i < index->indnatts; i++)
+			{
+				int	attno = index->indkey.values[i];
+				if (attno > 0)
+				{
+					Form_pg_attribute attr = TupleDescAttr(relation->rd_att, attno - 1);
+					if (!attr->attnotnull)
+						break;
+				}
+				else
+					/*
+					 * Disallow index on system column to be a replica identity.
+					 * Their values may differ on different instances.
+					 */
+					break;
+			}
+			if (i == index->indnatts)
+				uniqueIndex = index->indexrelid;
+		}
 	}
 
 	systable_endscan(indscan);
@@ -4596,6 +4627,8 @@ RelationGetIndexList(Relation relation)
 		relation->rd_replidindex = pkeyIndex;
 	else if (replident == REPLICA_IDENTITY_INDEX && OidIsValid(candidateIndex))
 		relation->rd_replidindex = candidateIndex;
+	else if (replident == REPLICA_IDENTITY_DEFAULT && OidIsValid(uniqueIndex))
+		relation->rd_replidindex = uniqueIndex;
 	else
 		relation->rd_replidindex = InvalidOid;
 	relation->rd_indexvalid = true;