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

Commitb69eae0

Browse files
knizhnikkelvich
authored andcommitted
Add loging of filtered transactions in logical decoder
1 parentbc15caf commitb69eae0

File tree

4 files changed

+94
-55
lines changed

4 files changed

+94
-55
lines changed

‎multimaster.c‎

Lines changed: 68 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,8 @@ typedef struct {
8383
boolisReplicated;/* transaction on replica */
8484
boolisDistributed;/* transaction performed INSERT/UPDATE/DELETE and has to be replicated to other nodes */
8585
boolisPrepared;/* transaction is perpared at first stage of 2PC */
86+
boolisSuspended;/* prepared transaction is suspended because coordinator node is switch to offline */
87+
boolisTransactionBlock;/* is transaction block */
8688
boolcontainsDML;/* transaction contains DML statements */
8789
XidStatusstatus;/* transaction status */
8890
csn_tsnapshot;/* transaction snaphsot */
@@ -711,7 +713,7 @@ MtmXactCallback(XactEvent event, void *arg)
711713
}
712714

713715
/*
714-
* Check if this is "normal" usertrnsaction which should be distributed to other nodes
716+
* Check if this is "normal" usertransaction which should be distributed to other nodes
715717
*/
716718
staticbool
717719
MtmIsUserTransaction()
@@ -733,6 +735,7 @@ MtmResetTransaction()
733735
x->gtid.xid=InvalidTransactionId;
734736
x->isDistributed= false;
735737
x->isPrepared= false;
738+
x->isSuspended= false;
736739
x->isTwoPhase= false;
737740
x->csn=
738741
x->status=TRANSACTION_STATUS_UNKNOWN;
@@ -762,6 +765,7 @@ MtmBeginTransaction(MtmCurrentTrans* x)
762765
x->isReplicated=MtmIsLogicalReceiver;
763766
x->isDistributed=MtmIsUserTransaction();
764767
x->isPrepared= false;
768+
x->isSuspended= false;
765769
x->isTwoPhase= false;
766770
x->isTransactionBlock=IsTransactionBlock();
767771
/* Application name can be changed usnig PGAPPNAME environment variable */
@@ -1003,14 +1007,18 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
10031007
}
10041008
if (ts->status!=TRANSACTION_STATUS_ABORTED&& !ts->votingCompleted) {
10051009
if (ts->isPrepared) {
1006-
elog(ERROR,"Commit of distributed transaction %s is suspended because node is switched to %s mode",ts->gid,MtmNodeStatusMnem[Mtm->status]);
1007-
}
1008-
if (Mtm->status!=MTM_ONLINE) {
1009-
elog(WARNING,"Commit of distributed transaction is canceled because node is switched to %s mode",MtmNodeStatusMnem[Mtm->status]);
1010-
}else {
1011-
elog(WARNING,"Commit of distributed transaction is canceled because cluster configuration was changed");
1010+
// GetNewTransactionId(false); /* force increment of transaction counter */
1011+
// elog(ERROR, "Commit of distributed transaction %s is suspended because node is switched to %s mode", ts->gid, MtmNodeStatusMnem[Mtm->status]);
1012+
elog(WARNING,"Commit of distributed transaction %s is suspended because node is switched to %s mode",ts->gid,MtmNodeStatusMnem[Mtm->status]);
1013+
x->isSuspended= true;
1014+
}else {
1015+
if (Mtm->status!=MTM_ONLINE) {
1016+
elog(WARNING,"Commit of distributed transaction is canceled because node is switched to %s mode",MtmNodeStatusMnem[Mtm->status]);
1017+
}else {
1018+
elog(WARNING,"Commit of distributed transaction is canceled because cluster configuration was changed");
1019+
}
1020+
MtmAbortTransaction(ts);
10121021
}
1013-
MtmAbortTransaction(ts);
10141022
}
10151023
x->status=ts->status;
10161024
MTM_LOG3("%d: Result of vote: %d",MyProcPid,ts->status);
@@ -1077,14 +1085,18 @@ MtmCommitPreparedTransaction(MtmCurrentTrans* x)
10771085
}
10781086
if (ts->status!=TRANSACTION_STATUS_ABORTED&& !ts->votingCompleted) {
10791087
if (ts->isPrepared) {
1080-
elog(ERROR,"Commit of distributed transaction %s is suspended because node is switched to %s mode",ts->gid,MtmNodeStatusMnem[Mtm->status]);
1081-
}
1082-
if (Mtm->status!=MTM_ONLINE) {
1083-
elog(WARNING,"Commit of distributed transaction is canceled because node is switched to %s mode",MtmNodeStatusMnem[Mtm->status]);
1084-
}else {
1085-
elog(WARNING,"Commit of distributed transaction is canceled because cluster configuration was changed");
1088+
// GetNewTransactionId(false); /* force increment of transaction counter */
1089+
// elog(ERROR, "Commit of distributed transaction %s is suspended because node is switched to %s mode", ts->gid, MtmNodeStatusMnem[Mtm->status]);
1090+
elog(WARNING,"Commit of distributed transaction %s is suspended because node is switched to %s mode",ts->gid,MtmNodeStatusMnem[Mtm->status]);
1091+
x->isSuspended= true;
1092+
}else {
1093+
if (Mtm->status!=MTM_ONLINE) {
1094+
elog(WARNING,"Commit of distributed transaction is canceled because node is switched to %s mode",MtmNodeStatusMnem[Mtm->status]);
1095+
}else {
1096+
elog(WARNING,"Commit of distributed transaction is canceled because cluster configuration was changed");
1097+
}
1098+
MtmAbortTransaction(ts);
10861099
}
1087-
MtmAbortTransaction(ts);
10881100
}
10891101
x->status=ts->status;
10901102
x->xid=ts->xid;
@@ -1292,6 +1304,7 @@ static void MtmStartRecovery()
12921304
MtmLock(LW_EXCLUSIVE);
12931305
BIT_SET(Mtm->disabledNodeMask,MtmNodeId-1);
12941306
MtmSwitchClusterMode(MTM_RECOVERY);
1307+
Mtm->recoveredLSN=InvalidXLogRecPtr;
12951308
MtmUnlock();
12961309
}
12971310

