log ddl at destination too; use unique session_id in parallel-safe message

Author: kelvich

src/ddl.c

@@ -35,6 +35,7 @@
 #include "access/relscan.h"
 #include "commands/vacuum.h"
 #include "utils/inval.h"
+#include "replication/origin.h"
 #include "miscadmin.h"
 
 #include "mm.h"
@@ -1048,10 +1049,15 @@ MtmExecutorFinish(QueryDesc *queryDesc)
 
 
 void
-MtmApplyDDLMessage(const char *messageBody)
+MtmApplyDDLMessage(const char *messageBody, bool transactional)
 {
 	int rc;
 
+	/* Write DDL to our WAL in case smbd going to recover from us */
+	Assert(replorigin_session_origin != InvalidRepOriginId);
+	LogLogicalMessage(transactional ? "D" : "C",
+					  messageBody, strlen(messageBody) + 1, transactional);
+
 	mtm_log(DMLStmtIncoming, "%d: Executing utility statement %s",
 			MyProcPid, messageBody);
 
@@ -1083,6 +1089,8 @@ MtmApplyDDLMessage(const char *messageBody)
 		MemoryContextSwitchTo(MtmApplyContext);
 		PushActiveSnapshot(GetTransactionSnapshot());
 
+		// XXX: assert that was non-transactional ddl
+
 		switch (nodeTag(MtmCapturedDDL))
 		{
 			case T_VacuumStmt:

src/include/ddl.h

@@ -23,7 +23,7 @@ extern void MtmDDLReplicationInit(void);
 extern void MtmDDLReplicationShmemStartup(void);
 extern bool MtmIsRelationLocal(Relation rel);
 extern void MtmDDLResetStatement(void);
-extern void MtmApplyDDLMessage(const char *messageBody);
+extern void MtmApplyDDLMessage(const char *messageBody, bool transactional);
 extern void MtmDDLResetApplyState(void);
 extern void MtmSetRemoteFunction(char const* list, void* extra);
 extern void MtmToggleDML(void);

src/include/mm.h

@@ -50,6 +50,7 @@ typedef struct
 	int node_id;
 	bool is_recovery;
 	bool parallel_allowed;
+	TimestampTz session_id;
 } MtmReceiverContext;
 
 /* XXX: drop that */

src/include/multimaster.h

@@ -193,4 +193,6 @@ extern void MtmUpdateControlFile(void);
 
 extern void MtmCheckSlots(void);
 
+extern TimestampTz MtmGetIncreasingTimestamp(void);
+
 #endif

src/include/pglogical_output.h

@@ -45,7 +45,7 @@ typedef struct MtmDecoderPrivate
 {
 	int		magic; // XXX
 	bool	is_recovery;
-	int		node_id;
+	int64	session_id;
 } MtmDecoderPrivate;
 
 typedef struct PGLogicalOutputData

src/multimaster.c

@@ -46,6 +46,15 @@ typedef enum
 	MTM_STATE_LOCK_ID
 } MtmLockIds;
 
+
+typedef struct
+{
+	TimestampTz	last_timestamp;
+	slock_t		mutex;
+} MtmTime;
+
+static MtmTime *mtm_time;
+
 #define Natts_mtm_nodes_state   17
 #define Natts_mtm_cluster_state 20
 typedef struct
@@ -201,6 +210,24 @@ void MtmUnlockNode(int nodeId)
  * -------------------------------------------
  */
 
+TimestampTz
+MtmGetIncreasingTimestamp()
+{
+	TimestampTz now = GetCurrentTimestamp();
+
+	/*
+	 * Don't let time move backward; if it hasn't advanced, use incremented
+	 * last value.
+	 */
+	SpinLockAcquire(&mtm_time->mutex);
+	if (now <= mtm_time->last_timestamp)
+		now = ++mtm_time->last_timestamp;
+	else
+		mtm_time->last_timestamp = now;
+	SpinLockRelease(&mtm_time->mutex);
+
+	return now;
+}
 
 timestamp_t MtmGetSystemTime(void)
 {
@@ -524,6 +551,14 @@ static void MtmInitialize()
 	int i;
 
 	LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);
+
+	mtm_time = (MtmTime *) ShmemInitStruct("mtm_time", sizeof(MtmTime), &found);
+	if (!found)
+	{
+		mtm_time->last_timestamp = 0;
+		SpinLockInit(&mtm_time->mutex);
+	}
+
 	Mtm = (MtmState*)ShmemInitStruct(MULTIMASTER_NAME, sizeof(MtmState) + sizeof(MtmNodeInfo)*(MtmMaxNodes-1), &found);
 	if (!found)
 	{
@@ -1152,7 +1187,7 @@ _PG_init(void)
 	 * the postmaster process.)	 We'll allocate or attach to the shared
 	 * resources in mtm_shmem_startup().
 	 */
