Correctly handle toast data in find_heap_tuple

Author: knizhnik

arbiter.c

@@ -355,9 +355,23 @@ static void MtmCheckResponse(MtmArbiterMessage* resp)
 	}
 }
 
+#include <execinfo.h>
+
 static void MtmScheduleHeartbeat()
 {
-	if (!stop) { 
+	if (last_sent_heartbeat != 0 && last_sent_heartbeat + MSEC_TO_USEC(MtmHeartbeatSendTimeout)*2 < now) { 
+#define MCO_MAX_BACK_TRACE_DEPTH 256
+		int sp;
+        void* back_trace[MCO_MAX_BACK_TRACE_DEPTH];
+        int depth = backtrace(back_trace, MCO_MAX_BACK_TRACE_DEPTH);
+		char** bt = backtrace_symbols(back_trace, depth);
+		MTM_LOG1("Hearbeat interrupt: more than %lld microseconds since last heartbeat", now - last_sent_heartbeat);
+		for (sp = 0; sp < depth; sp++) {
+			MTM_LOG1("\t%s", bt[sp]);
+		}
+		free(bt);
+	}
+	if (!stop) {
 		enable_timeout_after(heartbeat_timer, MtmHeartbeatSendTimeout);
 		send_heartbeat = true;
 	}

pglogical_apply.c

@@ -119,7 +119,9 @@ bool find_heap_tuple(TupleData *tup, Relation rel, TupleTableSlot *slot, bool lo
 				continue;
 			else if (nulls[i] ^ tup->isnull[i]) /* one is null and one is not null */
 				break;
-			else if (!datumIsEqual(tup->values[i], values[i], att->attbyval, att->attlen))
+			else if (!(att->attlen == -1
+					   ? datumIsEqual(PointerGetDatum(PG_DETOAST_DATUM_PACKED(tup->values[i])), PointerGetDatum(PG_DETOAST_DATUM_PACKED(values[i])), att->attbyval, -1)
+					   : datumIsEqual(tup->values[i], values[i], att->attbyval, att->attlen)))
 				break;
 		}
 		if (i == natts)
@@ -168,6 +170,7 @@ bool find_heap_tuple(TupleData *tup, Relation rel, TupleTableSlot *slot, bool lo
 					ereport(LOG,
 							(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
 							 MTM_ERRMSG("concurrent update, retrying")));
+					heap_rescan(scan, NULL);
 					goto retry;
 				  default:
 					MTM_ELOG(ERROR, "unexpected HTSU_Result after locking: %u", res);