Fix replication with replica identity full

The comparison with the target rows on the subscriber side was done with
datumIsEqual(), which can have false negatives.  For instance, it didn't
work reliably for text columns.  So use the equality operator provided
by the type cache instead.

Also add more user documentation about replica identity requirements.

Reported-by: Tatsuo Ishii <ishii@sraoss.co.jp>

1 files changed, 16 insertions, 3 deletions

diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index 6dae79a8f00..59f14e997f5 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -24,12 +24,14 @@
 #include "parser/parsetree.h"
 #include "storage/bufmgr.h"
 #include "storage/lmgr.h"
+#include "utils/builtins.h"
 #include "utils/datum.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/rel.h"
 #include "utils/snapmgr.h"
 #include "utils/syscache.h"
+#include "utils/typcache.h"
 #include "utils/tqual.h"
 
 
@@ -224,13 +226,15 @@ tuple_equals_slot(TupleDesc desc, HeapTuple tup, TupleTableSlot *slot)
 	Datum		values[MaxTupleAttributeNumber];
 	bool		isnull[MaxTupleAttributeNumber];
 	int			attrnum;
-	Form_pg_attribute att;
 
 	heap_deform_tuple(tup, desc, values, isnull);
 
 	/* Check equality of the attributes. */
 	for (attrnum = 0; attrnum < desc->natts; attrnum++)
 	{
+		Form_pg_attribute att;
+		TypeCacheEntry *typentry;
+
 		/*
 		 * If one value is NULL and other is not, then they are certainly not
 		 * equal
@@ -245,8 +249,17 @@ tuple_equals_slot(TupleDesc desc, HeapTuple tup, TupleTableSlot *slot)
 			continue;
 
 		att = desc->attrs[attrnum];
-		if (!datumIsEqual(values[attrnum], slot->tts_values[attrnum],
-						  att->attbyval, att->attlen))
+
+		typentry = lookup_type_cache(att->atttypid, TYPECACHE_EQ_OPR_FINFO);
+		if (!OidIsValid(typentry->eq_opr_finfo.fn_oid))
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_FUNCTION),
+					 errmsg("could not identify an equality operator for type %s",
+							format_type_be(att->atttypid))));
+
+		if (!DatumGetBool(FunctionCall2(&typentry->eq_opr_finfo,
+										values[attrnum],
+										slot->tts_values[attrnum])))
 			return false;
 	}