replicate create/delete tablespace

Author: kelvich

multimaster.c

@@ -164,6 +164,7 @@ MtmState* Mtm;
 VacuumStmt* MtmVacuumStmt;
 IndexStmt*  MtmIndexStmt;
 DropStmt*   MtmDropStmt;
+void*		MtmTablespaceStmt; /* CREATE/DELETE tablespace */
 MemoryContext MtmApplyContext;
 
 HTAB* MtmXid2State;
@@ -4636,7 +4637,7 @@ char* MtmGucSerialize(void)
 		appendStringInfoString(serialized_gucs, " TO ");
 
 		/* quite a crutch */
-		if (strstr(cur_entry->key, "_mem") != NULL)
+		if (strstr(cur_entry->key, "_mem") != NULL || *(cur_entry->value) == '\0')
 		{
 			appendStringInfoString(serialized_gucs, "'");
 			appendStringInfoString(serialized_gucs, cur_entry->value);
@@ -4661,20 +4662,26 @@ char* MtmGucSerialize(void)
 
 static void MtmProcessDDLCommand(char const* queryString, bool transactional)
 {
-	char* gucCtx = MtmGucSerialize();
-	if (*gucCtx) {
-		queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s; %s", gucCtx, queryString);
-	} else { 
-		queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s", queryString);
-	}
-	MTM_LOG3("Sending utility: %s", queryString);
-	if (transactional) {
+	if (MtmTx.isReplicated)
+		return;
+
+	if (transactional)
+	{
+		char *gucCtx = MtmGucSerialize();
+		if (*gucCtx)
+			queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s; %s", gucCtx, queryString);
+		else
+			queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s", queryString);
+
 		/* Transactional DDL */
+		MTM_LOG3("Sending DDL: %s", queryString);
 		LogLogicalMessage("D", queryString, strlen(queryString) + 1, true);
 		MtmTx.containsDML = true;
-	} else {	
-		MTM_LOG1("Execute concurrent DDL: %s", queryString);
+	}
+	else
+	{
 		/* Concurrent DDL */
+		MTM_LOG1("Sending concurrent DDL: %s", queryString);
 		XLogFlush(LogLogicalMessage("C", queryString, strlen(queryString) + 1, false));
 	}
 }
@@ -4748,8 +4755,6 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 		case T_ClosePortalStmt:
 		case T_FetchStmt:
 		case T_DoStmt:
-		case T_CreateTableSpaceStmt:
-		case T_AlterTableSpaceOptionsStmt:
 		case T_CommentStmt:
 		case T_PrepareStmt:
 		case T_ExecuteStmt:
@@ -4761,13 +4766,32 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
 		case T_ClusterStmt:
 		case T_VariableShowStmt:
 		case T_ReassignOwnedStmt:
-		case T_LockStmt:
+		case T_LockStmt: // XXX: check whether we should replicate that
 		case T_CheckPointStmt:
 		case T_ReindexStmt:
 		case T_ExplainStmt:
 			skipCommand = true;
 			break;
 
+		case T_CreateTableSpaceStmt:
+		case T_DropTableSpaceStmt:
+			{
+				if (MtmApplyContext != NULL)
+				{
+					MemoryContext oldContext = MemoryContextSwitchTo(MtmApplyContext);
+					Assert(oldContext != MtmApplyContext);
+					MtmTablespaceStmt = copyObject(parsetree);
+					MemoryContextSwitchTo(oldContext);
+					return;
+				}
+				else
+				{
+					skipCommand = true;
+					MtmProcessDDLCommand(queryString, false);
+				}
+			}
+			break;
+
 		case T_VacuumStmt:
 		  skipCommand = true;		  
 		  if (context == PROCESS_UTILITY_TOPLEVEL) {			  

multimaster.h

@@ -354,6 +354,7 @@ extern HTAB* MtmGid2State;
 extern VacuumStmt* MtmVacuumStmt;
 extern IndexStmt*  MtmIndexStmt;
 extern DropStmt*   MtmDropStmt;
+extern void*   MtmTablespaceStmt; /* CREATE/DELETE tablespace */
 extern MemoryContext MtmApplyContext;
 extern lsn_t MtmSenderWalEnd;
 extern timestamp_t MtmRefreshClusterStatusSchedule;

pglogical_apply.c

@@ -19,6 +19,7 @@
 
 #include "executor/spi.h"
 #include "commands/vacuum.h"
+#include "commands/tablespace.h"
 #include "commands/defrem.h"
 #include "parser/parse_utilcmd.h"
 
@@ -395,6 +396,8 @@ process_remote_message(StringInfo s)
 			MtmVacuumStmt = NULL;
 			MtmIndexStmt = NULL;
 			MtmDropStmt = NULL;
+			MtmTablespaceStmt = NULL;
+
 			rc = SPI_execute(messageBody, false, 0);
 			SPI_finish();
 			if (rc < 0) { 
@@ -423,9 +426,25 @@ process_remote_message(StringInfo s)
 								false,		/* skip_build */
 								false);		/* quiet */
 
-				} else if (MtmDropStmt != NULL) { 
+				}
+				else if (MtmDropStmt != NULL)
+				{
 					RemoveObjects(MtmDropStmt);
 				}
+				else if (MtmTablespaceStmt != NULL)
+				{
+					switch (nodeTag(MtmTablespaceStmt))
+					{
+						case T_CreateTableSpaceStmt:
+							CreateTableSpace((CreateTableSpaceStmt *) MtmTablespaceStmt);
+							break;
+						case T_DropTableSpaceStmt:
+							DropTableSpace((DropTableSpaceStmt *) MtmTablespaceStmt);
+							break;
+						default:
+							Assert(false);
+					}
+				}
 				if (ActiveSnapshotSet())
 					PopActiveSnapshot();
 			}

tests2/docker-entrypoint.sh

@@ -2,6 +2,7 @@
 
 if [ "$1" = 'postgres' ]; then
 	mkdir -p "$PGDATA"
+	mkdir -p /pg/src/src/test/regress/testtablespace
 
 	# look specifically for PG_VERSION, as it is expected in the DB dir
 	if [ ! -s "$PGDATA/PG_VERSION" ]; then