@@ -1603,6 +1616,7 @@ bool MtmRecoveryCaughtUp(int nodeId, XLogRecPtr slotLSN)
16031616
MTM_LOG1("%d: node %d is caugth-up without locking cluster",MyProcPid,nodeId);
16041617
/* We are lucky: caugth-up without locking cluster! */
16051618
}
1619+
Mtm->recoveredLSN=walLSN;
16061620
MtmEnableNode(nodeId);
16071621
Mtm->nConfigChanges+=1;
16081622
caughtUp= true;
@@ -2074,6 +2088,7 @@ static void MtmInitialize()
20742088
Mtm->walSenderLockerMask=0;
20752089
Mtm->nodeLockerMask=0;
20762090
Mtm->reconnectMask=0;
2091+
Mtm->recoveredLSN=InvalidXLogRecPtr;
20772092
Mtm->nLockers=0;
20782093
Mtm->nActiveTransactions=0;
20792094
Mtm->votingTransactions=NULL;
@@ -2101,13 +2116,14 @@ static void MtmInitialize()
21012116
Mtm->nodes[i].con=MtmConnections[i];
21022117
Mtm->nodes[i].flushPos=0;
21032118
Mtm->nodes[i].lastHeartbeat=0;
2104-
Mtm->nodes[i].restartLsn=0;
2119+
Mtm->nodes[i].restartLSN=InvalidXLogRecPtr;
21052120
Mtm->nodes[i].originId=InvalidRepOriginId;
21062121
Mtm->nodes[i].timeline=0;
2122+
Mtm->nodes[i].recoveredLSN=InvalidXLogRecPtr;
21072123
}
21082124
Mtm->nodes[MtmNodeId-1].originId=DoNotReplicateId;
21092125
/* All transaction originated from the current node should be ignored during recovery */
2110-
Mtm->nodes[MtmNodeId-1].restartLsn= (XLogRecPtr)PG_UINT64_MAX;
2126+
Mtm->nodes[MtmNodeId-1].restartLSN= (XLogRecPtr)PG_UINT64_MAX;
21112127
PGSemaphoreCreate(&Mtm->sendSemaphore);
21122128
PGSemaphoreReset(&Mtm->sendSemaphore);
21132129
SpinLockInit(&Mtm->spinlock);
@@ -2851,18 +2867,21 @@ void MtmFinishPreparedTransaction(MtmTransState* ts, bool commit)
28512867
*/
28522868
MtmReplicationModeMtmGetReplicationMode(intnodeId,sig_atomic_tvolatile*shutdown)
28532869
{
2854-
boolrecovery=false;
2870+
MtmReplicationModemode=REPLMODE_OPEN_EXISTED;
28552871

2856-
while (Mtm->status!=MTM_CONNECTED&&Mtm->status!=MTM_ONLINE)
2872+
while ((Mtm->status!=MTM_CONNECTED&&Mtm->status!=MTM_ONLINE)||BIT_CHECK(Mtm->disabledNodeMask,nodeId-1))
28572873
{
28582874
if (*shutdown)
28592875
{
28602876
returnREPLMODE_EXIT;
28612877
}
2862-
MTM_LOG2("%d: receiver slot mode %s",MyProcPid,MtmNodeStatusMnem[Mtm->status]);
28632878
MtmLock(LW_EXCLUSIVE);
2879+
if (BIT_CHECK(Mtm->disabledNodeMask,nodeId-1)) {
2880+
mode=REPLMODE_CREATE_NEW;
2881+
}
2882+
MTM_LOG2("%d: receiver slot mode %s",MyProcPid,MtmNodeStatusMnem[Mtm->status]);
28642883
if (Mtm->status==MTM_RECOVERY) {
2865-
recovery=true;
2884+
mode=REPLMODE_RECOVERED;
28662885
if ((Mtm->recoverySlot==0&& (Mtm->donorNodeId==MtmNodeId||Mtm->donorNodeId==nodeId))
28672886
||Mtm->recoverySlot==nodeId)
28682887
{
@@ -2880,13 +2899,14 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
28802899
/* delay opening of other slots until recovery is completed */
28812900
MtmSleep(STATUS_POLL_DELAY);
28822901
}
2883-
if (recovery) {
2902+
if (mode==REPLMODE_RECOVERED) {
28842903
MTM_LOG1("%d: Restart replication from node %d after end of recovery",MyProcPid,nodeId);
2904+
}elseif (mode==REPLMODE_CREATE_NEW) {
2905+
MTM_LOG1("%d: Start replication from recovered node %d",MyProcPid,nodeId);
28852906
}else {
28862907
MTM_LOG1("%d: Continue replication from node %d",MyProcPid,nodeId);
28872908
}
2888-
/* After recovery completion we need to drop all other slots to avoid receive of redundant data */
2889-
returnrecovery ?REPLMODE_RECOVERED :REPLMODE_NORMAL;
2909+
returnmode;
28902910
}
28912911

28922912
staticboolMtmIsBroadcast()
@@ -2965,7 +2985,7 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
29652985
MtmIsRecoverySession= true;
29662986
}elseif (strcmp(strVal(elem->arg),"recovered")==0) {
29672987
recoveryCompleted= true;
2968-
}elseif (strcmp(strVal(elem->arg),"normal")!=0) {
2988+
}elseif (strcmp(strVal(elem->arg),"open_existed")!=0&&strcmp(strVal(elem->arg),"create_new")!=0) {
29692989
elog(ERROR,"Illegal recovery mode %s",strVal(elem->arg));
29702990
}
29712991
}else {
@@ -2977,14 +2997,20 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
29772997
}else {
29782998
elog(ERROR,"Restart position is not specified");
29792999
}
3000+
}elseif (strcmp("mtm_recovered_pos",elem->defname)==0) {
3001+
if (elem->arg!=NULL&&strVal(elem->arg)!=NULL) {
3002+
sscanf(strVal(elem->arg),"%lx",&Mtm->nodes[MtmReplicationNodeId-1].recoveredLSN);
3003+
}else {
3004+
elog(ERROR,"Recovered position is not specified");
3005+
}
29803006
}
29813007
}
29823008
MtmLock(LW_EXCLUSIVE);
29833009
if (MtmIsRecoverySession) {
29843010
MTM_LOG1("%d: Node %d start recovery of node %d at position %lx",MyProcPid,MtmNodeId,MtmReplicationNodeId,recoveryStartPos);
29853011
Assert(MyReplicationSlot!=NULL);
29863012
if (recoveryStartPos<MyReplicationSlot->data.restart_lsn) {
2987-
elog(ERROR,"Specified recovery start position %lx is beyond restart lsn %lx",recoveryStartPos,MyReplicationSlot->data.restart_lsn);
3013+
elog(WARNING,"Specified recovery start position %lx is beyond restart lsn %lx",recoveryStartPos,MyReplicationSlot->data.restart_lsn);
29883014
}
29893015
if (!BIT_CHECK(Mtm->disabledNodeMask,MtmReplicationNodeId-1)) {
29903016
MtmDisableNode(MtmReplicationNodeId);
@@ -3133,17 +3159,17 @@ bool MtmFilterTransaction(char* record, int size)
31333159
default:
31343160
break;
31353161
}
3136-
duplicate=Mtm->status==MTM_RECOVERY&&origin_lsn!=InvalidXLogRecPtr&&origin_lsn <=Mtm->nodes[origin_node-1].restartLsn;
3162+
duplicate=Mtm->status==MTM_RECOVERY&&origin_lsn!=InvalidXLogRecPtr&&origin_lsn <=Mtm->nodes[origin_node-1].restartLSN;
31373163

31383164
MTM_LOG1("%s transaction %s from node %d lsn %lx, origin node %d, original lsn=%lx, current lsn=%lx",
3139-
duplicate ?"Ignore" :"Apply",gid,replication_node,end_lsn,origin_node,origin_lsn,Mtm->nodes[origin_node-1].restartLsn);
3165+
duplicate ?"Ignore" :"Apply",gid,replication_node,end_lsn,origin_node,origin_lsn,Mtm->nodes[origin_node-1].restartLSN);
31403166
if (Mtm->status==MTM_RECOVERY) {
3141-
if (Mtm->nodes[origin_node-1].restartLsn<origin_lsn) {
3142-
Mtm->nodes[origin_node-1].restartLsn=origin_lsn;
3167+
if (Mtm->nodes[origin_node-1].restartLSN<origin_lsn) {
3168+
Mtm->nodes[origin_node-1].restartLSN=origin_lsn;
31433169
}
31443170
}else {
3145-
if (Mtm->nodes[replication_node-1].restartLsn<end_lsn) {
3146-
Mtm->nodes[replication_node-1].restartLsn=end_lsn;
3171+
if (Mtm->nodes[replication_node-1].restartLSN<end_lsn) {
3172+
Mtm->nodes[replication_node-1].restartLSN=end_lsn;
31473173
}
31483174
}
31493175
returnduplicate;
@@ -3758,12 +3784,16 @@ static bool MtmTwoPhaseCommit(MtmCurrentTrans* x)
37583784
/* ??? Should we do explicit rollback */
37593785
}else {
37603786
CommitTransactionCommand();
3761-
StartTransactionCommand();
3762-
if (MtmGetCurrentTransactionStatus()==TRANSACTION_STATUS_ABORTED) {
3763-
FinishPreparedTransaction(x->gid, false);
3764-
elog(ERROR,"Transaction %s is aborted by DTM",x->gid);
3765-
}else {
3766-
FinishPreparedTransaction(x->gid, true);
3787+
if (x->isSuspended) {
3788+
elog(WARNING,"Transaction %s is left in prepared state because coordinator onde is not online",x->gid);
3789+
}else {
3790+
StartTransactionCommand();
3791+
if (x->status==TRANSACTION_STATUS_ABORTED) {
3792+
FinishPreparedTransaction(x->gid, false);
3793+
elog(ERROR,"Transaction %s is aborted by DTM",x->gid);
3794+
}else {
3795+
FinishPreparedTransaction(x->gid, true);
3796+
}
37673797
}
37683798
}
37693799
return true;

‎multimaster.h‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -128,10 +128,11 @@ typedef enum
128128

129129
typedefenum
130130
{
131-
REPLMODE_EXIT,/* receiver should exit */
132-
REPLMODE_RECOVERED,/* recovery of node is completed so drop old slot and restart replication from the current position in WAL */
133-
REPLMODE_RECOVERY,/* perform recorvery of the node by applying all data from the slot from specified point */
134-
REPLMODE_NORMAL/* normal mode: use existed slot or create new one and start receiving data from it from the specified position */
131+
REPLMODE_EXIT,/* receiver should exit */
132+
REPLMODE_RECOVERED,/* recovery of receiver node is completed so drop old slot and restart replication from the current position in WAL */
133+
REPLMODE_RECOVERY,/* perform recorvery of the node by applying all data from the slot from specified point */
134+
REPLMODE_CREATE_NEW,/* destination node is recovered: drop old slot and restart from roveredLsn position */
135+
REPLMODE_OPEN_EXISTED/* normal mode: use existed slot or create new one and start receiving data from it from the rememered position */
135136
}MtmReplicationMode;
136137

137138
typedefstruct
@@ -190,12 +191,13 @@ typedef struct
190191
intreceiverPid;
191192
XLogRecPtrflushPos;
192193
csn_toldestSnapshot;/* Oldest snapshot used by active transactions at this node */
193-
XLogRecPtrrestartLsn;
194+
XLogRecPtrrestartLSN;
194195
RepOriginIdoriginId;
195196
inttimeline;
196197
void*lockGraphData;
197198
intlockGraphAllocated;
198199
intlockGraphUsed;
200+
XLogRecPtrrecoveredLSN;
199201
}MtmNodeInfo;
200202

201203
typedefstructMtmTransState
@@ -266,6 +268,7 @@ typedef struct
266268
uint64gcCount;/* Number of global transactions performed since last GC */
267269
MtmMessageQueue*sendQueue;/* Messages to be sent by arbiter sender */
268270
MtmMessageQueue*freeQueue;/* Free messages */
271+
XLogRecPtrrecoveredLSN;/* LSN at the moment of recovery completion */
269272
BgwPoolpool;/* Pool of background workers for applying logical replication patches */
270273
MtmNodeInfonodes[1];/* [Mtm->nAllNodes]: per-node data */
271274
}MtmState;

‎pglogical_proto.c‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -187,7 +187,7 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
187187
Assert(false);
188188

189189
if (flags==PGLOGICAL_COMMIT||flags==PGLOGICAL_PREPARE) {
190-
Assert(txn->xid<1000||MtmTransactionRecords>=2);
190+
//Assert(txn->xid < 1000 || MtmTransactionRecords!= 1);
191191
// if (MtmIsFilteredTxn) {
192192
// Assert(MtmTransactionRecords == 0);
193193
// return;

‎pglogical_receiver.c‎

Lines changed: 17 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -196,9 +196,10 @@ feTimestampDifference(int64 start_time, int64 stop_time,
196196
staticcharconst*constMtmReplicationModeName[]=
197197
{
198198
"exit",
199-
"recovered",/* recovery of node is completed so drop old slot and restart replication from the current position in WAL */
200-
"recovery",/* perform recorvery of the node by applying all data from theslot from specified point */
201-
"normal"/* normal mode: use existed slot or create new one and start receiving data from it from the specified position */
199+
"recovered",/* recovery of node is completed so drop old slot and restart replication from the current position in WAL */
200+
"recovery",/* perform recorvery of the node by applying all data from theslot from specified point */
201+
"create_new",/* destination node is recovered: drop old slot and restart from roveredLsn position */
202+
"open_existed"/* normal mode: use existed slot or create new one and start receiving data from it from the rememered position */
202203
};
203204

204205
staticvoid
@@ -275,7 +276,7 @@ pglogical_receiver_main(Datum main_arg)
275276
}
276277
timeline=Mtm->nodes[nodeId-1].timeline;
277278
count=Mtm->recoveryCount;
278-
279+
279280
/* Establish connection to remote server */
280281
conn=PQconnectdb_safe(connString);
281282
status=PQstatus(conn);
@@ -287,7 +288,9 @@ pglogical_receiver_main(Datum main_arg)
287288
}
288289