-	RequestAddinShmemSpace(MTM_SHMEM_SIZE + MtmMaxNodes*MtmQueueSize);
+	RequestAddinShmemSpace(MTM_SHMEM_SIZE + MtmMaxNodes*MtmQueueSize + sizeof(MtmTime));
 	RequestNamedLWLockTranche(MULTIMASTER_NAME, 1 + MtmMaxNodes*2 + 1);
 
 	MtmMonitorInitialize();

src/pglogical_apply.c

@@ -502,6 +502,8 @@ process_remote_message(StringInfo s, MtmReceiverContext *receiver_ctx)
 	char const* messageBody = pq_getmsgbytes(s, messageSize);
 	bool standalone = false;
 
+	MtmBeginSession(MtmReplicationNodeId);
+
 	switch (action)
 	{
 		case 'C':
@@ -510,15 +512,15 @@ process_remote_message(StringInfo s, MtmReceiverContext *receiver_ctx)
 			SetCurrentStatementStartTimestamp();
 			MtmResetTransaction();
 			StartTransactionCommand();
-			MtmApplyDDLMessage(messageBody);
+			MtmApplyDDLMessage(messageBody, false);
 			CommitTransactionCommand(); // XXX
 			standalone = true;
 			break;
 		}
 		case 'D':
 		{
 			mtm_log(MtmApplyMessage, "Executing tx DDL message %s", messageBody);
-			MtmApplyDDLMessage(messageBody);
+			MtmApplyDDLMessage(messageBody, true);
 			break;
 		}
 		case 'L':
@@ -530,19 +532,19 @@ process_remote_message(StringInfo s, MtmReceiverContext *receiver_ctx)
 		}
 		case 'P':
 		{
-			int node_id = -1;
+			int64 session_id = 0;
 
-			sscanf(messageBody, "%d", &node_id);
+			sscanf(messageBody, INT64_FORMAT, &session_id);
 
-			Assert(node_id > 0);
+			Assert(session_id > 0);
 			// XXX assert that it is receiver itself
 
-			if (!receiver_ctx->is_recovery && node_id == MtmNodeId)
+			if (receiver_ctx->session_id == session_id)
 			{
+				Assert(!receiver_ctx->is_recovery);
 				Assert(!receiver_ctx->parallel_allowed);
 				Assert(receiver_ctx->node_id > 0);
 				Assert(receiver_ctx->node_id == MtmReplicationNodeId);
-				Assert(receiver_ctx->node_id != node_id);
 
 				receiver_ctx->parallel_allowed = true;
 				MtmStateProcessNeighborEvent(MtmReplicationNodeId, MTM_NEIGHBOR_WAL_RECEIVER_START, false);
@@ -553,6 +555,9 @@ process_remote_message(StringInfo s, MtmReceiverContext *receiver_ctx)
 		default:
 			Assert(false);
  	}
+
+	MtmEndSession(MtmReplicationNodeId, false);
+
 	return standalone;
 }
 

src/pglogical_proto.c

@@ -707,7 +707,7 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 
 	hooks_data = (MtmDecoderPrivate *) palloc0(sizeof(MtmDecoderPrivate));
 	args->private_data = hooks_data;
-	hooks_data->magic = 42042042;
+	hooks_data->session_id = 0;
 
 	Mtm->nodes[MtmReplicationNodeId-1].senderPid = MyProcPid;
 
@@ -726,10 +726,8 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 				{
 					hooks_data->is_recovery = false;
 				}
-				else if (strcmp(strVal(elem->arg), "open_existed") != 0 &&
-						 strcmp(strVal(elem->arg), "create_new") != 0)
+				else
 				{
-					// XXX: don't need that anymore
 					mtm_log(ERROR, "Illegal recovery mode %s", strVal(elem->arg));
 				}
 			}
@@ -738,8 +736,28 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 				mtm_log(ERROR, "Replication mode is not specified");
 			}
 		}
+		else if (strcmp("mtm_session_id", elem->defname) == 0)
+		{
+			if (elem->arg != NULL && strVal(elem->arg) != NULL)
+			{
+				int64 session_id = 0;
+				sscanf(strVal(elem->arg), INT64_FORMAT, &session_id);
+
+				if (session_id == 0)
+					mtm_log(ERROR, "Illegal mtm_session_id");
+
+				hooks_data->session_id = session_id;
+			}
+			else
+			{
+				mtm_log(ERROR, "mtm_session_id is not specified");
+			}
+		}
 	}
 
