Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commited18103

Browse files
knizhnikkelvich
authored andcommitted
Support concurrent drop
1 parentfc033ef commited18103

File tree

3 files changed

+32
-23
lines changed

3 files changed

+32
-23
lines changed

‎multimaster.c

Lines changed: 14 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -161,6 +161,7 @@ MtmState* Mtm;
161161

162162
VacuumStmt*MtmVacuumStmt;
163163
IndexStmt*MtmIndexStmt;
164+
DropStmt*MtmDropStmt;
164165
MemoryContextMtmApplyContext;
165166

166167
HTAB*MtmXid2State;
@@ -4051,7 +4052,7 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
40514052
}else {
40524053
MemoryContextoldContext=MemoryContextSwitchTo(MtmApplyContext);
40534054
Assert(oldContext!=MtmApplyContext);
4054-
MtmIndexStmt=(IndexStmt*)copyObject(parsetree);
4055+
MtmIndexStmt=indexStmt;
40554056
MemoryContextSwitchTo(oldContext);
40564057
return;
40574058
}
@@ -4062,11 +4063,19 @@ static void MtmProcessUtility(Node *parsetree, const char *queryString,
40624063
caseT_DropStmt:
40634064
{
40644065
DropStmt*stmt= (DropStmt*)parsetree;
4065-
if (stmt->removeType==OBJECT_INDEX&&stmt->concurrent&& !IsTransactionBlock()&& !MtmTx.isReplicated)
4066+
if (stmt->removeType==OBJECT_INDEX&&stmt->concurrent)
40664067
{
4067-
skipCommand= true;
4068-
MtmProcessDDLCommand(queryString, false, false);
4069-
MtmTx.isDistributed= false;
4068+
if (context==PROCESS_UTILITY_TOPLEVEL) {
4069+
MtmProcessDDLCommand(queryString, false, true);
4070+
MtmTx.isDistributed= false;
4071+
skipCommand= true;
4072+
}else {
4073+
MemoryContextoldContext=MemoryContextSwitchTo(MtmApplyContext);
4074+
Assert(oldContext!=MtmApplyContext);
4075+
MtmDropStmt=stmt;
4076+
MemoryContextSwitchTo(oldContext);
4077+
return;
4078+
}
40704079
}
40714080
}
40724081
break;

‎multimaster.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -304,6 +304,7 @@ extern HTAB* MtmXid2State;
304304
externHTAB*MtmGid2State;
305305
externVacuumStmt*MtmVacuumStmt;
306306
externIndexStmt*MtmIndexStmt;
307+
externDropStmt*MtmDropStmt;
307308
externMemoryContextMtmApplyContext;
308309

309310
externvoidMtmArbiterInitialize(void);

‎pglogical_apply.c

Lines changed: 17 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,8 @@ static void process_remote_insert(StringInfo s, Relation rel);
7979
staticvoidprocess_remote_update(StringInfos,Relationrel);
8080
staticvoidprocess_remote_delete(StringInfos,Relationrel);
8181

82+
staticMemoryContextTopContext;
83+
8284
/*
8385
* Search the index 'idxrel' for a tuple identified by 'skey' in 'rel'.
8486
*
@@ -377,34 +379,31 @@ process_remote_message(StringInfo s)
377379
case'D':
378380
{
379381
intrc;
380-
381382
MTM_LOG1("%d: Executing utility statement %s",MyProcPid,messageBody);
382383
SPI_connect();
383384
ActivePortal->sourceText=messageBody;
384385
MtmVacuumStmt=NULL;
385386
MtmIndexStmt=NULL;
387+
MtmDropStmt=NULL;
386388
rc=SPI_execute(messageBody, false,0);
387389
SPI_finish();
388390
if (rc<0) {
389391
elog(ERROR,"Failed to execute utility statement %s",messageBody);
390392
}else {
391393
if (MtmVacuumStmt!=NULL) {
392394
ExecVacuum(MtmVacuumStmt,1);
393-
}elseif (MtmIndexStmt!=NULL) {
394-
MemoryContextsaveCtx=TopTransactionContext;
395-
Oidrelid;
396-
397-
TopTransactionContext=MtmApplyContext;
398-
relid=RangeVarGetRelidExtended(MtmIndexStmt->relation,ShareUpdateExclusiveLock,
399-
false, false,
400-
NULL,
401-
NULL);
395+
}elseif (MtmIndexStmt!=NULL) {
396+
Oidrelid=RangeVarGetRelidExtended(MtmIndexStmt->relation,ShareUpdateExclusiveLock,
397+
false, false,
398+
NULL,
399+
NULL);
402400

403401
/* Run parse analysis ... */
404-
MtmIndexStmt=transformIndexStmt(relid,MtmIndexStmt,messageBody);
402+
//MtmIndexStmt = transformIndexStmt(relid, MtmIndexStmt, messageBody);
405403

406404
PushActiveSnapshot(GetTransactionSnapshot());
407405

406+
408407
DefineIndex(relid,/* OID of heap relation */
409408
MtmIndexStmt,
410409
InvalidOid,/* no predefined OID */
@@ -413,11 +412,11 @@ process_remote_message(StringInfo s)
413412
false,/* skip_build */
414413
false);/* quiet */
415414

416-
TopTransactionContext=saveCtx;
417-
418415
if (ActiveSnapshotSet())
419416
PopActiveSnapshot();
420417

418+
}elseif (MtmDropStmt!=NULL) {
419+
RemoveObjects(MtmDropStmt);
421420
}
422421
}
423422
if (standalone) {
@@ -1025,12 +1024,12 @@ void MtmExecutor(void* work, size_t size)
10251024

10261025
if (MtmApplyContext==NULL) {
10271026
MtmApplyContext=AllocSetContextCreate(TopMemoryContext,
1028-
"MessageContext",
1029-
ALLOCSET_DEFAULT_MINSIZE,
1030-
ALLOCSET_DEFAULT_INITSIZE,
1031-
ALLOCSET_DEFAULT_MAXSIZE);
1027+
"ApplyContext",
1028+
ALLOCSET_DEFAULT_MINSIZE,
1029+
ALLOCSET_DEFAULT_INITSIZE,
1030+
ALLOCSET_DEFAULT_MAXSIZE);
10321031
}
1033-
MemoryContextSwitchTo(MtmApplyContext);
1032+
TopContext=MemoryContextSwitchTo(MtmApplyContext);
10341033
replorigin_session_origin=InvalidRepOriginId;
10351034
PG_TRY();
10361035
{

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp