[PGPRO-4074] Logical decoding of updates without primary keys.

Writes full tuples on update/delete instead of logging nothing when no
reasonable identity is found. Note that upcoming patch will try to use
unique index as identity if it exists.

(cherry picked from commit 33dca03a26599e392726d7874eea61b51d04d6a4)

tags: multimaster
(cherry picked from commit 5d068561e4c544d1a2b163a662b9bd166f92a56b)

Author: arssher

contrib/test_decoding/expected/ddl.out

@@ -592,16 +592,16 @@ SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'inc
  table public.table_without_key: INSERT: id[integer]:2 data[integer]:2
  COMMIT
  BEGIN
- table public.table_without_key: DELETE: (no-tuple-data)
+ table public.table_without_key: DELETE: id[integer]:1 data[integer]:1
  COMMIT
  BEGIN
- table public.table_without_key: UPDATE: id[integer]:2 data[integer]:3
+ table public.table_without_key: UPDATE: old-key: id[integer]:2 data[integer]:2 new-tuple: id[integer]:2 data[integer]:3
  COMMIT
  BEGIN
- table public.table_without_key: UPDATE: id[integer]:-2 data[integer]:3
+ table public.table_without_key: UPDATE: old-key: id[integer]:2 data[integer]:3 new-tuple: id[integer]:-2 data[integer]:3
  COMMIT
  BEGIN
- table public.table_without_key: UPDATE: id[integer]:2 data[integer]:3
+ table public.table_without_key: UPDATE: old-key: id[integer]:-2 data[integer]:3 new-tuple: id[integer]:2 data[integer]:3
  COMMIT
  BEGIN
  table public.table_without_key: UPDATE: old-key: id[integer]:2 data[integer]:3 new-tuple: id[integer]:-2 data[integer]:3
@@ -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: (no-tuple-data)
+ table public.table_with_unique_not_null: DELETE: id[integer]:1 data[integer]:1
  COMMIT
  BEGIN
- table public.table_with_unique_not_null: UPDATE: id[integer]:2 data[integer]:3
+ table public.table_with_unique_not_null: UPDATE: old-key: id[integer]:2 data[integer]:2 new-tuple: id[integer]:2 data[integer]:3
  COMMIT
  BEGIN
- table public.table_with_unique_not_null: UPDATE: id[integer]:-2 data[integer]:3
+ table public.table_with_unique_not_null: UPDATE: old-key: id[integer]:2 data[integer]:3 new-tuple: id[integer]:-2 data[integer]:3
  COMMIT
  BEGIN
- table public.table_with_unique_not_null: UPDATE: id[integer]:2 data[integer]:3
+ table public.table_with_unique_not_null: UPDATE: old-key: id[integer]:-2 data[integer]:3 new-tuple: id[integer]:2 data[integer]:3
  COMMIT
  BEGIN
- table public.table_with_unique_not_null: DELETE: (no-tuple-data)
+ table public.table_with_unique_not_null: DELETE: id[integer]:2 data[integer]:3
  COMMIT
  BEGIN
  table public.table_with_unique_not_null: INSERT: id[integer]:3 data[integer]:1

src/backend/access/heap/heapam.c

@@ -2735,7 +2735,8 @@ heap_delete(Relation relation, ItemPointer tid,
 
 		if (old_key_tuple != NULL)
 		{
-			if (relation->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
+			if (relation->rd_rel->relreplident == REPLICA_IDENTITY_FULL
+				|| !OidIsValid(relation->rd_replidindex))
 				xlrec.flags |= XLH_DELETE_CONTAINS_OLD_TUPLE;
 			else
 				xlrec.flags |= XLH_DELETE_CONTAINS_OLD_KEY;
@@ -7393,7 +7394,8 @@ log_heap_update(Relation reln, Buffer oldbuf,
 		xlrec.flags |= XLH_UPDATE_CONTAINS_NEW_TUPLE;
 		if (old_key_tuple)
 		{
-			if (reln->rd_rel->relreplident == REPLICA_IDENTITY_FULL)
+			if (reln->rd_rel->relreplident == REPLICA_IDENTITY_FULL
+				|| !OidIsValid(reln->rd_replidindex))
 				xlrec.flags |= XLH_UPDATE_CONTAINS_OLD_TUPLE;
 			else
 				xlrec.flags |= XLH_UPDATE_CONTAINS_OLD_KEY;
@@ -7620,7 +7622,10 @@ ExtractReplicaIdentity(Relation relation, HeapTuple tp, bool key_changed,
 	if (replident == REPLICA_IDENTITY_NOTHING)
 		return NULL;
 
-	if (replident == REPLICA_IDENTITY_FULL)
+	if (!relation->rd_indexvalid)
+		RelationGetIndexList(relation);
+
+	if (replident == REPLICA_IDENTITY_FULL || !OidIsValid(relation->rd_replidindex))
 	{
 		/*
 		 * When logging the entire old tuple, it very well could contain