+	if (hooks_data->session_id == 0)
+		mtm_log(ERROR, "mtm_session_id is not specified");
+
 	/*
 	 * Set proper originId mappings.
 	 *
@@ -814,11 +832,11 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 		if (!hooks_data->is_recovery)
 		{
 			XLogRecPtr msg_xptr;
-			char *dest_id = psprintf("%d", MtmReplicationNodeId);
+			char *session_id = psprintf(INT64_FORMAT, hooks_data->session_id);
 
 			LWLockAcquire(MtmCommitBarrier, LW_EXCLUSIVE);
 			MtmStateProcessNeighborEvent(MtmReplicationNodeId, MTM_NEIGHBOR_WAL_SENDER_START_RECOVERED, false);
-			msg_xptr = LogLogicalMessage("P", dest_id, strlen(dest_id) + 1, false);
+			msg_xptr = LogLogicalMessage("P", session_id, strlen(session_id) + 1, false);
 			LWLockRelease(MtmCommitBarrier);
 
 			XLogFlush(msg_xptr);

src/pglogical_receiver.c

@@ -446,7 +446,7 @@ pglogical_receiver_main(Datum main_arg)
 	PGconn *conn;
 	PGresult *res;
 	MtmReplicationMode mode;
-	MtmReceiverContext receiver_ctx = {nodeId, false, false};
+	MtmReceiverContext receiver_ctx = {nodeId, false, false, 0};
 
 	ByteBuffer buf;
 	/* Buffer for COPY data */
@@ -590,14 +590,24 @@ pglogical_receiver_main(Datum main_arg)
 			Mtm->nodes[nodeId-1].manualRecovery = false;
 		}
 
-		appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %x/%x (\"startup_params_format\" '1', \"max_proto_version\" '%d',  \"min_proto_version\" '%d', \"forward_changesets\" '1', \"mtm_replication_mode\" '%s', \"mtm_restart_pos\" '%llx')",
+		receiver_ctx.session_id = MtmGetIncreasingTimestamp();
+		receiver_ctx.is_recovery = mode == REPLMODE_RECOVERY;
+		receiver_ctx.parallel_allowed = false;
+
+		appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %x/%x ("
+									"\"startup_params_format\" '1',"
+									"\"max_proto_version\" '%d',"
+									"\"min_proto_version\" '%d',"
+									"\"forward_changesets\" '1',"
+									"\"mtm_replication_mode\" '%s',"
+									"\"mtm_session_id\" '"INT64_FORMAT"')",
 						  slotName,
 						  (uint32) (originStartPos >> 32),
 						  (uint32) originStartPos,
 						  MULTIMASTER_MAX_PROTO_VERSION,
 						  MULTIMASTER_MIN_PROTO_VERSION,
 						  MtmReplicationModeName[mode],
-						  originStartPos
+						  receiver_ctx.session_id
 			);
 		res = PQexec(conn, query->data);
 		if (PQresultStatus(res) != PGRES_COPY_BOTH)
@@ -608,9 +618,6 @@ pglogical_receiver_main(Datum main_arg)
 		PQclear(res);
 		resetPQExpBuffer(query);
 
-		receiver_ctx.is_recovery = mode == REPLMODE_RECOVERY;
-		receiver_ctx.parallel_allowed = false;
-
 		for(;;)
 		{
 			int rc, hdr_len;

tests_testgres/ddl.py

@@ -12,7 +12,7 @@ class TestDDL(unittest.TestCase):
     def setUpClass(cls):
         cls.cluster = Cluster(NUM_NODES)
         cls.cluster.print_conninfo()
-        cls.cluster.start().await_online((0,1,2))
+        cls.cluster.start().install().await_online((0,1,2))
         # cls.cluster.print_conninfo()
 
     @classmethod

tests_testgres/mm_cluster.py

@@ -6,6 +6,7 @@
 # Execute this file in order to run 3-node mm cluster
 
 import time
+import os
 
 from testgres import PostgresNode, NodeStatus, NodeBackup
 from testgres import reserve_port, release_port
@@ -135,11 +136,10 @@ def print_conninfo(self):
                                             self.username))
 
     def install(self):
-        self.node_any().poll_query_until(dbname=self.dbname,
-                                         username=self.username,
-                                         query="create extension multimaster",
-                                         raise_programming_error=False,
-                                         expected=None)
+        for node in self.nodes:
+            node.safe_psql("create extension multimaster;",
+                         dbname=self.dbname,
+                         username=self.username)
 
         return self
 
@@ -262,9 +262,12 @@ def __init__(self,
                  use_logging=False,
                  master=None):
 
+        if base_dir is None:
+            base_dir = os.path.dirname(os.path.abspath(__file__))
+
         super(ClusterNode, self).__init__(name=name,
                                           port=pg_port,
-                                          base_dir=base_dir)
+                                          base_dir=base_dir + '/../tmp_check/' + name)
 
         self.mm_port = mm_port