Preserve commit order

Author: knizhnik

multimaster.c

@@ -207,6 +207,7 @@ int   MtmHeartbeatSendTimeout;
 int   MtmHeartbeatRecvTimeout;
 bool  MtmUseRaftable;
 bool  MtmUseDtm;
+bool  MtmPreserveCommitOrder;
 bool  MtmVolksWagenMode;
 
 TransactionId  MtmUtilityProcessedInXid;
@@ -2341,6 +2342,19 @@ _PG_init(void)
 		NULL
 	);
 
+	DefineCustomBoolVariable(
+		"multimaster.preserve_commit_order",
+		"Transactions from one node will be committed in same order al all nodes",
+		NULL,
+		&MtmPreserveCommitOrder,
+		true,
+		PGC_BACKEND,
+		0,
+		NULL,
+		NULL,
+		NULL
+	);
+
 	DefineCustomBoolVariable(
 		"multimaster.volkswagen_mode",
 		"Pretend to be normal postgres. This means skip some NOTICE's and use local sequences. Default false.",

multimaster.h

@@ -58,7 +58,7 @@
 #define MB (1024*1024L)
 
 #define USEC_TO_MSEC(t) ((t)/1000)
-#define MSEC_TO_USEC(t) ((t)*1000)
+#define MSEC_TO_USEC(t) ((timestamp_t)(t)*1000)
 
 #define Natts_mtm_ddl_log 2
 #define Anum_mtm_ddl_log_issued		1
@@ -287,6 +287,7 @@ extern int   MtmTransSpillThreshold;
 extern int   MtmHeartbeatSendTimeout;
 extern int   MtmHeartbeatRecvTimeout;
 extern bool  MtmUseDtm;
+extern bool  MtmPreserveCommitOrder;
 extern HTAB* MtmXid2State;
 extern HTAB* MtmGid2State;
 

pglogical_receiver.c

@@ -528,7 +528,12 @@ pglogical_receiver_main(Datum main_arg)
 							spill_file = -1;
 							resetStringInfo(&spill_info);
 						} else { 
-							MtmExecute(buf.data, buf.used);
+							if (MtmPreserveCommitOrder && buf.used == rc - hdr_len) {
+								/* Perform commit-prepared and rollback-prepared requested directly in receiver */
+								MtmExecutor(nodeId, buf.data, buf.used);
+							} else {
+								MtmExecute(buf.data, buf.used);
+							}
 						}
 						ByteBufferReset(&buf);
 					}