289290
query=createPQExpBuffer();
290-
if (mode==REPLMODE_NORMAL&&timeline!=Mtm->nodes[nodeId-1].timeline) {/* recreate slot */
291+
if ((mode==REPLMODE_OPEN_EXISTED&&timeline!=Mtm->nodes[nodeId-1].timeline)
292+
||mode==REPLMODE_CREATE_NEW)
293+
{/* recreate slot */
291294
appendPQExpBuffer(query,"DROP_REPLICATION_SLOT \"%s\"",slotName);
292295
res=PQexec(conn,query->data);
293296
PQclear(res);
@@ -320,7 +323,7 @@ pglogical_receiver_main(Datum main_arg)
320323

321324
/* Start logical replication at specified position */
322325
if (mode==REPLMODE_RECOVERED) {
323-
originStartPos=Mtm->nodes[nodeId-1].restartLsn;
326+
originStartPos=Mtm->nodes[nodeId-1].restartLSN;
324327
MTM_LOG1("Restart replication from node %d from position %lx",nodeId,originStartPos);
325328
}
326329
if (originStartPos==InvalidXLogRecPtr&& !newTimeline) {
@@ -339,23 +342,26 @@ pglogical_receiver_main(Datum main_arg)
339342
MTM_LOG1("Start logical receiver at position %lx from node %d",originStartPos,nodeId);
340343
}else {
341344
originStartPos=replorigin_get_progress(originId, false);
342-
if (Mtm->nodes[nodeId-1].restartLsn<originStartPos) {
343-
Mtm->nodes[nodeId-1].restartLsn=originStartPos;
345+
if (Mtm->nodes[nodeId-1].restartLSN<originStartPos) {
346+
Mtm->nodes[nodeId-1].restartLSN=originStartPos;
344347
}
345348
MTM_LOG1("Restart logical receiver at position %lx with origin=%d from node %d",originStartPos,originId,nodeId);
346349
}
347350
Mtm->nodes[nodeId-1].originId=originId;
348351
CommitTransactionCommand();
352+
}elseif (mode==REPLMODE_CREATE_NEW) {
353+
originStartPos=Mtm->nodes[nodeId-1].recoveredLSN;
349354
}
350355

351-
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\" '%lx')",
356+
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\" '%lx', \"mtm_recovered_pos\" '%lx')",
352357
slotName,
353358
(uint32) (originStartPos >>32),
354359
(uint32)originStartPos,
355360
MULTIMASTER_MAX_PROTO_VERSION,
356361
MULTIMASTER_MIN_PROTO_VERSION,
357362
MtmReplicationModeName[mode],
358-
originStartPos
363+
originStartPos,
364+
Mtm->recoveredLSN
359365
);
360366
res=PQexec(conn,query->data);
361367
if (PQresultStatus(res)!=PGRES_COPY_BOTH)
@@ -511,7 +517,7 @@ pglogical_receiver_main(Datum main_arg)
511517
output_written_lsn=Max(walEnd,output_written_lsn);
512518
continue;
513519
}
514-
mode=REPLMODE_NORMAL;
520+
mode=REPLMODE_OPEN_EXISTED;
515521
}
516522
MTM_LOG3("%ld: Receive message %c from node %d",MtmGetSystemTime(),stmt[0],nodeId);
517523
if (buf.used >=MtmTransSpillThreshold*MB) {

0 commit comments

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