🌐 AI搜索 & 代理 主页
Skip to content

Commit af8b8d1

Browse files
kelvichkelvich
committed
replicate create/delete tablespace
1 parent bc91849 commit af8b8d1

File tree

4 files changed

+60
-15
lines changed

4 files changed

+60
-15
lines changed

‎multimaster.c‎

Lines changed: 38 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -164,6 +164,7 @@ MtmState* Mtm;
164164
VacuumStmt* MtmVacuumStmt;
165165
IndexStmt* MtmIndexStmt;
166166
DropStmt* MtmDropStmt;
167+
void* MtmTablespaceStmt; /* CREATE/DELETE tablespace */
167168
MemoryContext MtmApplyContext;
168169

169170
HTAB* MtmXid2State;
@@ -4636,7 +4637,7 @@ char* MtmGucSerialize(void)
46364637
appendStringInfoString(serialized_gucs, " TO ");
46374638

46384639
/* quite a crutch */
4639-
if (strstr(cur_entry->key, "_mem") != NULL)
4640+
if (strstr(cur_entry->key, "_mem") != NULL || *(cur_entry->value) == '\0')
46404641
{
46414642
appendStringInfoString(serialized_gucs, "'");
46424643
appendStringInfoString(serialized_gucs, cur_entry->value);
@@ -4661,20 +4662,26 @@ char* MtmGucSerialize(void)
46614662

46624663
static void MtmProcessDDLCommand(char const* queryString, bool transactional)
46634664
{
4664-
char* gucCtx = MtmGucSerialize();
4665-
if (*gucCtx) {
4666-
queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s; %s", gucCtx, queryString);
4667-
} else {
4668-
queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s", queryString);
4669-
}
4670-
MTM_LOG3("Sending utility: %s", queryString);
4671-
if (transactional) {
4665+
if (MtmTx.isReplicated)
4666+
return;
4667+
4668+
if (transactional)
4669+
{
4670+
char *gucCtx = MtmGucSerialize();
4671+
if (*gucCtx)
4672+
queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s; %s", gucCtx, queryString);
4673+
else
4674+
queryString = psprintf("RESET SESSION AUTHORIZATION; reset all; %s", queryString);
4675+
46724676
/* Transactional DDL */
4677+
MTM_LOG3("Sending DDL: %s", queryString);
46734678
LogLogicalMessage("D", queryString, strlen(queryString) + 1, true);
46744679
MtmTx.containsDML = true;
4675-
} else {
4676-
MTM_LOG1("Execute concurrent DDL: %s", queryString);
4680+
}
4681+
else
4682+
{
46774683
/* Concurrent DDL */
4684+
MTM_LOG1("Sending concurrent DDL: %s", queryString);
46784685
XLogFlush(LogLogicalMessage("C", queryString, strlen(queryString) + 1, false));
46794686
}
46804687
}
@@ -4748,8 +4755,6 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
47484755
case T_ClosePortalStmt:
47494756
case T_FetchStmt:
47504757
case T_DoStmt:
4751-
case T_CreateTableSpaceStmt:
4752-
case T_AlterTableSpaceOptionsStmt:
47534758
case T_CommentStmt:
47544759
case T_PrepareStmt:
47554760
case T_ExecuteStmt:
@@ -4761,13 +4766,32 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
47614766
case T_ClusterStmt:
47624767
case T_VariableShowStmt:
47634768
case T_ReassignOwnedStmt:
4764-
case T_LockStmt:
4769+
case T_LockStmt: // XXX: check whether we should replicate that
47654770
case T_CheckPointStmt:
47664771
case T_ReindexStmt:
47674772
case T_ExplainStmt:
47684773
skipCommand = true;
47694774
break;
47704775

4776+
case T_CreateTableSpaceStmt:
4777+
case T_DropTableSpaceStmt:
4778+
{
4779+
if (MtmApplyContext != NULL)
4780+
{
4781+
MemoryContext oldContext = MemoryContextSwitchTo(MtmApplyContext);
4782+
Assert(oldContext != MtmApplyContext);
4783+
MtmTablespaceStmt = copyObject(parsetree);
4784+
MemoryContextSwitchTo(oldContext);
4785+
return;
4786+
}
4787+
else
4788+
{
4789+
skipCommand = true;
4790+
MtmProcessDDLCommand(queryString, false);
4791+
}
4792+
}
4793+
break;
4794+
47714795
case T_VacuumStmt:
47724796
skipCommand = true;
47734797
if (context == PROCESS_UTILITY_TOPLEVEL) {

‎multimaster.h‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -354,6 +354,7 @@ extern HTAB* MtmGid2State;
354354
extern VacuumStmt* MtmVacuumStmt;
355355
extern IndexStmt* MtmIndexStmt;
356356
extern DropStmt* MtmDropStmt;
357+
extern void* MtmTablespaceStmt; /* CREATE/DELETE tablespace */
357358
extern MemoryContext MtmApplyContext;
358359
extern lsn_t MtmSenderWalEnd;
359360
extern timestamp_t MtmRefreshClusterStatusSchedule;

‎pglogical_apply.c‎

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919

2020
#include "executor/spi.h"
2121
#include "commands/vacuum.h"
22+
#include "commands/tablespace.h"
2223
#include "commands/defrem.h"
2324
#include "parser/parse_utilcmd.h"
2425

@@ -395,6 +396,8 @@ process_remote_message(StringInfo s)
395396
MtmVacuumStmt = NULL;
396397
MtmIndexStmt = NULL;
397398
MtmDropStmt = NULL;
399+
MtmTablespaceStmt = NULL;
400+
398401
rc = SPI_execute(messageBody, false, 0);
399402
SPI_finish();
400403
if (rc < 0) {
@@ -423,9 +426,25 @@ process_remote_message(StringInfo s)
423426
false, /* skip_build */
424427
false); /* quiet */
425428

426-
} else if (MtmDropStmt != NULL) {
429+
}
430+
else if (MtmDropStmt != NULL)
431+
{
427432
RemoveObjects(MtmDropStmt);
428433
}
434+
else if (MtmTablespaceStmt != NULL)
435+
{
436+
switch (nodeTag(MtmTablespaceStmt))
437+
{
438+
case T_CreateTableSpaceStmt:
439+
CreateTableSpace((CreateTableSpaceStmt *) MtmTablespaceStmt);
440+
break;
441+
case T_DropTableSpaceStmt:
442+
DropTableSpace((DropTableSpaceStmt *) MtmTablespaceStmt);
443+
break;
444+
default:
445+
Assert(false);
446+
}
447+
}
429448
if (ActiveSnapshotSet())
430449
PopActiveSnapshot();
431450
}

‎tests2/docker-entrypoint.sh‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
if [ "$1" = 'postgres' ]; then
44
mkdir -p "$PGDATA"
5+
mkdir -p /pg/src/src/test/regress/testtablespace
56

67
# look specifically for PG_VERSION, as it is expected in the DB dir
78
if [ ! -s "$PGDATA/PG_VERSION" ]; then

0 commit comments

Comments
 (0)