Handle CREATE/DROP INDEX CONCURRENTLY without 2PC

Author: kelvich

multimaster.c

@@ -148,7 +148,7 @@ static void MtmShmemStartup(void);
 static BgwPool* MtmPoolConstructor(void);
 static bool MtmRunUtilityStmt(PGconn* conn, char const* sql, char **errmsg);
 static void MtmBroadcastUtilityStmt(char const* sql, bool ignoreError);
-static bool MtmProcessDDLCommand(char const* queryString);
+static bool MtmProcessDDLCommand(char const* queryString, bool transactional);
 
 MtmState* Mtm;
 
@@ -3721,7 +3721,7 @@ static char * MtmGucSerialize(void)
  * -------------------------------------------
  */
 
-static bool MtmProcessDDLCommand(char const* queryString)
+static bool MtmProcessDDLCommand(char const* queryString, bool transactional)
 {
 	char	   *queryWithContext;
 	char	   *gucContext;
@@ -3740,7 +3740,12 @@ static bool MtmProcessDDLCommand(char const* queryString)
 	}
 
 	MTM_LOG1("Sending utility: %s", queryWithContext);
-	LogLogicalMessage("G", queryWithContext, strlen(queryWithContext)+1, true);
+	if (transactional)
+		/* DDL */
+		LogLogicalMessage("D", queryWithContext, strlen(queryWithContext) + 1, true);
+	else
+		/* CONCURRENT DDL */
+		LogLogicalMessage("C", queryWithContext, strlen(queryWithContext) + 1, false);
 
 	MtmTx.containsDML = true;
 	return false;
@@ -3886,6 +3891,30 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 			}
 			break;
 
+		case T_IndexStmt:
+			{
+				IndexStmt *indexStmt = (IndexStmt *) parsetree;
+				if (indexStmt->concurrent && !IsTransactionBlock())
+				{
+					skipCommand = true;
+					MtmProcessDDLCommand(queryString, false);
+					MtmTx.isDistributed = false;
+				}
+			}
+			break;
+
+		case T_DropStmt:
+			{
+				DropStmt *stmt = (DropStmt *) parsetree;
+				if (stmt->removeType == OBJECT_INDEX && stmt->concurrent && !IsTransactionBlock())
+				{
+					skipCommand = true;
+					MtmProcessDDLCommand(queryString, false);
+					MtmTx.isDistributed = false;
+				}
+			}
+			break;
+
 		/* Copy need some special care */
 	    case T_CopyStmt:
 		{
@@ -3923,7 +3952,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 	{
 		if (!skipCommand && !MtmTx.isReplicated && (MtmUtilityProcessedInXid == InvalidTransactionId)) {
 			MtmUtilityProcessedInXid = GetCurrentTransactionId();
-			MtmProcessDDLCommand(queryString);
+			MtmProcessDDLCommand(queryString, true);
 			executed = true;
 		}
 	}
@@ -3950,7 +3979,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 		MtmTx.snapshot = INVALID_CSN;
 	}
 
-	if (executed)
+	if (executed && !skipCommand)
 	{
 		MtmFinishDDLCommand();
 	}

pglogical_apply.c

@@ -70,7 +70,6 @@ static void UserTableUpdateOpenIndexes(EState *estate, TupleTableSlot *slot);
 static void UserTableUpdateIndexes(EState *estate, TupleTableSlot *slot);
 
 static bool process_remote_begin(StringInfo s);
-static void process_remote_transactional_message(StringInfo s);
 static void process_remote_message(StringInfo s);
 static void process_remote_commit(StringInfo s);
 static void process_remote_insert(StringInfo s, Relation rel);
@@ -355,35 +354,43 @@ process_remote_begin(StringInfo s)
 }
 
 static void
-process_remote_transactional_message(StringInfo s)
+process_remote_message(StringInfo s)
 {
-	int rc;
+	char action = pq_getmsgbyte(s);
 	int messageSize = pq_getmsgint(s, 4);
-	char const* stmt = pq_getmsgbytes(s, messageSize);
+	char const* messageBody = pq_getmsgbytes(s, messageSize);
 
-	MTM_LOG1("%d: Executing utility statement %s", MyProcPid, stmt);
-	SPI_connect();	
-	ActivePortal->sourceText = stmt;
-	rc = SPI_execute(stmt, false, 0);
-	SPI_finish();
-	if (rc < 0)
-		elog(ERROR, "Failed to execute utility statement %s", stmt);
+	switch (action)
+	{
+		case 'C':
+		{
+			MTM_LOG1("%d: Executing non-tx utility statement %s", MyProcPid, messageBody);
+			SetCurrentStatementStartTimestamp();
+			StartTransactionCommand();
+			/* intentional falldown to the next case */
+		}
+		case 'D':
+		{
+			int rc;
+
+			MTM_LOG1("%d: Executing utility statement %s", MyProcPid, messageBody);
+			SPI_connect();
+			ActivePortal->sourceText = messageBody;
+			rc = SPI_execute(messageBody, false, 0);
+			SPI_finish();
+			if (rc < 0)
+				elog(ERROR, "Failed to execute utility statement %s", messageBody);
+			break;
+		}
+		case 'L':
+		{
+			MTM_LOG3("%ld: Process deadlock message with size %d from %d", MtmGetSystemTime(), messageSize, MtmReplicationNodeId);
+			MtmUpdateLockGraph(MtmReplicationNodeId, messageBody, messageSize);
+			break;
+		}
+	}
 
-	//XXX: create messages for tables localization too.
-	// if (strcmp(relname, MULTIMASTER_LOCAL_TABLES_TABLE) == 0) { 
-	// 	char* schema = TextDatumGetCString(new_tuple.values[Anum_mtm_local_tables_rel_schema-1]);
-	// 	char* name = TextDatumGetCString(new_tuple.values[Anum_mtm_local_tables_rel_name-1]);
-	// 	MtmMakeTableLocal(schema, name);
-	// }
-}	
 
-static void
-process_remote_message(StringInfo s)
-{
-	int messageSize = pq_getmsgint(s, 4);
-	char const* messageBody = pq_getmsgbytes(s, messageSize);
-	MTM_LOG3("%ld: Process deadlock message with size %d from %d", MtmGetSystemTime(), messageSize, MtmReplicationNodeId);
-	MtmUpdateLockGraph(MtmReplicationNodeId, messageBody, messageSize);
 }
 
 static void
@@ -1049,16 +1056,10 @@ void MtmExecutor(void* work, size_t size)
 				s.len = save_len;
 				continue;
 			}
-			case 'G':
-			case 'E':
-			{
-				process_remote_transactional_message(&s);
-				continue;
-			}
-			case 'L':
+			case 'M':
 			{
 				process_remote_message(&s);
-				break;
+				continue;
 			}
             default:
                 elog(ERROR, "unknown action of type %c", action);

pglogical_proto.c

@@ -137,7 +137,7 @@ pglogical_write_message(StringInfo out,
 	{
 		MTM_LOG1("Send deadlock message to node %d", MtmReplicationNodeId);
 	}
-	else if (*prefix == 'G')
+	else if (*prefix == 'D')
 	{
 		if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN)
 		{
@@ -149,8 +149,14 @@ pglogical_write_message(StringInfo out,
 	else if (*prefix == 'E')
 	{
 		DDLInProress = false;
+		/*
+		 * we use End message only as indicator of DDL transaction finish,
+		 * so no need to send that to replicas.
+		 */
+		return;
 	}
 
+	pq_sendbyte(out, 'M');
 	pq_sendbyte(out, *prefix);
 	pq_sendint(out, sz, 4);
 	pq_sendbytes(out, message, sz);