• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

taosdata / TDengine / #3552

11 Dec 2024 06:08AM UTC coverage: 62.526% (+0.7%) from 61.798%
#3552

push

travis-ci

web-flow
Merge pull request #29092 from taosdata/fix/3.0/TD-33146

fix:[TD-33146] stmt_get_tag_fields return error code

124833 of 255773 branches covered (48.81%)

Branch coverage included in aggregate %.

209830 of 279467 relevant lines covered (75.08%)

19111707.6 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

70.65
/source/dnode/mnode/impl/src/mndTrans.c
1
/*
2
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3
 *
4
 * This program is free software: you can use, redistribute, and/or modify
5
 * it under the terms of the GNU Affero General Public License, version 3
6
 * or later ("AGPL"), as published by the Free Software Foundation.
7
 *
8
 * This program is distributed in the hope that it will be useful, but WITHOUT
9
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10
 * FITNESS FOR A PARTICULAR PURPOSE.
11
 *
12
 * You should have received a copy of the GNU Affero General Public License
13
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
14
 */
15

16
#define _DEFAULT_SOURCE
17
#include "mndDb.h"
18
#include "mndPrivilege.h"
19
#include "mndShow.h"
20
#include "mndStb.h"
21
#include "mndSubscribe.h"
22
#include "mndSync.h"
23
#include "mndTrans.h"
24
#include "mndUser.h"
25

26
#define TRANS_VER1_NUMBER  1
27
#define TRANS_VER2_NUMBER  2
28
#define TRANS_ARRAY_SIZE   8
29
#define TRANS_RESERVE_SIZE 44
30

31
static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans);
32
static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *OldTrans, STrans *pOld);
33
static int32_t mndTransDelete(SSdb *pSdb, STrans *pTrans, bool callFunc);
34

35
static int32_t mndTransAppendLog(SArray *pArray, SSdbRaw *pRaw);
36
static int32_t mndTransAppendAction(SArray *pArray, STransAction *pAction);
37
static void    mndTransDropLogs(SArray *pArray);
38
static void    mndTransDropActions(SArray *pArray);
39

40
static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf);
41
static int32_t mndTransExecuteRedoLogs(SMnode *pMnode, STrans *pTrans, bool topHalf);
42
static int32_t mndTransExecuteUndoLogs(SMnode *pMnode, STrans *pTrans, bool topHalf);
43
static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans, bool topHalf);
44
static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans, bool topHalf);
45
static int32_t mndTransExecuteCommitActions(SMnode *pMnode, STrans *pTrans, bool topHalf);
46
static bool    mndTransPerformRedoLogStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
47
static bool    mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
48
static bool    mndTransPerformUndoLogStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
49
static bool    mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
50
static bool    mndTransPerformCommitActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
51
static bool    mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
52
static bool    mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
53
static bool    mndTransPerformFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf);
54

55
static inline bool mndTransIsInSyncContext(bool topHalf) { return !topHalf; }
553,649✔
56

57
static bool mndCannotExecuteTrans(SMnode *pMnode, bool topHalf) {
461,203✔
58
  bool isLeader = mndIsLeader(pMnode);
461,203✔
59
  bool ret = (!pMnode->deploy && !isLeader) || mndTransIsInSyncContext(topHalf);
461,203✔
60
  if (ret) mDebug("cannot execute trans action, deploy:%d, isLeader:%d, topHalf:%d", pMnode->deploy, isLeader, topHalf);
461,203✔
61
  return ret;
461,203✔
62
}
63

64
static inline char *mndStrExecutionContext(bool topHalf) { return topHalf ? "transContext" : "syncContext"; }
550,445✔
65

66
static void    mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans);
67
static int32_t mndProcessTransTimer(SRpcMsg *pReq);
68
static int32_t mndProcessTtl(SRpcMsg *pReq);
69
static int32_t mndProcessKillTransReq(SRpcMsg *pReq);
70

71
static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
72
static void    mndCancelGetNextTrans(SMnode *pMnode, void *pIter);
73

74
static int32_t tsMaxTransId = 0;
75

76
int32_t mndInitTrans(SMnode *pMnode) {
2,031✔
77
  SSdbTable table = {
2,031✔
78
      .sdbType = SDB_TRANS,
79
      .keyType = SDB_KEY_INT32,
80
      .encodeFp = (SdbEncodeFp)mndTransEncode,
81
      .decodeFp = (SdbDecodeFp)mndTransDecode,
82
      .insertFp = (SdbInsertFp)mndTransActionInsert,
83
      .updateFp = (SdbUpdateFp)mndTransActionUpdate,
84
      .deleteFp = (SdbDeleteFp)mndTransDelete,
85
  };
86

87
  mndSetMsgHandle(pMnode, TDMT_MND_TRANS_TIMER, mndProcessTransTimer);
2,031✔
88
  mndSetMsgHandle(pMnode, TDMT_MND_KILL_TRANS, mndProcessKillTransReq);
2,031✔
89

90
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TRANS, mndRetrieveTrans);
2,031✔
91
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TRANS, mndCancelGetNextTrans);
2,031✔
92
  return sdbSetTable(pMnode->pSdb, table);
2,031✔
93
}
94

95
void mndCleanupTrans(SMnode *pMnode) {}
2,029✔
96

97
static int32_t mndTransGetActionsSize(SArray *pArray) {
637,348✔
98
  int32_t actionNum = taosArrayGetSize(pArray);
637,348✔
99
  int32_t rawDataLen = 0;
637,348✔
100

101
  for (int32_t i = 0; i < actionNum; ++i) {
1,621,303✔
102
    STransAction *pAction = taosArrayGet(pArray, i);
983,955✔
103
    if (pAction->actionType == TRANS_ACTION_RAW) {
983,955✔
104
      rawDataLen += (sizeof(STransAction) + sdbGetRawTotalSize(pAction->pRaw));
454,372✔
105
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
529,583!
106
      rawDataLen += (sizeof(STransAction) + pAction->contLen);
529,583✔
107
    } else {
108
      // empty
109
    }
110
    rawDataLen += sizeof(int8_t);
983,955✔
111
  }
112

113
  return rawDataLen;
637,348✔
114
}
115

116
static int32_t mndTransEncodeAction(SSdbRaw *pRaw, int32_t *offset, SArray *pActions, int32_t actionsNum) {
637,348✔
117
  int32_t code = 0;
637,348✔
118
  int32_t lino = 0;
637,348✔
119
  int32_t dataPos = *offset;
637,348✔
120
  int8_t  unused = 0;
637,348✔
121
  int32_t ret = -1;
637,348✔
122

123
  for (int32_t i = 0; i < actionsNum; ++i) {
1,621,303✔
124
    STransAction *pAction = taosArrayGet(pActions, i);
983,955✔
125
    SDB_SET_INT32(pRaw, dataPos, pAction->id, _OVER)
983,955!
126
    SDB_SET_INT32(pRaw, dataPos, pAction->errCode, _OVER)
983,955!
127
    SDB_SET_INT32(pRaw, dataPos, pAction->acceptableCode, _OVER)
983,955!
128
    SDB_SET_INT32(pRaw, dataPos, pAction->retryCode, _OVER)
983,955!
129
    SDB_SET_INT8(pRaw, dataPos, pAction->actionType, _OVER)
983,955!
130
    SDB_SET_INT8(pRaw, dataPos, pAction->stage, _OVER)
983,955!
131
    SDB_SET_INT8(pRaw, dataPos, pAction->reserved, _OVER)
983,955!
132
    if (pAction->actionType == TRANS_ACTION_RAW) {
983,955✔
133
      int32_t len = sdbGetRawTotalSize(pAction->pRaw);
454,372✔
134
      SDB_SET_INT8(pRaw, dataPos, unused /*pAction->rawWritten*/, _OVER)
454,372!
135
      SDB_SET_INT32(pRaw, dataPos, len, _OVER)
454,372!
136
      SDB_SET_BINARY(pRaw, dataPos, (void *)pAction->pRaw, len, _OVER)
454,372!
137
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
529,583!
138
      SDB_SET_BINARY(pRaw, dataPos, (void *)&pAction->epSet, sizeof(SEpSet), _OVER)
529,583!
139
      SDB_SET_INT16(pRaw, dataPos, pAction->msgType, _OVER)
529,583!
140
      SDB_SET_INT8(pRaw, dataPos, unused /*pAction->msgSent*/, _OVER)
529,583!
141
      SDB_SET_INT8(pRaw, dataPos, unused /*pAction->msgReceived*/, _OVER)
529,583!
142
      SDB_SET_INT32(pRaw, dataPos, pAction->contLen, _OVER)
529,583!
143
      SDB_SET_BINARY(pRaw, dataPos, pAction->pCont, pAction->contLen, _OVER)
529,583!
144
    } else {
145
      // nothing
146
    }
147
  }
148
  ret = 0;
637,348✔
149

150
_OVER:
637,348✔
151
  *offset = dataPos;
637,348✔
152
  return ret;
637,348✔
153
}
154

155
SSdbRaw *mndTransEncode(STrans *pTrans) {
159,337✔
156
  int32_t code = 0;
159,337✔
157
  int32_t lino = 0;
159,337✔
158
  terrno = TSDB_CODE_INVALID_MSG;
159,337✔
159
  int8_t sver = taosArrayGetSize(pTrans->prepareActions) ? TRANS_VER2_NUMBER : TRANS_VER1_NUMBER;
159,337✔
160

161
  int32_t rawDataLen = sizeof(STrans) + TRANS_RESERVE_SIZE + pTrans->paramLen;
159,337✔
162
  rawDataLen += mndTransGetActionsSize(pTrans->prepareActions);
159,337✔
163
  rawDataLen += mndTransGetActionsSize(pTrans->redoActions);
159,337✔
164
  rawDataLen += mndTransGetActionsSize(pTrans->undoActions);
159,337✔
165
  rawDataLen += mndTransGetActionsSize(pTrans->commitActions);
159,337✔
166

167
  SSdbRaw *pRaw = sdbAllocRaw(SDB_TRANS, sver, rawDataLen);
159,337✔
168
  if (pRaw == NULL) {
159,337!
169
    mError("trans:%d, failed to alloc raw since %s", pTrans->id, terrstr());
×
170
    return NULL;
×
171
  }
172

173
  int32_t dataPos = 0;
159,337✔
174
  SDB_SET_INT32(pRaw, dataPos, pTrans->id, _OVER)
159,337!
175
  SDB_SET_INT8(pRaw, dataPos, pTrans->stage, _OVER)
159,337!
176
  SDB_SET_INT8(pRaw, dataPos, pTrans->policy, _OVER)
159,337!
177
  SDB_SET_INT8(pRaw, dataPos, pTrans->conflict, _OVER)
159,337!
178
  SDB_SET_INT8(pRaw, dataPos, pTrans->exec, _OVER)
159,337!
179
  SDB_SET_INT8(pRaw, dataPos, pTrans->oper, _OVER)
159,337!
180
  SDB_SET_INT8(pRaw, dataPos, 0, _OVER)
159,337!
181
  SDB_SET_INT16(pRaw, dataPos, pTrans->originRpcType, _OVER)
159,337!
182
  SDB_SET_INT64(pRaw, dataPos, pTrans->createdTime, _OVER)
159,337!
183
  SDB_SET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_TABLE_FNAME_LEN, _OVER)
159,337!
184
  SDB_SET_BINARY(pRaw, dataPos, pTrans->stbname, TSDB_TABLE_FNAME_LEN, _OVER)
159,337!
185
  SDB_SET_INT32(pRaw, dataPos, pTrans->actionPos, _OVER)
159,337!
186

187
  int32_t prepareActionNum = taosArrayGetSize(pTrans->prepareActions);
159,337✔
188
  int32_t redoActionNum = taosArrayGetSize(pTrans->redoActions);
159,337✔
189
  int32_t undoActionNum = taosArrayGetSize(pTrans->undoActions);
159,337✔
190
  int32_t commitActionNum = taosArrayGetSize(pTrans->commitActions);
159,337✔
191

192
  if (sver > TRANS_VER1_NUMBER) {
159,337✔
193
    SDB_SET_INT32(pRaw, dataPos, prepareActionNum, _OVER)
56,048!
194
  }
195
  SDB_SET_INT32(pRaw, dataPos, redoActionNum, _OVER)
159,337!
196
  SDB_SET_INT32(pRaw, dataPos, undoActionNum, _OVER)
159,337!
197
  SDB_SET_INT32(pRaw, dataPos, commitActionNum, _OVER)
159,337!
198

199
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->prepareActions, prepareActionNum) < 0) goto _OVER;
159,337!
200
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->redoActions, redoActionNum) < 0) goto _OVER;
159,337!
201
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->undoActions, undoActionNum) < 0) goto _OVER;
159,337!
202
  if (mndTransEncodeAction(pRaw, &dataPos, pTrans->commitActions, commitActionNum) < 0) goto _OVER;
159,337!
203

204
  SDB_SET_INT32(pRaw, dataPos, pTrans->startFunc, _OVER)
159,337!
205
  SDB_SET_INT32(pRaw, dataPos, pTrans->stopFunc, _OVER)
159,337!
206
  SDB_SET_INT32(pRaw, dataPos, pTrans->paramLen, _OVER)
159,337!
207
  if (pTrans->param != NULL) {
159,337!
208
    SDB_SET_BINARY(pRaw, dataPos, pTrans->param, pTrans->paramLen, _OVER)
×
209
  }
210

211
  SDB_SET_BINARY(pRaw, dataPos, pTrans->opername, TSDB_TRANS_OPER_LEN, _OVER)
159,337!
212

213
  int32_t arbGroupNum = taosHashGetSize(pTrans->arbGroupIds);
159,337✔
214
  SDB_SET_INT32(pRaw, dataPos, arbGroupNum, _OVER)
159,337!
215
  void *pIter = NULL;
159,337✔
216
  pIter = taosHashIterate(pTrans->arbGroupIds, NULL);
159,337✔
217
  while (pIter) {
159,355✔
218
    int32_t arbGroupId = *(int32_t *)pIter;
18✔
219
    SDB_SET_INT32(pRaw, dataPos, arbGroupId, _OVER)
18!
220
    pIter = taosHashIterate(pTrans->arbGroupIds, pIter);
18✔
221
  }
222

223
  SDB_SET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
159,337!
224
  SDB_SET_DATALEN(pRaw, dataPos, _OVER)
159,337!
225

226
  terrno = 0;
159,337✔
227

228
_OVER:
159,337✔
229
  if (terrno != 0) {
159,337!
230
    mError("trans:%d, failed to encode to raw:%p maxlen:%d len:%d since %s", pTrans->id, pRaw, sdbGetRawTotalSize(pRaw),
×
231
           dataPos, terrstr());
232
    sdbFreeRaw(pRaw);
×
233
    return NULL;
×
234
  }
235

236
  mTrace("trans:%d, encode to raw:%p, row:%p len:%d", pTrans->id, pRaw, pTrans, dataPos);
159,337✔
237
  return pRaw;
159,337✔
238
}
239

240
static int32_t mndTransDecodeAction(SSdbRaw *pRaw, int32_t *offset, SArray *pActions, int32_t actionNum) {
1,186,556✔
241
  int32_t      code = 0;
1,186,556✔
242
  int32_t      lino = 0;
1,186,556✔
243
  STransAction action = {0};
1,186,556✔
244
  int32_t      dataPos = *offset;
1,186,556✔
245
  int8_t       unused = 0;
1,186,556✔
246
  int8_t       stage = 0;
1,186,556✔
247
  int8_t       actionType = 0;
1,186,556✔
248
  int32_t      dataLen = 0;
1,186,556✔
249
  int32_t      ret = -1;
1,186,556✔
250

251
  for (int32_t i = 0; i < actionNum; ++i) {
3,033,822✔
252
    memset(&action, 0, sizeof(action));
1,847,266✔
253
    SDB_GET_INT32(pRaw, dataPos, &action.id, _OVER)
1,847,266!
254
    SDB_GET_INT32(pRaw, dataPos, &action.errCode, _OVER)
1,847,266!
255
    SDB_GET_INT32(pRaw, dataPos, &action.acceptableCode, _OVER)
1,847,266!
256
    SDB_GET_INT32(pRaw, dataPos, &action.retryCode, _OVER)
1,847,266!
257
    SDB_GET_INT8(pRaw, dataPos, &actionType, _OVER)
1,847,266!
258
    action.actionType = actionType;
1,847,266✔
259
    SDB_GET_INT8(pRaw, dataPos, &stage, _OVER)
1,847,266!
260
    action.stage = stage;
1,847,266✔
261
    SDB_GET_INT8(pRaw, dataPos, &action.reserved, _OVER)
1,847,266!
262
    if (action.actionType == TRANS_ACTION_RAW) {
1,847,266✔
263
      SDB_GET_INT8(pRaw, dataPos, &unused /*&action.rawWritten*/, _OVER)
837,850!
264
      SDB_GET_INT32(pRaw, dataPos, &dataLen, _OVER)
837,850!
265
      action.pRaw = taosMemoryMalloc(dataLen);
837,850✔
266
      if (action.pRaw == NULL) goto _OVER;
837,850!
267
      mTrace("raw:%p, is created", action.pRaw);
837,850✔
268
      SDB_GET_BINARY(pRaw, dataPos, (void *)action.pRaw, dataLen, _OVER);
837,850!
269
      if (taosArrayPush(pActions, &action) == NULL) goto _OVER;
837,850!
270
      action.pRaw = NULL;
837,850✔
271
    } else if (action.actionType == TRANS_ACTION_MSG) {
1,009,416!
272
      SDB_GET_BINARY(pRaw, dataPos, (void *)&action.epSet, sizeof(SEpSet), _OVER);
1,009,416!
273
      tmsgUpdateDnodeEpSet(&action.epSet);
1,009,416✔
274
      SDB_GET_INT16(pRaw, dataPos, &action.msgType, _OVER)
1,009,416!
275
      SDB_GET_INT8(pRaw, dataPos, &unused /*&action.msgSent*/, _OVER)
1,009,416!
276
      SDB_GET_INT8(pRaw, dataPos, &unused /*&action.msgReceived*/, _OVER)
1,009,416!
277
      SDB_GET_INT32(pRaw, dataPos, &action.contLen, _OVER)
1,009,416!
278
      action.pCont = taosMemoryMalloc(action.contLen);
1,009,416✔
279
      if (action.pCont == NULL) goto _OVER;
1,009,416!
280
      SDB_GET_BINARY(pRaw, dataPos, action.pCont, action.contLen, _OVER);
1,009,416!
281
      if (taosArrayPush(pActions, &action) == NULL) goto _OVER;
1,009,416!
282
      action.pCont = NULL;
1,009,416✔
283
    } else {
284
      if (taosArrayPush(pActions, &action) == NULL) goto _OVER;
×
285
    }
286
  }
287
  ret = 0;
1,186,556✔
288

289
_OVER:
1,186,556✔
290
  *offset = dataPos;
1,186,556✔
291
  taosMemoryFreeClear(action.pCont);
1,186,556!
292
  return ret;
1,186,556✔
293
}
294

295
SSdbRow *mndTransDecode(SSdbRaw *pRaw) {
296,639✔
296
  terrno = TSDB_CODE_INVALID_MSG;
296,639✔
297
  int32_t  code = 0;
296,639✔
298
  int32_t  lino = 0;
296,639✔
299
  SSdbRow *pRow = NULL;
296,639✔
300
  STrans  *pTrans = NULL;
296,639✔
301
  char    *pData = NULL;
296,639✔
302
  int32_t  dataLen = 0;
296,639✔
303
  int8_t   sver = 0;
296,639✔
304
  int32_t  prepareActionNum = 0;
296,639✔
305
  int32_t  redoActionNum = 0;
296,639✔
306
  int32_t  undoActionNum = 0;
296,639✔
307
  int32_t  commitActionNum = 0;
296,639✔
308
  int32_t  dataPos = 0;
296,639✔
309
  int32_t  arbgroupIdNum = 0;
296,639✔
310

311
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto _OVER;
296,639!
312

313
  if (sver != TRANS_VER1_NUMBER && sver != TRANS_VER2_NUMBER) {
296,639!
314
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
×
315
    goto _OVER;
×
316
  }
317

318
  pRow = sdbAllocRow(sizeof(STrans));
296,639✔
319
  if (pRow == NULL) goto _OVER;
296,639!
320

321
  pTrans = sdbGetRowObj(pRow);
296,639✔
322
  if (pTrans == NULL) goto _OVER;
296,639!
323

324
  SDB_GET_INT32(pRaw, dataPos, &pTrans->id, _OVER)
296,639!
325

326
  int8_t stage = 0;
296,639✔
327
  int8_t policy = 0;
296,639✔
328
  int8_t conflict = 0;
296,639✔
329
  int8_t exec = 0;
296,639✔
330
  int8_t oper = 0;
296,639✔
331
  int8_t reserved = 0;
296,639✔
332
  int8_t actionType = 0;
296,639✔
333
  SDB_GET_INT8(pRaw, dataPos, &stage, _OVER)
296,639!
334
  SDB_GET_INT8(pRaw, dataPos, &policy, _OVER)
296,639!
335
  SDB_GET_INT8(pRaw, dataPos, &conflict, _OVER)
296,639!
336
  SDB_GET_INT8(pRaw, dataPos, &exec, _OVER)
296,639!
337
  SDB_GET_INT8(pRaw, dataPos, &oper, _OVER)
296,639!
338
  SDB_GET_INT8(pRaw, dataPos, &reserved, _OVER)
296,639!
339
  pTrans->stage = stage;
296,639✔
340
  pTrans->policy = policy;
296,639✔
341
  pTrans->conflict = conflict;
296,639✔
342
  pTrans->exec = exec;
296,639✔
343
  pTrans->oper = oper;
296,639✔
344
  SDB_GET_INT16(pRaw, dataPos, &pTrans->originRpcType, _OVER)
296,639!
345
  SDB_GET_INT64(pRaw, dataPos, &pTrans->createdTime, _OVER)
296,639!
346
  SDB_GET_BINARY(pRaw, dataPos, pTrans->dbname, TSDB_TABLE_FNAME_LEN, _OVER)
296,639!
347
  SDB_GET_BINARY(pRaw, dataPos, pTrans->stbname, TSDB_TABLE_FNAME_LEN, _OVER)
296,639!
348
  SDB_GET_INT32(pRaw, dataPos, &pTrans->actionPos, _OVER)
296,639!
349

350
  if (sver > TRANS_VER1_NUMBER) {
296,639✔
351
    SDB_GET_INT32(pRaw, dataPos, &prepareActionNum, _OVER)
110,356!
352
  }
353
  SDB_GET_INT32(pRaw, dataPos, &redoActionNum, _OVER)
296,639!
354
  SDB_GET_INT32(pRaw, dataPos, &undoActionNum, _OVER)
296,639!
355
  SDB_GET_INT32(pRaw, dataPos, &commitActionNum, _OVER)
296,639!
356

357
  pTrans->prepareActions = taosArrayInit(prepareActionNum, sizeof(STransAction));
296,639✔
358
  pTrans->redoActions = taosArrayInit(redoActionNum, sizeof(STransAction));
296,639✔
359
  pTrans->undoActions = taosArrayInit(undoActionNum, sizeof(STransAction));
296,639✔
360
  pTrans->commitActions = taosArrayInit(commitActionNum, sizeof(STransAction));
296,639✔
361

362
  if (pTrans->prepareActions == NULL) goto _OVER;
296,639!
363
  if (pTrans->redoActions == NULL) goto _OVER;
296,639!
364
  if (pTrans->undoActions == NULL) goto _OVER;
296,639!
365
  if (pTrans->commitActions == NULL) goto _OVER;
296,639!
366

367
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->prepareActions, prepareActionNum) < 0) goto _OVER;
296,639!
368
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->redoActions, redoActionNum) < 0) goto _OVER;
296,639!
369
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->undoActions, undoActionNum) < 0) goto _OVER;
296,639!
370
  if (mndTransDecodeAction(pRaw, &dataPos, pTrans->commitActions, commitActionNum) < 0) goto _OVER;
296,639!
371

372
  SDB_GET_INT32(pRaw, dataPos, &pTrans->startFunc, _OVER)
296,639!
373
  SDB_GET_INT32(pRaw, dataPos, &pTrans->stopFunc, _OVER)
296,639!
374
  SDB_GET_INT32(pRaw, dataPos, &pTrans->paramLen, _OVER)
296,639!
375
  if (pTrans->paramLen != 0) {
296,639!
376
    pTrans->param = taosMemoryMalloc(pTrans->paramLen);
×
377
    if (pTrans->param == NULL) goto _OVER;
×
378
    SDB_GET_BINARY(pRaw, dataPos, pTrans->param, pTrans->paramLen, _OVER);
×
379
  }
380

381
  SDB_GET_BINARY(pRaw, dataPos, pTrans->opername, TSDB_TRANS_OPER_LEN, _OVER);
296,639!
382

383
  pTrans->arbGroupIds = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);
296,639✔
384

385
  SDB_GET_INT32(pRaw, dataPos, &arbgroupIdNum, _OVER)
296,639!
386
  for (int32_t i = 0; i < arbgroupIdNum; ++i) {
296,669✔
387
    int32_t arbGroupId = 0;
30✔
388
    SDB_GET_INT32(pRaw, dataPos, &arbGroupId, _OVER)
30!
389
    if ((terrno = taosHashPut(pTrans->arbGroupIds, &arbGroupId, sizeof(int32_t), NULL, 0)) != 0) goto _OVER;
30!
390
  }
391

392
  SDB_GET_RESERVE(pRaw, dataPos, TRANS_RESERVE_SIZE, _OVER)
296,639!
393

394
  terrno = 0;
296,639✔
395

396
_OVER:
296,639✔
397
  if (terrno != 0 && pTrans != NULL) {
296,639!
398
    mError("trans:%d, failed to parse from raw:%p since %s", pTrans->id, pRaw, terrstr());
×
399
    mndTransDropData(pTrans);
×
400
    taosMemoryFreeClear(pRow);
×
401
    return NULL;
×
402
  }
403

404
  if (pTrans != NULL) {
296,639!
405
    mTrace("trans:%d, decode from raw:%p, row:%p", pTrans->id, pRaw, pTrans);
296,639✔
406
  }
407
  return pRow;
296,639✔
408
}
409

410
static const char *mndTransStr(ETrnStage stage) {
1,921,066✔
411
  switch (stage) {
1,921,066!
412
    case TRN_STAGE_PREPARE:
132,773✔
413
      return "prepare";
132,773✔
414
    case TRN_STAGE_REDO_ACTION:
959,243✔
415
      return "redoAction";
959,243✔
416
    case TRN_STAGE_ROLLBACK:
34✔
417
      return "rollback";
34✔
418
    case TRN_STAGE_UNDO_ACTION:
304✔
419
      return "undoAction";
304✔
420
    case TRN_STAGE_COMMIT:
251,623✔
421
      return "commit";
251,623✔
422
    case TRN_STAGE_COMMIT_ACTION:
197,668✔
423
      return "commitAction";
197,668✔
424
    case TRN_STAGE_FINISH:
379,393✔
425
      return "finished";
379,393✔
426
    case TRN_STAGE_PRE_FINISH:
28✔
427
      return "pre-finish";
28✔
428
    default:
×
429
      return "invalid";
×
430
  }
431
}
432

433
static void mndSetTransLastAction(STrans *pTrans, STransAction *pAction) {
500,042✔
434
  if (pAction != NULL) {
500,042✔
435
    pTrans->lastAction = pAction->id;
368,206✔
436
    pTrans->lastMsgType = pAction->msgType;
368,206✔
437
    pTrans->lastEpset = pAction->epSet;
368,206✔
438
    pTrans->lastErrorNo = pAction->errCode;
368,206✔
439
  } else {
440
    pTrans->lastAction = 0;
131,836✔
441
    pTrans->lastMsgType = 0;
131,836✔
442
    memset(&pTrans->lastEpset, 0, sizeof(pTrans->lastEpset));
131,836✔
443
    pTrans->lastErrorNo = 0;
131,836✔
444
  }
445
}
500,042✔
446

447
static void mndTransTestStartFunc(SMnode *pMnode, void *param, int32_t paramLen) {
×
448
  mInfo("test trans start, param:%s, len:%d", (char *)param, paramLen);
×
449
}
×
450

451
static void mndTransTestStopFunc(SMnode *pMnode, void *param, int32_t paramLen) {
×
452
  mInfo("test trans stop, param:%s, len:%d", (char *)param, paramLen);
×
453
}
×
454

455
static TransCbFp mndTransGetCbFp(ETrnFunc ftype) {
2,156✔
456
  switch (ftype) {
2,156!
457
    case TRANS_START_FUNC_TEST:
×
458
      return mndTransTestStartFunc;
×
459
    case TRANS_STOP_FUNC_TEST:
×
460
      return mndTransTestStopFunc;
×
461
    case TRANS_START_FUNC_MQ_REB:
1,078✔
462
      return mndRebCntInc;
1,078✔
463
    case TRANS_STOP_FUNC_MQ_REB:
1,078✔
464
      return mndRebCntDec;
1,078✔
465
    default:
×
466
      return NULL;
×
467
  }
468
}
469

470
static int32_t mndTransActionInsert(SSdb *pSdb, STrans *pTrans) {
55,247✔
471
  mInfo("trans:%d, perform insert action, row:%p stage:%s, callfunc:1, startFunc:%d", pTrans->id, pTrans,
55,247!
472
        mndTransStr(pTrans->stage), pTrans->startFunc);
473

474
  (void)taosThreadMutexInit(&pTrans->mutex, NULL);
55,247✔
475

476
  if (pTrans->startFunc > 0) {
55,247✔
477
    TransCbFp fp = mndTransGetCbFp(pTrans->startFunc);
1,078✔
478
    if (fp) {
1,078!
479
      (*fp)(pSdb->pMnode, pTrans->param, pTrans->paramLen);
1,078✔
480
    }
481
    // pTrans->startFunc = 0;
482
  }
483

484
  if (pTrans->stage == TRN_STAGE_COMMIT) {
55,247✔
485
    pTrans->stage = TRN_STAGE_COMMIT_ACTION;
3✔
486
    mInfo("trans:%d, stage from commit to commitAction since perform update action", pTrans->id);
3!
487
  }
488

489
  if (pTrans->stage == TRN_STAGE_ROLLBACK) {
55,247✔
490
    pTrans->stage = TRN_STAGE_UNDO_ACTION;
2✔
491
    mInfo("trans:%d, stage from rollback to undoAction since perform update action", pTrans->id);
2!
492
  }
493

494
  if (pTrans->stage == TRN_STAGE_PRE_FINISH) {
55,247!
495
    pTrans->stage = TRN_STAGE_FINISH;
×
496
    mInfo("trans:%d, stage from pre-finish to finished since perform update action", pTrans->id);
×
497
  }
498

499
  return 0;
55,247✔
500
}
501

502
void mndTransDropData(STrans *pTrans) {
344,182✔
503
  if (pTrans->prepareActions != NULL) {
344,182!
504
    mndTransDropActions(pTrans->prepareActions);
344,182✔
505
    pTrans->prepareActions = NULL;
344,182✔
506
  }
507
  if (pTrans->redoActions != NULL) {
344,182!
508
    mndTransDropActions(pTrans->redoActions);
344,182✔
509
    pTrans->redoActions = NULL;
344,182✔
510
  }
511
  if (pTrans->undoActions != NULL) {
344,182!
512
    mndTransDropActions(pTrans->undoActions);
344,182✔
513
    pTrans->undoActions = NULL;
344,182✔
514
  }
515
  if (pTrans->commitActions != NULL) {
344,182!
516
    mndTransDropActions(pTrans->commitActions);
344,182✔
517
    pTrans->commitActions = NULL;
344,182✔
518
  }
519
  if (pTrans->arbGroupIds != NULL) {
344,182!
520
    taosHashCleanup(pTrans->arbGroupIds);
344,182✔
521
  }
522
  if (pTrans->pRpcArray != NULL) {
344,182✔
523
    taosArrayDestroy(pTrans->pRpcArray);
47,543✔
524
    pTrans->pRpcArray = NULL;
47,543✔
525
  }
526
  if (pTrans->rpcRsp != NULL) {
344,182✔
527
    taosMemoryFree(pTrans->rpcRsp);
15,997✔
528
    pTrans->rpcRsp = NULL;
15,997✔
529
    pTrans->rpcRspLen = 0;
15,997✔
530
  }
531
  if (pTrans->param != NULL) {
344,182!
532
    taosMemoryFree(pTrans->param);
×
533
    pTrans->param = NULL;
×
534
    pTrans->paramLen = 0;
×
535
  }
536
  (void)taosThreadMutexDestroy(&pTrans->mutex);
344,182✔
537
}
344,182✔
538

539
static int32_t mndTransDelete(SSdb *pSdb, STrans *pTrans, bool callFunc) {
175,922✔
540
  mInfo("trans:%d, perform delete action, row:%p stage:%s callfunc:%d, stopFunc:%d", pTrans->id, pTrans,
175,922!
541
        mndTransStr(pTrans->stage), callFunc, pTrans->stopFunc);
542

543
  if (pTrans->stopFunc > 0 && callFunc) {
175,922✔
544
    TransCbFp fp = mndTransGetCbFp(pTrans->stopFunc);
1,078✔
545
    if (fp) {
1,078!
546
      (*fp)(pSdb->pMnode, pTrans->param, pTrans->paramLen);
1,078✔
547
    }
548
    // pTrans->stopFunc = 0;
549
  }
550

551
  mndTransDropData(pTrans);
175,922✔
552
  return 0;
175,922✔
553
}
554

555
static void mndTransUpdateActions(SArray *pOldArray, SArray *pNewArray) {
262,540✔
556
  for (int32_t i = 0; i < taosArrayGetSize(pOldArray); ++i) {
727,312✔
557
    STransAction *pOldAction = taosArrayGet(pOldArray, i);
464,772✔
558
    STransAction *pNewAction = taosArrayGet(pNewArray, i);
464,772✔
559
    pOldAction->rawWritten = pNewAction->rawWritten;
464,772✔
560
    pOldAction->msgSent = pNewAction->msgSent;
464,772✔
561
    pOldAction->msgReceived = pNewAction->msgReceived;
464,772✔
562
    pOldAction->errCode = pNewAction->errCode;
464,772✔
563
  }
564
}
262,540✔
565

566
static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) {
65,635✔
567
  mInfo("trans:%d, perform update action, old row:%p stage:%s create:%" PRId64 ", new row:%p stage:%s create:%" PRId64,
65,635!
568
        pOld->id, pOld, mndTransStr(pOld->stage), pOld->createdTime, pNew, mndTransStr(pNew->stage), pNew->createdTime);
569

570
  if (pOld->createdTime != pNew->createdTime) {
65,635!
571
    mError("trans:%d, failed to perform update action since createTime not match, old row:%p stage:%s create:%" PRId64
×
572
           ", new row:%p stage:%s create:%" PRId64,
573
           pOld->id, pOld, mndTransStr(pOld->stage), pOld->createdTime, pNew, mndTransStr(pNew->stage),
574
           pNew->createdTime);
575
    // only occured while sync timeout
576
    TAOS_RETURN(TSDB_CODE_MND_TRANS_SYNC_TIMEOUT);
×
577
  }
578

579
  mndTransUpdateActions(pOld->prepareActions, pNew->prepareActions);
65,635✔
580
  mndTransUpdateActions(pOld->redoActions, pNew->redoActions);
65,635✔
581
  mndTransUpdateActions(pOld->undoActions, pNew->undoActions);
65,635✔
582
  mndTransUpdateActions(pOld->commitActions, pNew->commitActions);
65,635✔
583
  pOld->stage = pNew->stage;
65,635✔
584
  pOld->actionPos = pNew->actionPos;
65,635✔
585

586
  if (pOld->stage == TRN_STAGE_COMMIT) {
65,635✔
587
    pOld->stage = TRN_STAGE_COMMIT_ACTION;
55,031✔
588
    mInfo("trans:%d, stage from commit to commitAction since perform update action", pNew->id);
55,031!
589
  }
590

591
  if (pOld->stage == TRN_STAGE_ROLLBACK) {
65,635✔
592
    pOld->stage = TRN_STAGE_UNDO_ACTION;
8✔
593
    mInfo("trans:%d, stage from rollback to undoAction since perform update action", pNew->id);
8!
594
  }
595

596
  if (pOld->stage == TRN_STAGE_PRE_FINISH) {
65,635✔
597
    pOld->stage = TRN_STAGE_FINISH;
8✔
598
    mInfo("trans:%d, stage from pre-finish to finished since perform update action", pNew->id);
8!
599
  }
600

601
  return 0;
65,635✔
602
}
603

604
STrans *mndAcquireTrans(SMnode *pMnode, int32_t transId) {
283,095✔
605
  STrans *pTrans = sdbAcquire(pMnode->pSdb, SDB_TRANS, &transId);
283,095✔
606
  if (pTrans == NULL) {
283,095✔
607
    terrno = TSDB_CODE_MND_TRANS_NOT_EXIST;
4,085✔
608
  }
609
  return pTrans;
283,095✔
610
}
611

612
void mndReleaseTrans(SMnode *pMnode, STrans *pTrans) {
279,011✔
613
  SSdb *pSdb = pMnode->pSdb;
279,011✔
614
  if (pTrans != NULL) mInfo("vgId:1, trans:%d, release transaction", pTrans->id);
279,011!
615
  sdbRelease(pSdb, pTrans);
279,011✔
616
}
279,011✔
617

618
STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnConflct conflict, const SRpcMsg *pReq,
47,543✔
619
                       const char *opername) {
620
  STrans *pTrans = taosMemoryCalloc(1, sizeof(STrans));
47,543✔
621
  if (pTrans == NULL) {
47,543!
622
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
623
    mError("failed to create transaction since %s", terrstr());
×
624
    return NULL;
×
625
  }
626

627
  if (opername != NULL) {
47,543!
628
    tstrncpy(pTrans->opername, opername, TSDB_TRANS_OPER_LEN);
47,543✔
629
  }
630

631
  int32_t sdbMaxId = sdbGetMaxId(pMnode->pSdb, SDB_TRANS);
47,543✔
632
  sdbReadLock(pMnode->pSdb, SDB_TRANS);
47,543✔
633
  pTrans->id = TMAX(sdbMaxId, tsMaxTransId + 1);
47,543✔
634
  sdbUnLock(pMnode->pSdb, SDB_TRANS);
47,543✔
635
  pTrans->stage = TRN_STAGE_PREPARE;
47,543✔
636
  pTrans->policy = policy;
47,543✔
637
  pTrans->conflict = conflict;
47,543✔
638
  pTrans->exec = TRN_EXEC_PARALLEL;
47,543✔
639
  pTrans->createdTime = taosGetTimestampMs();
47,543✔
640
  pTrans->prepareActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
47,543✔
641
  pTrans->redoActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
47,543✔
642
  pTrans->undoActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
47,543✔
643
  pTrans->commitActions = taosArrayInit(TRANS_ARRAY_SIZE, sizeof(STransAction));
47,543✔
644
  pTrans->arbGroupIds = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_ENTRY_LOCK);
47,543✔
645
  pTrans->pRpcArray = taosArrayInit(1, sizeof(SRpcHandleInfo));
47,543✔
646
  pTrans->mTraceId = pReq ? TRACE_GET_ROOTID(&pReq->info.traceId) : tGenIdPI64();
47,543✔
647
  taosInitRWLatch(&pTrans->lockRpcArray);
47,543✔
648
  (void)taosThreadMutexInit(&pTrans->mutex, NULL);
47,543✔
649

650
  if (pTrans->redoActions == NULL || pTrans->undoActions == NULL || pTrans->commitActions == NULL ||
47,543!
651
      pTrans->pRpcArray == NULL) {
47,543!
652
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
653
    mError("failed to create transaction since %s", terrstr());
×
654
    mndTransDrop(pTrans);
×
655
    return NULL;
×
656
  }
657

658
  if (pReq != NULL) {
47,543✔
659
    if (taosArrayPush(pTrans->pRpcArray, &pReq->info) == NULL) {
68,796!
660
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
661
      return NULL;
×
662
    }
663
    pTrans->originRpcType = pReq->msgType;
34,398✔
664
  }
665

666
  mInfo("trans:%d, create transaction:%s, origin:%s", pTrans->id, pTrans->opername, opername);
47,543!
667

668
  mTrace("trans:%d, local object is created, data:%p", pTrans->id, pTrans);
47,543✔
669
  return pTrans;
47,543✔
670
}
671

672
static void mndTransDropActions(SArray *pArray) {
1,376,728✔
673
  int32_t size = taosArrayGetSize(pArray);
1,376,728✔
674
  for (int32_t i = 0; i < size; ++i) {
3,488,734✔
675
    STransAction *pAction = taosArrayGet(pArray, i);
2,112,006✔
676
    if (pAction->actionType == TRANS_ACTION_RAW) {
2,112,006✔
677
      taosMemoryFreeClear(pAction->pRaw);
966,454!
678
    } else if (pAction->actionType == TRANS_ACTION_MSG) {
1,145,552!
679
      taosMemoryFreeClear(pAction->pCont);
1,145,552!
680
    } else {
681
      // nothing
682
    }
683
  }
684

685
  taosArrayDestroy(pArray);
1,376,728✔
686
}
1,376,728✔
687

688
void mndTransDrop(STrans *pTrans) {
47,931✔
689
  if (pTrans != NULL) {
47,931✔
690
    mndTransDropData(pTrans);
47,543✔
691
    mTrace("trans:%d, local object is freed, data:%p", pTrans->id, pTrans);
47,543✔
692
    taosMemoryFreeClear(pTrans);
47,543!
693
  }
694
}
47,931✔
695

696
static int32_t mndTransAppendAction(SArray *pArray, STransAction *pAction) {
264,740✔
697
  pAction->id = taosArrayGetSize(pArray);
264,740✔
698

699
  void *ptr = taosArrayPush(pArray, pAction);
264,740✔
700
  if (ptr == NULL) {
264,740!
701
    TAOS_RETURN(terrno);
×
702
  }
703

704
  return 0;
264,740✔
705
}
706

707
int32_t mndTransAppendRedolog(STrans *pTrans, SSdbRaw *pRaw) {
2,162✔
708
  STransAction action = {
2,162✔
709
      .stage = TRN_STAGE_REDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw, .mTraceId = pTrans->mTraceId};
2,162✔
710
  return mndTransAppendAction(pTrans->redoActions, &action);
2,162✔
711
}
712

713
int32_t mndTransAppendNullLog(STrans *pTrans) {
×
714
  STransAction action = {.stage = TRN_STAGE_REDO_ACTION, .actionType = TRANS_ACTION_NULL};
×
715
  return mndTransAppendAction(pTrans->redoActions, &action);
×
716
}
717

718
int32_t mndTransAppendUndolog(STrans *pTrans, SSdbRaw *pRaw) {
15,736✔
719
  STransAction action = {.stage = TRN_STAGE_UNDO_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw};
15,736✔
720
  return mndTransAppendAction(pTrans->undoActions, &action);
15,736✔
721
}
722

723
int32_t mndTransAppendCommitlog(STrans *pTrans, SSdbRaw *pRaw) {
85,303✔
724
  STransAction action = {.stage = TRN_STAGE_COMMIT_ACTION, .actionType = TRANS_ACTION_RAW, .pRaw = pRaw};
85,303✔
725
  return mndTransAppendAction(pTrans->commitActions, &action);
85,303✔
726
}
727

728
int32_t mndTransAppendPrepareLog(STrans *pTrans, SSdbRaw *pRaw) {
25,403✔
729
  STransAction action = {
25,403✔
730
      .pRaw = pRaw, .stage = TRN_STAGE_PREPARE, .actionType = TRANS_ACTION_RAW, .mTraceId = pTrans->mTraceId};
25,403✔
731
  return mndTransAppendAction(pTrans->prepareActions, &action);
25,403✔
732
}
733

734
int32_t mndTransAppendRedoAction(STrans *pTrans, STransAction *pAction) {
98,277✔
735
  pAction->stage = TRN_STAGE_REDO_ACTION;
98,277✔
736
  pAction->actionType = TRANS_ACTION_MSG;
98,277✔
737
  pAction->mTraceId = pTrans->mTraceId;
98,277✔
738
  return mndTransAppendAction(pTrans->redoActions, pAction);
98,277✔
739
}
740

741
int32_t mndTransAppendUndoAction(STrans *pTrans, STransAction *pAction) {
37,859✔
742
  pAction->stage = TRN_STAGE_UNDO_ACTION;
37,859✔
743
  pAction->actionType = TRANS_ACTION_MSG;
37,859✔
744
  return mndTransAppendAction(pTrans->undoActions, pAction);
37,859✔
745
}
746

747
void mndTransSetRpcRsp(STrans *pTrans, void *pCont, int32_t contLen) {
15,997✔
748
  pTrans->rpcRsp = pCont;
15,997✔
749
  pTrans->rpcRspLen = contLen;
15,997✔
750
}
15,997✔
751

752
void mndTransSetCb(STrans *pTrans, ETrnFunc startFunc, ETrnFunc stopFunc, void *param, int32_t paramLen) {
1,068✔
753
  pTrans->startFunc = startFunc;
1,068✔
754
  pTrans->stopFunc = stopFunc;
1,068✔
755
  pTrans->param = param;
1,068✔
756
  pTrans->paramLen = paramLen;
1,068✔
757
}
1,068✔
758

759
int32_t mndSetRpcInfoForDbTrans(SMnode *pMnode, SRpcMsg *pMsg, EOperType oper, const char *dbname) {
×
760
  STrans *pTrans = NULL;
×
761
  void   *pIter = NULL;
×
762
  int32_t code = -1;
×
763

764
  while (1) {
765
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
×
766
    if (pIter == NULL) break;
×
767

768
    if (pTrans->oper == oper) {
×
769
      if (strcasecmp(dbname, pTrans->dbname) == 0) {
×
770
        mInfo("trans:%d, db:%s oper:%d matched with input", pTrans->id, dbname, oper);
×
771
        taosWLockLatch(&pTrans->lockRpcArray);
×
772
        if (pTrans->pRpcArray == NULL) {
×
773
          pTrans->pRpcArray = taosArrayInit(4, sizeof(SRpcHandleInfo));
×
774
        }
775
        if (pTrans->pRpcArray != NULL && taosArrayPush(pTrans->pRpcArray, &pMsg->info) != NULL) {
×
776
          code = 0;
×
777
        }
778
        taosWUnLockLatch(&pTrans->lockRpcArray);
×
779

780
        sdbRelease(pMnode->pSdb, pTrans);
×
781
        break;
×
782
      }
783
    }
784

785
    sdbRelease(pMnode->pSdb, pTrans);
×
786
  }
787
  return code;
×
788
}
789

790
void mndTransSetDbName(STrans *pTrans, const char *dbname, const char *stbname) {
34,475✔
791
  if (dbname != NULL) {
34,475!
792
    tstrncpy(pTrans->dbname, dbname, TSDB_DB_FNAME_LEN);
34,475✔
793
  }
794
  if (stbname != NULL) {
34,475✔
795
    tstrncpy(pTrans->stbname, stbname, TSDB_TABLE_FNAME_LEN);
25,673✔
796
  }
797
}
34,475✔
798

799
void mndTransAddArbGroupId(STrans *pTrans, int32_t groupId) {
6✔
800
  if (taosHashPut(pTrans->arbGroupIds, &groupId, sizeof(int32_t), NULL, 0) != 0) {
6!
801
    mError("trans:%d, failed to put groupid into hash, groupId:%d", pTrans->id, groupId);
×
802
  }
803
}
6✔
804

805
void mndTransSetSerial(STrans *pTrans) { pTrans->exec = TRN_EXEC_SERIAL; }
3,024✔
806

807
void mndTransSetParallel(STrans *pTrans) { pTrans->exec = TRN_EXEC_PARALLEL; }
×
808

809
void mndTransSetChangeless(STrans *pTrans) { pTrans->changeless = true; }
68✔
810

811
void mndTransSetOper(STrans *pTrans, EOperType oper) { pTrans->oper = oper; }
4,649✔
812

813
static int32_t mndTransSync(SMnode *pMnode, STrans *pTrans) {
103,993✔
814
  int32_t  code = 0;
103,993✔
815
  SSdbRaw *pRaw = mndTransEncode(pTrans);
103,993✔
816
  if (pRaw == NULL) {
103,993!
817
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
818
    if (terrno != 0) code = terrno;
×
819
    mError("trans:%d, failed to encode while sync trans since %s", pTrans->id, tstrerror(code));
×
820
    TAOS_RETURN(code);
×
821
  }
822
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRaw, SDB_STATUS_READY));
103,993!
823

824
  mInfo("trans:%d, sync to other mnodes, stage:%s createTime:%" PRId64, pTrans->id, mndTransStr(pTrans->stage),
103,993!
825
        pTrans->createdTime);
826
  code = mndSyncPropose(pMnode, pRaw, pTrans->id);
103,993✔
827
  if (code != 0) {
103,993✔
828
    mError("trans:%d, failed to sync, errno:%s code:0x%x createTime:%" PRId64 " saved trans:%d", pTrans->id,
25!
829
           tstrerror(code), code, pTrans->createdTime, pMnode->syncMgmt.transId);
830
    sdbFreeRaw(pRaw);
25✔
831
    TAOS_RETURN(code);
25✔
832
  }
833

834
  sdbFreeRaw(pRaw);
103,968✔
835
  mInfo("trans:%d, sync finished, createTime:%" PRId64, pTrans->id, pTrans->createdTime);
103,968!
836
  TAOS_RETURN(code);
103,968✔
837
}
838

839
static bool mndCheckDbConflict(const char *conflict, STrans *pTrans) {
4,421✔
840
  if (conflict[0] == 0) return false;
4,421!
841
  if (strcasecmp(conflict, pTrans->dbname) == 0) return true;
4,421✔
842
  return false;
4,224✔
843
}
844

845
static bool mndCheckStbConflict(const char *conflict, STrans *pTrans) {
21,754✔
846
  if (conflict[0] == 0) return false;
21,754✔
847
  if (strcasecmp(conflict, pTrans->stbname) == 0) return true;
20,342✔
848
  return false;
20,277✔
849
}
850

851
static void mndTransLogConflict(STrans *pNew, STrans *pTrans, bool conflict, bool *globalConflict) {
26,175✔
852
  if (conflict) {
26,175✔
853
    mError("trans:%d, db:%s stb:%s type:%d, can't execute since conflict with trans:%d db:%s stb:%s type:%d", pNew->id,
262!
854
           pNew->dbname, pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
855
    *globalConflict = true;
262✔
856
  } else {
857
    mInfo("trans:%d, db:%s stb:%s type:%d, not conflict with trans:%d db:%s stb:%s type:%d", pNew->id, pNew->dbname,
25,913!
858
          pNew->stbname, pNew->conflict, pTrans->id, pTrans->dbname, pTrans->stbname, pTrans->conflict);
859
  }
860
}
26,175✔
861

862
static bool mndCheckTransConflict(SMnode *pMnode, STrans *pNew) {
137,296✔
863
  STrans *pTrans = NULL;
137,296✔
864
  void   *pIter = NULL;
137,296✔
865
  bool    conflict = false;
137,296✔
866

867
  if (pNew->conflict == TRN_CONFLICT_NOTHING) return conflict;
137,296✔
868

869
  int32_t size = sdbGetSize(pMnode->pSdb, SDB_TRANS);
101,380✔
870
  mInfo("trans:%d, trans hash size %d", pNew->id, size);
101,380!
871

872
  while (1) {
873
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
125,045✔
874
    if (pIter == NULL) break;
125,045✔
875

876
    if (pNew->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
23,665✔
877

878
    if (pNew->conflict == TRN_CONFLICT_DB) {
23,665✔
879
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
3,818✔
880
      if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
3,818✔
881
        mndTransLogConflict(pNew, pTrans, mndCheckDbConflict(pNew->dbname, pTrans), &conflict);
2,660✔
882
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);
2,660✔
883
      }
884
    }
885

886
    if (pNew->conflict == TRN_CONFLICT_DB_INSIDE) {
23,665✔
887
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
19,837✔
888
      if (pTrans->conflict == TRN_CONFLICT_DB) {
19,837✔
889
        mndTransLogConflict(pNew, pTrans, mndCheckDbConflict(pNew->dbname, pTrans), &conflict);
1,761✔
890
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);
1,761✔
891
      }
892
      if (pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
19,837✔
893
        mndTransLogConflict(pNew, pTrans, mndCheckStbConflict(pNew->stbname, pTrans), &conflict);  // for stb
17,333✔
894
      }
895
    }
896

897
    if (pNew->conflict == TRN_CONFLICT_ARBGROUP) {
23,665!
898
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL) conflict = true;
×
899
      if (pTrans->conflict == TRN_CONFLICT_ARBGROUP) {
×
900
        void *pGidIter = taosHashIterate(pNew->arbGroupIds, NULL);
×
901
        while (pGidIter != NULL) {
×
902
          int32_t groupId = *(int32_t *)pGidIter;
×
903
          if (taosHashGet(pTrans->arbGroupIds, &groupId, sizeof(int32_t)) != NULL) {
×
904
            taosHashCancelIterate(pNew->arbGroupIds, pGidIter);
×
905
            mndTransLogConflict(pNew, pTrans, true, &conflict);
×
906
            break;
×
907
          } else {
908
            mndTransLogConflict(pNew, pTrans, false, &conflict);
×
909
          }
910
          pGidIter = taosHashIterate(pNew->arbGroupIds, pGidIter);
×
911
        }
912
      }
913
    }
914

915
    if (pNew->conflict == TRN_CONFLICT_TSMA) {
23,665!
916
      if (pTrans->conflict == TRN_CONFLICT_GLOBAL || pTrans->conflict == TRN_CONFLICT_TSMA) {
×
917
        mndTransLogConflict(pNew, pTrans, true, &conflict);
×
918
      } else {
919
        mndTransLogConflict(pNew, pTrans, false, &conflict);
×
920
      }
921
    }
922

923
    sdbRelease(pMnode->pSdb, pTrans);
23,665✔
924
  }
925

926
  return conflict;
101,380✔
927
}
928

929
int32_t mndTransCheckConflict(SMnode *pMnode, STrans *pTrans) {
137,296✔
930
  int32_t code = 0;
137,296✔
931
  if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
137,296✔
932
    if (strlen(pTrans->dbname) == 0 && strlen(pTrans->stbname) == 0) {
90,934!
933
      code = TSDB_CODE_MND_TRANS_CONFLICT;
×
934
      mError("trans:%d, failed to check tran conflict since db not set", pTrans->id);
×
935
      TAOS_RETURN(code);
×
936
    }
937
  }
938

939
  if (mndCheckTransConflict(pMnode, pTrans)) {
137,296✔
940
    code = TSDB_CODE_MND_TRANS_CONFLICT;
302✔
941
    mError("trans:%d, failed to check tran conflict since %s", pTrans->id, tstrerror(code));
302!
942
    TAOS_RETURN(code);
302✔
943
  }
944

945
  TAOS_RETURN(code);
136,994✔
946
}
947

948
int32_t mndTransCheckConflictWithCompact(SMnode *pMnode, STrans *pTrans) {
416✔
949
  int32_t      code = 0;
416✔
950
  void        *pIter = NULL;
416✔
951
  bool         conflict = false;
416✔
952
  SCompactObj *pCompact = NULL;
416✔
953

954
  while (1) {
4✔
955
    bool thisConflict = false;
420✔
956
    pIter = sdbFetch(pMnode->pSdb, SDB_COMPACT, pIter, (void **)&pCompact);
420✔
957
    if (pIter == NULL) break;
420✔
958

959
    if (pTrans->conflict == TRN_CONFLICT_GLOBAL) {
4✔
960
      thisConflict = true;
2✔
961
    }
962
    if (pTrans->conflict == TRN_CONFLICT_DB || pTrans->conflict == TRN_CONFLICT_DB_INSIDE) {
4!
963
      if (strcasecmp(pTrans->dbname, pCompact->dbname) == 0) thisConflict = true;
2!
964
    }
965

966
    if (thisConflict) {
4!
967
      mError("trans:%d, db:%s stb:%s type:%d, can't execute since conflict with compact:%d db:%s", pTrans->id,
4!
968
             pTrans->dbname, pTrans->stbname, pTrans->conflict, pCompact->compactId, pCompact->dbname);
969
      conflict = true;
4✔
970
    } else {
971
      mInfo("trans:%d, db:%s stb:%s type:%d, not conflict with compact:%d db:%s", pTrans->id, pTrans->dbname,
×
972
            pTrans->stbname, pTrans->conflict, pCompact->compactId, pCompact->dbname);
973
    }
974
    sdbRelease(pMnode->pSdb, pCompact);
4✔
975
  }
976

977
  if (conflict) {
416✔
978
    code = TSDB_CODE_MND_TRANS_CONFLICT_COMPACT;
4✔
979
    mError("trans:%d, failed to check tran conflict with compact since %s", pTrans->id, tstrerror(code));
4!
980
    TAOS_RETURN(code);
4✔
981
  }
982

983
  TAOS_RETURN(code);
412✔
984
}
985

986
static bool mndTransActionsOfSameType(SArray *pActions) {
105,890✔
987
  int32_t size = taosArrayGetSize(pActions);
105,890✔
988
  ETrnAct lastActType = TRANS_ACTION_NULL;
105,890✔
989
  bool    same = true;
105,890✔
990
  for (int32_t i = 0; i < size; ++i) {
303,867✔
991
    STransAction *pAction = taosArrayGet(pActions, i);
197,977✔
992
    if (i > 0) {
197,977✔
993
      if (lastActType != pAction->actionType) {
112,165!
994
        same = false;
×
995
        break;
×
996
      }
997
    }
998
    lastActType = pAction->actionType;
197,977✔
999
  }
1000
  return same;
105,890✔
1001
}
1002

1003
static int32_t mndTransCheckParallelActions(SMnode *pMnode, STrans *pTrans) {
47,198✔
1004
  int32_t code = 0;
47,198✔
1005
  if (pTrans->exec == TRN_EXEC_PARALLEL) {
47,198✔
1006
    if (mndTransActionsOfSameType(pTrans->redoActions) == false) {
44,276!
1007
      code = TSDB_CODE_MND_TRANS_INVALID_STAGE;
×
1008
      mError("trans:%d, types of parallel redo actions are not the same", pTrans->id);
×
1009
      TAOS_RETURN(code);
×
1010
    }
1011

1012
    if (pTrans->policy == TRN_POLICY_ROLLBACK) {
44,276✔
1013
      if (mndTransActionsOfSameType(pTrans->undoActions) == false) {
14,416!
1014
        code = TSDB_CODE_MND_TRANS_INVALID_STAGE;
×
1015
        mError("trans:%d, types of parallel undo actions are not the same", pTrans->id);
×
1016
        TAOS_RETURN(code);
×
1017
      }
1018
    }
1019
  }
1020

1021
  TAOS_RETURN(code);
47,198✔
1022
}
1023

1024
static int32_t mndTransCheckCommitActions(SMnode *pMnode, STrans *pTrans) {
47,198✔
1025
  int32_t code = 0;
47,198✔
1026
  if (!pTrans->changeless && taosArrayGetSize(pTrans->commitActions) <= 0) {
47,198!
1027
    code = TSDB_CODE_MND_TRANS_CLOG_IS_NULL;
×
1028
    mError("trans:%d, commit actions of non-changeless trans are empty", pTrans->id);
×
1029
    TAOS_RETURN(code);
×
1030
  }
1031
  if (mndTransActionsOfSameType(pTrans->commitActions) == false) {
47,198!
1032
    code = TSDB_CODE_MND_TRANS_INVALID_STAGE;
×
1033
    mError("trans:%d, types of commit actions are not the same", pTrans->id);
×
1034
    TAOS_RETURN(code);
×
1035
  }
1036

1037
  TAOS_RETURN(code);
47,198✔
1038
}
1039

1040
int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
47,198✔
1041
  int32_t code = 0;
47,198✔
1042
  if (pTrans == NULL) {
47,198!
1043
    return TSDB_CODE_INVALID_PARA;
×
1044
  }
1045

1046
  TAOS_CHECK_RETURN(mndTransCheckConflict(pMnode, pTrans));
47,198!
1047

1048
  TAOS_CHECK_RETURN(mndTransCheckParallelActions(pMnode, pTrans));
47,198!
1049

1050
  TAOS_CHECK_RETURN(mndTransCheckCommitActions(pMnode, pTrans));
47,198!
1051

1052
  mInfo("trans:%d, prepare transaction", pTrans->id);
47,198!
1053
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
47,198✔
1054
    mError("trans:%d, failed to prepare since %s", pTrans->id, tstrerror(code));
16!
1055
    sdbWriteLock(pMnode->pSdb, SDB_TRANS);
16✔
1056
    tsMaxTransId = TMAX(pTrans->id, tsMaxTransId);
16✔
1057
    sdbUnLock(pMnode->pSdb, SDB_TRANS);
16✔
1058
    TAOS_RETURN(code);
16✔
1059
  }
1060
  mInfo("trans:%d, prepare finished", pTrans->id);
47,182!
1061

1062
  STrans *pNew = mndAcquireTrans(pMnode, pTrans->id);
47,182✔
1063
  if (pNew == NULL) {
47,182!
1064
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
1065
    if (terrno != 0) code = terrno;
×
1066
    mError("trans:%d, failed to read from sdb since %s", pTrans->id, tstrerror(code));
×
1067
    TAOS_RETURN(code);
×
1068
  }
1069

1070
  pNew->pRpcArray = pTrans->pRpcArray;
47,182✔
1071
  pNew->rpcRsp = pTrans->rpcRsp;
47,182✔
1072
  pNew->rpcRspLen = pTrans->rpcRspLen;
47,182✔
1073
  pNew->mTraceId = pTrans->mTraceId;
47,182✔
1074
  pTrans->pRpcArray = NULL;
47,182✔
1075
  pTrans->rpcRsp = NULL;
47,182✔
1076
  pTrans->rpcRspLen = 0;
47,182✔
1077

1078
  mndTransExecute(pMnode, pNew);
47,182✔
1079
  mndReleaseTrans(pMnode, pNew);
47,182✔
1080
  // TDOD change to TAOS_RETURN(code);
1081
  return 0;
47,182✔
1082
}
1083

1084
static int32_t mndTransCommit(SMnode *pMnode, STrans *pTrans) {
47,186✔
1085
  int32_t code = 0;
47,186✔
1086
  mInfo("trans:%d, commit transaction", pTrans->id);
47,186!
1087
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
47,186✔
1088
    mError("trans:%d, failed to commit since %s", pTrans->id, tstrerror(code));
7!
1089
    TAOS_RETURN(code);
7✔
1090
  }
1091
  mInfo("trans:%d, commit finished", pTrans->id);
47,179!
1092
  TAOS_RETURN(code);
47,179✔
1093
}
1094

1095
static int32_t mndTransRollback(SMnode *pMnode, STrans *pTrans) {
4✔
1096
  int32_t code = 0;
4✔
1097
  mInfo("trans:%d, rollback transaction", pTrans->id);
4!
1098
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
4✔
1099
    mError("trans:%d, failed to rollback since %s", pTrans->id, tstrerror(code));
2!
1100
    TAOS_RETURN(code);
2✔
1101
  }
1102
  mInfo("trans:%d, rollback finished", pTrans->id);
2!
1103
  TAOS_RETURN(code);
2✔
1104
}
1105

1106
static int32_t mndTransPreFinish(SMnode *pMnode, STrans *pTrans) {
4✔
1107
  int32_t code = 0;
4✔
1108
  mInfo("trans:%d, pre-finish transaction", pTrans->id);
4!
1109
  if ((code = mndTransSync(pMnode, pTrans)) != 0) {
4!
1110
    mError("trans:%d, failed to pre-finish since %s", pTrans->id, tstrerror(code));
×
1111
    TAOS_RETURN(code);
×
1112
  }
1113
  mInfo("trans:%d, pre-finish finished", pTrans->id);
4!
1114
  TAOS_RETURN(code);
4✔
1115
}
1116

1117
static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
277,385✔
1118
  bool    sendRsp = false;
277,385✔
1119
  int32_t code = pTrans->code;
277,385✔
1120

1121
  if (pTrans->stage == TRN_STAGE_FINISH) {
277,385✔
1122
    sendRsp = true;
102,224✔
1123
  }
1124

1125
  if (pTrans->policy == TRN_POLICY_ROLLBACK) {
277,385✔
1126
    if (pTrans->stage == TRN_STAGE_UNDO_ACTION || pTrans->stage == TRN_STAGE_ROLLBACK) {
75,548✔
1127
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
51✔
1128
      sendRsp = true;
51✔
1129
    }
1130
  } else {
1131
    if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
201,837✔
1132
      if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_APP_IS_STARTING ||
132,641!
1133
          code == TSDB_CODE_SYN_PROPOSE_NOT_READY) {
1134
        if (pTrans->failedTimes > 60) sendRsp = true;
×
1135
      } else {
1136
        if (pTrans->failedTimes > 6) sendRsp = true;
132,641✔
1137
      }
1138
      if (code == 0) code = TSDB_CODE_MND_TRANS_UNKNOW_ERROR;
132,641✔
1139
    }
1140
  }
1141

1142
  if (!sendRsp) {
277,385✔
1143
    return;
175,108✔
1144
  } else {
1145
    mInfo("vgId:1, trans:%d, start to send rsp, stage:%s failedTimes:%d code:0x%x", pTrans->id,
102,277!
1146
          mndTransStr(pTrans->stage), pTrans->failedTimes, code);
1147
  }
1148

1149
  mInfo("vgId:1, trans:%d, start to lock rpc array", pTrans->id);
102,277!
1150
  taosWLockLatch(&pTrans->lockRpcArray);
102,277✔
1151
  int32_t size = taosArrayGetSize(pTrans->pRpcArray);
102,277✔
1152
  if (size <= 0) {
102,277✔
1153
    taosWUnLockLatch(&pTrans->lockRpcArray);
68,258✔
1154
    return;
68,258✔
1155
  }
1156

1157
  for (int32_t i = 0; i < size; ++i) {
68,038✔
1158
    SRpcHandleInfo *pInfo = taosArrayGet(pTrans->pRpcArray, i);
34,019✔
1159
    if (pInfo->handle != NULL) {
34,019✔
1160
      if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED) {
32,437!
1161
        code = TSDB_CODE_MND_TRANS_NETWORK_UNAVAILL;
1✔
1162
      }
1163
      if (code == TSDB_CODE_SYN_TIMEOUT) {
32,437!
1164
        code = TSDB_CODE_MND_TRANS_SYNC_TIMEOUT;
×
1165
      }
1166

1167
      if (i != 0 && code == 0) {
32,437!
1168
        code = TSDB_CODE_MNODE_NOT_FOUND;
×
1169
      }
1170
      mInfo("vgId:1, trans:%d, client:%d start to send rsp, code:0x%x stage:%s app:%p", pTrans->id, i, code,
32,437!
1171
            mndTransStr(pTrans->stage), pInfo->ahandle);
1172

1173
      SRpcMsg rspMsg = {.code = code, .info = *pInfo};
32,437✔
1174

1175
      if (pTrans->originRpcType == TDMT_MND_CREATE_DB) {
32,437✔
1176
        mInfo("trans:%d, origin msgtype:%s", pTrans->id, TMSG_INFO(pTrans->originRpcType));
4,623!
1177
        SDbObj *pDb = mndAcquireDb(pMnode, pTrans->dbname);
4,623✔
1178
        if (pDb != NULL) {
4,623!
1179
          for (int32_t j = 0; j < 12; j++) {
5,432✔
1180
            bool ready = mndIsDbReady(pMnode, pDb);
5,430✔
1181
            if (!ready) {
5,430✔
1182
              mInfo("trans:%d, db:%s not ready yet, wait %d times", pTrans->id, pTrans->dbname, j);
809!
1183
              taosMsleep(1000);
809✔
1184
            } else {
1185
              break;
4,621✔
1186
            }
1187
          }
1188
        }
1189
        mndReleaseDb(pMnode, pDb);
4,623✔
1190
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_STB) {
27,814✔
1191
        void   *pCont = NULL;
9,023✔
1192
        int32_t contLen = 0;
9,023✔
1193
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
9,023✔
1194
          mndTransSetRpcRsp(pTrans, pCont, contLen);
9,020✔
1195
        }
1196
      } else if (pTrans->originRpcType == TDMT_MND_CREATE_INDEX) {
18,791✔
1197
        void   *pCont = NULL;
492✔
1198
        int32_t contLen = 0;
492✔
1199
        if (0 == mndBuildSMCreateStbRsp(pMnode, pTrans->dbname, pTrans->stbname, &pCont, &contLen)) {
492!
1200
          mndTransSetRpcRsp(pTrans, pCont, contLen);
492✔
1201
        }
1202
      }
1203

1204
      if (pTrans->rpcRspLen != 0) {
32,437✔
1205
        void *rpcCont = rpcMallocCont(pTrans->rpcRspLen);
15,997✔
1206
        if (rpcCont != NULL) {
15,997!
1207
          memcpy(rpcCont, pTrans->rpcRsp, pTrans->rpcRspLen);
15,997✔
1208
          rspMsg.pCont = rpcCont;
15,997✔
1209
          rspMsg.contLen = pTrans->rpcRspLen;
15,997✔
1210
        }
1211
      }
1212

1213
      tmsgSendRsp(&rspMsg);
32,437✔
1214

1215
      mInfo("vgId:1, trans:%d, client:%d send rsp finished, code:0x%x stage:%s app:%p", pTrans->id, i, code,
32,437!
1216
            mndTransStr(pTrans->stage), pInfo->ahandle);
1217
    }
1218
  }
1219
  taosArrayClear(pTrans->pRpcArray);
34,019✔
1220
  taosWUnLockLatch(&pTrans->lockRpcArray);
34,019✔
1221
}
1222

1223
int32_t mndTransProcessRsp(SRpcMsg *pRsp) {
104,156✔
1224
  int32_t code = 0;
104,156✔
1225
  SMnode *pMnode = pRsp->info.node;
104,156✔
1226
  int64_t signature = (int64_t)(pRsp->info.ahandle);
104,156✔
1227
  int32_t transId = (int32_t)(signature >> 32);
104,156✔
1228
  int32_t action = (int32_t)((signature << 32) >> 32);
104,156✔
1229

1230
  STrans *pTrans = mndAcquireTrans(pMnode, transId);
104,156✔
1231
  if (pTrans == NULL) {
104,156!
1232
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
1233
    if (terrno != 0) code = terrno;
×
1234
    mError("trans:%d, failed to get transId from vnode rsp since %s", transId, tstrerror(code));
×
1235
    goto _OVER;
×
1236
  }
1237

1238
  SArray *pArray = NULL;
104,156✔
1239
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
104,156✔
1240
    pArray = pTrans->redoActions;
104,132✔
1241
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
24!
1242
    pArray = pTrans->undoActions;
24✔
1243
  } else {
1244
    mError("trans:%d, invalid trans stage:%d while recv action rsp", pTrans->id, pTrans->stage);
×
1245
    goto _OVER;
×
1246
  }
1247

1248
  if (pArray == NULL) {
104,156!
1249
    mError("trans:%d, invalid trans stage:%d", transId, pTrans->stage);
×
1250
    goto _OVER;
×
1251
  }
1252

1253
  int32_t actionNum = taosArrayGetSize(pArray);
104,156✔
1254
  if (action < 0 || action >= actionNum) {
104,156!
1255
    mError("trans:%d, invalid action:%d", transId, action);
×
1256
    goto _OVER;
×
1257
  }
1258

1259
  STransAction *pAction = taosArrayGet(pArray, action);
104,156✔
1260
  if (pAction != NULL) {
104,156!
1261
    pAction->msgReceived = 1;
104,156✔
1262
    pAction->errCode = pRsp->code;
104,156✔
1263
    pTrans->lastErrorNo = pRsp->code;
104,156✔
1264

1265
    mInfo("trans:%d, %s:%d response is received, received code:0x%x(%s), accept:0x%x(%s) retry:0x%x(%s)", transId,
104,156!
1266
          mndTransStr(pAction->stage), action, pRsp->code, tstrerror(pRsp->code), pAction->acceptableCode,
1267
          tstrerror(pAction->acceptableCode), pAction->retryCode, tstrerror(pAction->retryCode));
1268
  } else {
1269
    mInfo("trans:%d, invalid action, index:%d, code:0x%x", transId, action, pRsp->code);
×
1270
  }
1271

1272
  mndTransExecute(pMnode, pTrans);
104,156✔
1273

1274
_OVER:
104,156✔
1275
  mndReleaseTrans(pMnode, pTrans);
104,156✔
1276
  TAOS_RETURN(code);
104,156✔
1277
}
1278

1279
static void mndTransResetAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction) {
6,067✔
1280
  pAction->rawWritten = 0;
6,067✔
1281
  pAction->msgSent = 0;
6,067✔
1282
  pAction->msgReceived = 0;
6,067✔
1283
  if (pAction->errCode == TSDB_CODE_SYN_NEW_CONFIG_ERROR || pAction->errCode == TSDB_CODE_SYN_INTERNAL_ERROR ||
6,067!
1284
      pAction->errCode == TSDB_CODE_SYN_NOT_LEADER) {
6,067✔
1285
    pAction->epSet.inUse = (pAction->epSet.inUse + 1) % pAction->epSet.numOfEps;
31✔
1286
    mInfo("trans:%d, %s:%d execute status is reset and set epset inuse:%d", pTrans->id, mndTransStr(pAction->stage),
31!
1287
          pAction->id, pAction->epSet.inUse);
1288
  } else {
1289
    mInfo("trans:%d, %s:%d execute status is reset", pTrans->id, mndTransStr(pAction->stage), pAction->id);
6,036!
1290
  }
1291
  pAction->errCode = 0;
6,067✔
1292
}
6,067✔
1293

1294
static void mndTransResetActions(SMnode *pMnode, STrans *pTrans, SArray *pArray) {
2✔
1295
  int32_t numOfActions = taosArrayGetSize(pArray);
2✔
1296

1297
  for (int32_t action = 0; action < numOfActions; ++action) {
10✔
1298
    STransAction *pAction = taosArrayGet(pArray, action);
8✔
1299
    if (pAction->msgSent && pAction->msgReceived &&
8!
1300
        (pAction->errCode == 0 || pAction->errCode == pAction->acceptableCode))
8!
1301
      continue;
6✔
1302
    if (pAction->rawWritten && (pAction->errCode == 0 || pAction->errCode == pAction->acceptableCode)) continue;
2!
1303

1304
    mndTransResetAction(pMnode, pTrans, pAction);
2✔
1305
  }
1306
}
2✔
1307

1308
// execute in sync context
1309
static int32_t mndTransWriteSingleLog(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
215,606✔
1310
  if (pAction->rawWritten) return 0;
215,606✔
1311
  if (topHalf) {
128,330!
1312
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
×
1313
  }
1314

1315
  if (pAction->pRaw->type >= SDB_MAX) {
128,330!
1316
    pAction->rawWritten = true;
×
1317
    pAction->errCode = 0;
×
1318
    mndSetTransLastAction(pTrans, pAction);
×
1319
    mInfo("skip sdb raw type:%d since it is not supported", pAction->pRaw->type);
×
1320
    TAOS_RETURN(TSDB_CODE_SUCCESS);
×
1321
  }
1322

1323
  int32_t code = sdbWriteWithoutFree(pMnode->pSdb, pAction->pRaw);
128,330✔
1324
  if (code == 0 || terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
128,330!
1325
    pAction->rawWritten = true;
128,330✔
1326
    pAction->errCode = 0;
128,330✔
1327
    code = 0;
128,330✔
1328
    mInfo("trans:%d, %s:%d write to sdb, type:%s status:%s", pTrans->id, mndTransStr(pAction->stage), pAction->id,
128,330!
1329
          sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1330

1331
    mndSetTransLastAction(pTrans, pAction);
128,330✔
1332
  } else {
1333
    pAction->errCode = (terrno != 0) ? terrno : code;
×
1334
    mError("trans:%d, %s:%d failed to write sdb since %s, type:%s status:%s", pTrans->id, mndTransStr(pAction->stage),
×
1335
           pAction->id, tstrerror(code), sdbTableName(pAction->pRaw->type), sdbStatusName(pAction->pRaw->status));
1336
    mndSetTransLastAction(pTrans, pAction);
×
1337
  }
1338

1339
  TAOS_RETURN(code);
128,330✔
1340
}
1341

1342
// execute in trans context
1343
static int32_t mndTransSendSingleMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
666,218✔
1344
  if (pAction->msgSent) return 0;
666,218✔
1345
  if (mndCannotExecuteTrans(pMnode, topHalf)) {
148,093✔
1346
    TAOS_RETURN(TSDB_CODE_MND_TRANS_CTX_SWITCH);
43,794✔
1347
  }
1348

1349
  int64_t signature = pTrans->id;
104,299✔
1350
  signature = (signature << 32);
104,299✔
1351
  signature += pAction->id;
104,299✔
1352

1353
  SRpcMsg rpcMsg = {.msgType = pAction->msgType, .contLen = pAction->contLen, .info.ahandle = (void *)signature};
104,299✔
1354
  rpcMsg.pCont = rpcMallocCont(pAction->contLen);
104,299✔
1355
  if (rpcMsg.pCont == NULL) {
104,299!
1356
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
1357
    return -1;
×
1358
  }
1359
  rpcMsg.info.traceId.rootId = pTrans->mTraceId;
104,299✔
1360
  rpcMsg.info.notFreeAhandle = 1;
104,299✔
1361

1362
  memcpy(rpcMsg.pCont, pAction->pCont, pAction->contLen);
104,299✔
1363

1364
  char    detail[1024] = {0};
104,299✔
1365
  int32_t len = tsnprintf(detail, sizeof(detail), "msgType:%s numOfEps:%d inUse:%d", TMSG_INFO(pAction->msgType),
104,299!
1366
                          pAction->epSet.numOfEps, pAction->epSet.inUse);
104,299✔
1367
  for (int32_t i = 0; i < pAction->epSet.numOfEps; ++i) {
215,156✔
1368
    len += tsnprintf(detail + len, sizeof(detail) - len, " ep:%d-%s:%u", i, pAction->epSet.eps[i].fqdn,
110,857✔
1369
                     pAction->epSet.eps[i].port);
110,857✔
1370
  }
1371

1372
  int32_t code = tmsgSendReq(&pAction->epSet, &rpcMsg);
104,299✔
1373
  if (code == 0) {
104,299✔
1374
    pAction->msgSent = 1;
104,295✔
1375
    // pAction->msgReceived = 0;
1376
    pAction->errCode = TSDB_CODE_ACTION_IN_PROGRESS;
104,295✔
1377
    mInfo("trans:%d, %s:%d is sent, %s", pTrans->id, mndTransStr(pAction->stage), pAction->id, detail);
104,295!
1378

1379
    mndSetTransLastAction(pTrans, pAction);
104,295✔
1380
  } else {
1381
    pAction->msgSent = 0;
4✔
1382
    pAction->msgReceived = 0;
4✔
1383
    pAction->errCode = (terrno != 0) ? terrno : code;
4!
1384
    mError("trans:%d, %s:%d not send since %s, %s", pTrans->id, mndTransStr(pAction->stage), pAction->id, terrstr(),
4!
1385
           detail);
1386

1387
    mndSetTransLastAction(pTrans, pAction);
4✔
1388
  }
1389

1390
  TAOS_RETURN(code);
104,299✔
1391
}
1392

1393
static int32_t mndTransExecNullMsg(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
×
1394
  if (!topHalf) return TSDB_CODE_MND_TRANS_CTX_SWITCH;
×
1395
  pAction->rawWritten = 0;
×
1396
  pAction->errCode = 0;
×
1397
  mInfo("trans:%d, %s:%d confirm action executed", pTrans->id, mndTransStr(pAction->stage), pAction->id);
×
1398

1399
  mndSetTransLastAction(pTrans, pAction);
×
1400
  return 0;
×
1401
}
1402

1403
static int32_t mndTransExecSingleAction(SMnode *pMnode, STrans *pTrans, STransAction *pAction, bool topHalf) {
881,824✔
1404
  if (pAction->actionType == TRANS_ACTION_RAW) {
881,824✔
1405
    return mndTransWriteSingleLog(pMnode, pTrans, pAction, topHalf);
215,606✔
1406
  } else if (pAction->actionType == TRANS_ACTION_MSG) {
666,218!
1407
    return mndTransSendSingleMsg(pMnode, pTrans, pAction, topHalf);
666,218✔
1408
  } else {
1409
    return mndTransExecNullMsg(pMnode, pTrans, pAction, topHalf);
×
1410
  }
1411
}
1412

1413
static int32_t mndTransExecSingleActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf) {
259,173✔
1414
  int32_t numOfActions = taosArrayGetSize(pArray);
259,173✔
1415
  int32_t code = 0;
259,173✔
1416

1417
  for (int32_t action = 0; action < numOfActions; ++action) {
1,031,605✔
1418
    STransAction *pAction = taosArrayGet(pArray, action);
807,761✔
1419
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf);
807,761✔
1420
    if (code != 0) {
807,761✔
1421
      mInfo("trans:%d, action:%d not executed since %s. numOfActions:%d", pTrans->id, action, tstrerror(code),
35,329!
1422
            numOfActions);
1423
      break;
35,329✔
1424
    }
1425
  }
1426

1427
  return code;
259,173✔
1428
}
1429

1430
static int32_t mndTransExecuteActions(SMnode *pMnode, STrans *pTrans, SArray *pArray, bool topHalf) {
290,449✔
1431
  int32_t numOfActions = taosArrayGetSize(pArray);
290,449✔
1432
  int32_t code = 0;
290,449✔
1433
  if (numOfActions == 0) return 0;
290,449✔
1434

1435
  if ((code = mndTransExecSingleActions(pMnode, pTrans, pArray, topHalf)) != 0) {
259,173✔
1436
    return code;
35,329✔
1437
  }
1438

1439
  int32_t       numOfExecuted = 0;
223,844✔
1440
  int32_t       errCode = 0;
223,844✔
1441
  STransAction *pErrAction = NULL;
223,844✔
1442
  for (int32_t action = 0; action < numOfActions; ++action) {
996,276✔
1443
    STransAction *pAction = taosArrayGet(pArray, action);
772,432✔
1444
    if (pAction->msgReceived || pAction->rawWritten) {
772,432✔
1445
      numOfExecuted++;
475,505✔
1446
      if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
475,505✔
1447
        errCode = pAction->errCode;
3✔
1448
        pErrAction = pAction;
3✔
1449
      }
1450
    } else {
1451
      pErrAction = pAction;
296,927✔
1452
    }
1453
  }
1454

1455
  mndSetTransLastAction(pTrans, pErrAction);
223,844✔
1456

1457
  if (numOfExecuted == numOfActions) {
223,844✔
1458
    if (errCode == 0) {
131,838✔
1459
      mInfo("trans:%d, all %d actions execute successfully", pTrans->id, numOfActions);
131,836!
1460
      return 0;
131,836✔
1461
    } else {
1462
      mError("trans:%d, all %d actions executed, code:0x%x", pTrans->id, numOfActions, errCode & 0XFFFF);
2!
1463
      mndTransResetActions(pMnode, pTrans, pArray);
2✔
1464
      terrno = errCode;
2✔
1465
      return errCode;
2✔
1466
    }
1467
  } else {
1468
    mInfo("trans:%d, %d of %d actions executed", pTrans->id, numOfExecuted, numOfActions);
92,006!
1469

1470
    for (int32_t action = 0; action < numOfActions; ++action) {
593,062✔
1471
      STransAction *pAction = taosArrayGet(pArray, action);
501,056✔
1472
      mDebug("trans:%d, %s:%d Sent:%d, Received:%d, errCode:0x%x, acceptableCode:0x%x, retryCode:0x%x", pTrans->id,
501,056✔
1473
             mndTransStr(pAction->stage), pAction->id, pAction->msgSent, pAction->msgReceived, pAction->errCode,
1474
             pAction->acceptableCode, pAction->retryCode);
1475
      if (pAction->msgSent) {
501,056!
1476
        if (pAction->msgReceived) {
501,056✔
1477
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
204,129✔
1478
            mndTransResetAction(pMnode, pTrans, pAction);
1✔
1479
            mInfo("trans:%d, %s:%d reset", pTrans->id, mndTransStr(pAction->stage), pAction->id);
1!
1480
          }
1481
        }
1482
      }
1483
    }
1484
    return TSDB_CODE_ACTION_IN_PROGRESS;
92,006✔
1485
  }
1486
}
1487

1488
static int32_t mndTransExecuteRedoActions(SMnode *pMnode, STrans *pTrans, bool topHalf) {
188,185✔
1489
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->redoActions, topHalf);
188,185✔
1490
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
188,185✔
1491
    mError("trans:%d, failed to execute redoActions since:%s, code:0x%x, in %s", pTrans->id, terrstr(), terrno,
6!
1492
           mndStrExecutionContext(topHalf));
1493
  }
1494
  return code;
188,185✔
1495
}
1496

1497
static int32_t mndTransExecuteUndoActions(SMnode *pMnode, STrans *pTrans, bool topHalf) {
53✔
1498
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->undoActions, topHalf);
53✔
1499
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
53!
1500
    mError("trans:%d, failed to execute undoActions since %s. in %s", pTrans->id, terrstr(),
×
1501
           mndStrExecutionContext(topHalf));
1502
  }
1503
  return code;
53✔
1504
}
1505

1506
static int32_t mndTransExecuteCommitActions(SMnode *pMnode, STrans *pTrans, bool topHalf) {
102,211✔
1507
  int32_t code = mndTransExecuteActions(pMnode, pTrans, pTrans->commitActions, topHalf);
102,211✔
1508
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
102,211!
1509
    mError("trans:%d, failed to execute commitActions since %s. in %s", pTrans->id, terrstr(),
×
1510
           mndStrExecutionContext(topHalf));
1511
  }
1512
  return code;
102,211✔
1513
}
1514

1515
static int32_t mndTransExecuteActionsSerial(SMnode *pMnode, STrans *pTrans, SArray *pActions, bool topHalf) {
34,109✔
1516
  int32_t code = 0;
34,109✔
1517
  int32_t numOfActions = taosArrayGetSize(pActions);
34,109✔
1518
  if (numOfActions == 0) return code;
34,109✔
1519

1520
  if (pTrans->actionPos >= numOfActions) {
34,105✔
1521
    return code;
3,263✔
1522
  }
1523

1524
  mInfo("trans:%d, execute %d actions serial, begin at action:%d, stage:%s", pTrans->id, numOfActions,
30,842!
1525
        pTrans->actionPos, mndTransStr(pTrans->stage));
1526

1527
  for (int32_t action = pTrans->actionPos; action < numOfActions; ++action) {
46,488✔
1528
    STransAction *pAction = taosArrayGet(pActions, action);
43,569✔
1529

1530
    mInfo("trans:%d, current action:%d, stage:%s, actionType(1:msg,2:log):%d", pTrans->id, pTrans->actionPos,
43,569!
1531
          mndTransStr(pAction->stage), pAction->actionType);
1532

1533
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf);
43,569✔
1534
    if (code == 0) {
43,569✔
1535
      if (pAction->msgSent) {
35,100✔
1536
        if (pAction->msgReceived) {
31,096✔
1537
          if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
13,821✔
1538
            code = pAction->errCode;
6,064✔
1539
            mndTransResetAction(pMnode, pTrans, pAction);
6,064✔
1540
          } else {
1541
            mInfo("trans:%d, %s:%d execute successfully", pTrans->id, mndTransStr(pAction->stage), action);
7,757!
1542
          }
1543
        } else {
1544
          code = TSDB_CODE_ACTION_IN_PROGRESS;
17,275✔
1545
        }
1546
      } else if (pAction->rawWritten) {
4,004!
1547
        if (pAction->errCode != 0 && pAction->errCode != pAction->acceptableCode) {
4,004!
1548
          code = pAction->errCode;
×
1549
        } else {
1550
          mInfo("trans:%d, %s:%d write successfully", pTrans->id, mndTransStr(pAction->stage), action);
4,004!
1551
        }
1552
      } else {
1553
      }
1554
    }
1555

1556
    if (code == 0) {
43,569✔
1557
      pTrans->failedTimes = 0;
11,761✔
1558
    }
1559
    mndSetTransLastAction(pTrans, pAction);
43,569✔
1560

1561
    if (mndCannotExecuteTrans(pMnode, topHalf)) {
43,569✔
1562
      pTrans->lastErrorNo = code;
10,629✔
1563
      pTrans->code = code;
10,629✔
1564
      mInfo("trans:%d, %s:%d, cannot execute next action in %s, code:%s", pTrans->id, mndTransStr(pAction->stage),
10,629!
1565
            action, mndStrExecutionContext(topHalf), tstrerror(code));
1566
      break;
10,629✔
1567
    }
1568

1569
    if (code == 0) {
32,940✔
1570
      pTrans->code = 0;
9,601✔
1571
      pTrans->actionPos++;
9,601✔
1572
      mInfo("trans:%d, %s:%d is executed and need sync to other mnodes", pTrans->id, mndTransStr(pAction->stage),
9,601!
1573
            pAction->id);
1574
      (void)taosThreadMutexUnlock(&pTrans->mutex);
9,601✔
1575
      code = mndTransSync(pMnode, pTrans);
9,601✔
1576
      (void)taosThreadMutexLock(&pTrans->mutex);
9,601✔
1577
      if (code != 0) {
9,601!
1578
        pTrans->actionPos--;
×
1579
        pTrans->code = terrno;
×
1580
        mError("trans:%d, %s:%d is executed and failed to sync to other mnodes since %s", pTrans->id,
×
1581
               mndTransStr(pAction->stage), pAction->id, terrstr());
1582
        break;
×
1583
      }
1584
    } else if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
23,339✔
1585
      mInfo("trans:%d, %s:%d is in progress and wait it finish", pTrans->id, mndTransStr(pAction->stage), pAction->id);
17,275!
1586
      break;
17,275✔
1587
    } else if (code == pAction->retryCode || code == TSDB_CODE_SYN_PROPOSE_NOT_READY ||
6,064!
1588
               code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_SYN_NOT_LEADER) {
50✔
1589
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and retry", pTrans->id, mndTransStr(pAction->stage), pAction->id,
6,045!
1590
            code, tstrerror(code));
1591
      pTrans->lastErrorNo = code;
6,045✔
1592
      taosMsleep(300);
6,045✔
1593
      action--;
6,045✔
1594
      continue;
6,045✔
1595
    } else {
1596
      terrno = code;
19✔
1597
      pTrans->lastErrorNo = code;
19✔
1598
      pTrans->code = code;
19✔
1599
      mInfo("trans:%d, %s:%d receive code:0x%x(%s) and wait another schedule, failedTimes:%d", pTrans->id,
19!
1600
            mndTransStr(pAction->stage), pAction->id, code, tstrerror(code), pTrans->failedTimes);
1601
      break;
19✔
1602
    }
1603
  }
1604

1605
  return code;
30,842✔
1606
}
1607

1608
static int32_t mndTransExecuteRedoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
34,109✔
1609
  int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
34,109✔
1610
  (void)taosThreadMutexLock(&pTrans->mutex);
34,109✔
1611
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
34,109!
1612
    code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->redoActions, topHalf);
34,109✔
1613
  }
1614
  (void)taosThreadMutexUnlock(&pTrans->mutex);
34,109✔
1615
  return code;
34,109✔
1616
}
1617

1618
static int32_t mndTransExecuteUndoActionsSerial(SMnode *pMnode, STrans *pTrans, bool topHalf) {
×
1619
  int32_t code = TSDB_CODE_ACTION_IN_PROGRESS;
×
1620
  (void)taosThreadMutexLock(&pTrans->mutex);
×
1621
  if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
×
1622
    code = mndTransExecuteActionsSerial(pMnode, pTrans, pTrans->undoActions, topHalf);
×
1623
  }
1624
  (void)taosThreadMutexUnlock(&pTrans->mutex);
×
1625
  return code;
×
1626
}
1627

1628
bool mndTransPerformPrepareStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
55,081✔
1629
  bool    continueExec = true;
55,081✔
1630
  int32_t code = 0;
55,081✔
1631
  terrno = 0;
55,081✔
1632

1633
  int32_t numOfActions = taosArrayGetSize(pTrans->prepareActions);
55,081✔
1634
  if (numOfActions == 0) goto _OVER;
55,081✔
1635

1636
  mInfo("trans:%d, execute %d prepare actions.", pTrans->id, numOfActions);
19,577!
1637

1638
  for (int32_t action = 0; action < numOfActions; ++action) {
50,071✔
1639
    STransAction *pAction = taosArrayGet(pTrans->prepareActions, action);
30,494✔
1640
    code = mndTransExecSingleAction(pMnode, pTrans, pAction, topHalf);
30,494✔
1641
    if (code != 0) {
30,494!
1642
      terrno = code;
×
1643
      mError("trans:%d, failed to execute prepare action:%d, numOfActions:%d, since %s", pTrans->id, action,
×
1644
             numOfActions, tstrerror(code));
1645
      return false;
×
1646
    }
1647
  }
1648

1649
_OVER:
19,577✔
1650
  pTrans->stage = TRN_STAGE_REDO_ACTION;
55,081✔
1651
  mInfo("trans:%d, stage from prepare to redoAction", pTrans->id);
55,081!
1652
  return continueExec;
55,081✔
1653
}
1654

1655
static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
222,294✔
1656
  bool    continueExec = true;
222,294✔
1657
  int32_t code = 0;
222,294✔
1658
  terrno = 0;
222,294✔
1659

1660
  if (pTrans->exec == TRN_EXEC_SERIAL) {
222,294✔
1661
    code = mndTransExecuteRedoActionsSerial(pMnode, pTrans, topHalf);
34,109✔
1662
  } else {
1663
    code = mndTransExecuteRedoActions(pMnode, pTrans, topHalf);
188,185✔
1664
  }
1665

1666
  if (code != 0 && code != TSDB_CODE_MND_TRANS_CTX_SWITCH && mndTransIsInSyncContext(topHalf)) {
222,294!
1667
    pTrans->lastErrorNo = code;
×
1668
    pTrans->code = code;
×
1669
    mInfo(
×
1670
        "trans:%d, failed to execute, will retry redo action stage in 100 ms , in %s, "
1671
        "continueExec:%d, code:%s",
1672
        pTrans->id, mndStrExecutionContext(topHalf), continueExec, tstrerror(code));
1673
    taosMsleep(100);
×
1674
    return true;
×
1675
  } else {
1676
    if (mndCannotExecuteTrans(pMnode, topHalf)) {
222,294✔
1677
      mInfo("trans:%d, cannot continue to execute redo action stage in %s, continueExec:%d, code:%s", pTrans->id,
65,835!
1678
            mndStrExecutionContext(topHalf), continueExec, tstrerror(code));
1679
      return false;
65,835✔
1680
    }
1681
  }
1682
  terrno = code;
156,459✔
1683

1684
  if (code == 0) {
156,459✔
1685
    pTrans->code = 0;
47,186✔
1686
    pTrans->stage = TRN_STAGE_COMMIT;
47,186✔
1687
    mInfo("trans:%d, stage from redoAction to commit", pTrans->id);
47,186!
1688
    continueExec = true;
47,186✔
1689
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
109,273!
1690
    mInfo("trans:%d, stage keep on redoAction since %s", pTrans->id, tstrerror(code));
109,248!
1691
    continueExec = false;
109,248✔
1692
  } else {
1693
    pTrans->failedTimes++;
25✔
1694
    pTrans->code = terrno;
25✔
1695
    if (pTrans->policy == TRN_POLICY_ROLLBACK) {
25✔
1696
      if (pTrans->lastAction != 0) {
4✔
1697
        STransAction *pAction = taosArrayGet(pTrans->redoActions, pTrans->lastAction);
2✔
1698
        if (pAction->retryCode != 0 && pAction->retryCode == pAction->errCode) {
2!
1699
          if (pTrans->failedTimes < 6) {
×
1700
            mError("trans:%d, stage keep on redoAction since action:%d code:0x%x not 0x%x, failedTimes:%d", pTrans->id,
×
1701
                   pTrans->lastAction, pTrans->code, pAction->retryCode, pTrans->failedTimes);
1702
            taosMsleep(1000);
×
1703
            continueExec = true;
×
1704
            return true;
×
1705
          }
1706
        }
1707
      }
1708

1709
      pTrans->stage = TRN_STAGE_ROLLBACK;
4✔
1710
      pTrans->actionPos = 0;
4✔
1711
      mError("trans:%d, stage from redoAction to rollback since %s, and set actionPos to %d", pTrans->id, terrstr(),
4!
1712
             pTrans->actionPos);
1713
      continueExec = true;
4✔
1714
    } else {
1715
      mError("trans:%d, stage keep on redoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
21!
1716
      continueExec = false;
21✔
1717
    }
1718
  }
1719

1720
  return continueExec;
156,459✔
1721
}
1722

1723
// execute in trans context
1724
static bool mndTransPerformCommitStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
47,186✔
1725
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
47,186!
1726

1727
  bool    continueExec = true;
47,186✔
1728
  int32_t code = mndTransCommit(pMnode, pTrans);
47,186✔
1729

1730
  if (code == 0) {
47,186✔
1731
    pTrans->code = 0;
47,179✔
1732
    pTrans->stage = TRN_STAGE_COMMIT_ACTION;
47,179✔
1733
    mInfo("trans:%d, stage from commit to commitAction", pTrans->id);
47,179!
1734
    continueExec = true;
47,179✔
1735
  } else {
1736
    pTrans->code = terrno;
7✔
1737
    pTrans->failedTimes++;
7✔
1738
    mError("trans:%d, stage keep on commit since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
7!
1739
    continueExec = false;
7✔
1740
  }
1741

1742
  return continueExec;
47,186✔
1743
}
1744

1745
static bool mndTransPerformCommitActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
102,211✔
1746
  bool    continueExec = true;
102,211✔
1747
  int32_t code = mndTransExecuteCommitActions(pMnode, pTrans, topHalf);
102,211✔
1748

1749
  if (code == 0) {
102,211!
1750
    pTrans->code = 0;
102,211✔
1751
    pTrans->stage = TRN_STAGE_FINISH;  // TRN_STAGE_PRE_FINISH is not necessary
102,211✔
1752
    mInfo("trans:%d, stage from commitAction to finished", pTrans->id);
102,211!
1753
    continueExec = true;
102,211✔
1754
  } else if (code == TSDB_CODE_MND_TRANS_CTX_SWITCH && topHalf) {
×
1755
    pTrans->code = 0;
×
1756
    pTrans->stage = TRN_STAGE_COMMIT;
×
1757
    mInfo("trans:%d, back to commit stage", pTrans->id);
×
1758
    continueExec = true;
×
1759
  } else {
1760
    pTrans->code = terrno;
×
1761
    pTrans->failedTimes++;
×
1762
    mError("trans:%d, stage keep on commitAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1763
    continueExec = false;
×
1764
  }
1765

1766
  return continueExec;
102,211✔
1767
}
1768

1769
static bool mndTransPerformUndoActionStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
53✔
1770
  bool    continueExec = true;
53✔
1771
  int32_t code = 0;
53✔
1772

1773
  if (pTrans->exec == TRN_EXEC_SERIAL) {
53!
1774
    code = mndTransExecuteUndoActionsSerial(pMnode, pTrans, topHalf);
×
1775
  } else {
1776
    code = mndTransExecuteUndoActions(pMnode, pTrans, topHalf);
53✔
1777
  }
1778

1779
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
53✔
1780
  terrno = code;
37✔
1781

1782
  if (code == 0) {
37✔
1783
    pTrans->stage = TRN_STAGE_PRE_FINISH;
4✔
1784
    mInfo("trans:%d, stage from undoAction to pre-finish", pTrans->id);
4!
1785
    continueExec = true;
4✔
1786
  } else if (code == TSDB_CODE_ACTION_IN_PROGRESS || code == TSDB_CODE_MND_TRANS_CTX_SWITCH) {
33!
1787
    mInfo("trans:%d, stage keep on undoAction since %s", pTrans->id, tstrerror(code));
33!
1788
    continueExec = false;
33✔
1789
  } else {
1790
    pTrans->failedTimes++;
×
1791
    mError("trans:%d, stage keep on undoAction since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1792
    continueExec = false;
×
1793
  }
1794

1795
  return continueExec;
37✔
1796
}
1797

1798
// in trans context
1799
static bool mndTransPerformRollbackStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
4✔
1800
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
4!
1801

1802
  bool    continueExec = true;
4✔
1803
  int32_t code = mndTransRollback(pMnode, pTrans);
4✔
1804

1805
  if (code == 0) {
4✔
1806
    pTrans->stage = TRN_STAGE_UNDO_ACTION;
2✔
1807
    continueExec = true;
2✔
1808
  } else {
1809
    pTrans->failedTimes++;
2✔
1810
    mError("trans:%d, stage keep on rollback since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
2!
1811
    continueExec = false;
2✔
1812
  }
1813

1814
  return continueExec;
4✔
1815
}
1816

1817
// excute in trans context
1818
static bool mndTransPerformPreFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
4✔
1819
  if (mndCannotExecuteTrans(pMnode, topHalf)) return false;
4!
1820

1821
  bool    continueExec = true;
4✔
1822
  int32_t code = mndTransPreFinish(pMnode, pTrans);
4✔
1823

1824
  if (code == 0) {
4!
1825
    pTrans->stage = TRN_STAGE_FINISH;
4✔
1826
    mInfo("trans:%d, stage from pre-finish to finish", pTrans->id);
4!
1827
    continueExec = true;
4✔
1828
  } else {
1829
    pTrans->failedTimes++;
×
1830
    mError("trans:%d, stage keep on pre-finish since %s, failedTimes:%d", pTrans->id, terrstr(), pTrans->failedTimes);
×
1831
    continueExec = false;
×
1832
  }
1833

1834
  return continueExec;
4✔
1835
}
1836

1837
static bool mndTransPerformFinishStage(SMnode *pMnode, STrans *pTrans, bool topHalf) {
102,223✔
1838
  bool continueExec = false;
102,223✔
1839
  if (topHalf) return continueExec;
102,223✔
1840

1841
  SSdbRaw *pRaw = mndTransEncode(pTrans);
55,040✔
1842
  if (pRaw == NULL) {
55,040!
1843
    mError("trans:%d, failed to encode while finish trans since %s", pTrans->id, terrstr());
×
1844
    return false;
×
1845
  }
1846
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED));
55,040!
1847

1848
  int32_t code = sdbWrite(pMnode->pSdb, pRaw);
55,040✔
1849
  if (code != 0) {
55,040!
1850
    mError("trans:%d, failed to write sdb since %s", pTrans->id, terrstr());
×
1851
  }
1852

1853
  mInfo("trans:%d, execute finished, code:0x%x, failedTimes:%d createTime:%" PRId64, pTrans->id, pTrans->code,
55,040!
1854
        pTrans->failedTimes, pTrans->createdTime);
1855
  return continueExec;
55,040✔
1856
}
1857

1858
void mndTransExecuteImp(SMnode *pMnode, STrans *pTrans, bool topHalf) {
277,385✔
1859
  bool continueExec = true;
277,385✔
1860

1861
  while (continueExec) {
751,360✔
1862
    mInfo("trans:%d, continue to execute stage:%s in %s, createTime:%" PRId64 "", pTrans->id,
473,975!
1863
          mndTransStr(pTrans->stage), mndStrExecutionContext(topHalf), pTrans->createdTime);
1864
    pTrans->lastExecTime = taosGetTimestampMs();
473,975✔
1865
    switch (pTrans->stage) {
473,975!
1866
      case TRN_STAGE_PREPARE:
×
1867
        continueExec = mndTransPerformPrepareStage(pMnode, pTrans, topHalf);
×
1868
        break;
×
1869
      case TRN_STAGE_REDO_ACTION:
222,294✔
1870
        continueExec = mndTransPerformRedoActionStage(pMnode, pTrans, topHalf);
222,294✔
1871
        break;
222,294✔
1872
      case TRN_STAGE_COMMIT:
47,186✔
1873
        continueExec = mndTransPerformCommitStage(pMnode, pTrans, topHalf);
47,186✔
1874
        break;
47,186✔
1875
      case TRN_STAGE_COMMIT_ACTION:
102,211✔
1876
        continueExec = mndTransPerformCommitActionStage(pMnode, pTrans, topHalf);
102,211✔
1877
        break;
102,211✔
1878
      case TRN_STAGE_ROLLBACK:
4✔
1879
        continueExec = mndTransPerformRollbackStage(pMnode, pTrans, topHalf);
4✔
1880
        break;
4✔
1881
      case TRN_STAGE_UNDO_ACTION:
53✔
1882
        continueExec = mndTransPerformUndoActionStage(pMnode, pTrans, topHalf);
53✔
1883
        break;
53✔
1884
      case TRN_STAGE_PRE_FINISH:
4✔
1885
        continueExec = mndTransPerformPreFinishStage(pMnode, pTrans, topHalf);
4✔
1886
        break;
4✔
1887
      case TRN_STAGE_FINISH:
102,223✔
1888
        continueExec = mndTransPerformFinishStage(pMnode, pTrans, topHalf);
102,223✔
1889
        break;
102,223✔
1890
      default:
×
1891
        continueExec = false;
×
1892
        break;
×
1893
    }
1894
  }
1895

1896
  mndTransSendRpcRsp(pMnode, pTrans);
277,385✔
1897
}
277,385✔
1898

1899
// start trans, pullup, receive rsp, kill
1900
void mndTransExecute(SMnode *pMnode, STrans *pTrans) {
156,668✔
1901
  bool topHalf = true;
156,668✔
1902
  mndTransExecuteImp(pMnode, pTrans, topHalf);
156,668✔
1903
}
156,668✔
1904

1905
// update trans
1906
void mndTransRefresh(SMnode *pMnode, STrans *pTrans) {
120,717✔
1907
  bool topHalf = false;
120,717✔
1908
  mndTransExecuteImp(pMnode, pTrans, topHalf);
120,717✔
1909
}
120,717✔
1910

1911
static int32_t mndProcessTransTimer(SRpcMsg *pReq) {
33,011✔
1912
  mTrace("start to process trans timer");
33,011✔
1913
  mndTransPullup(pReq->info.node);
33,011✔
1914
  return 0;
33,011✔
1915
}
1916

1917
int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) {
×
1918
  SArray *pArray = NULL;
×
1919
  if (pTrans->stage == TRN_STAGE_REDO_ACTION) {
×
1920
    pArray = pTrans->redoActions;
×
1921
  } else if (pTrans->stage == TRN_STAGE_UNDO_ACTION) {
×
1922
    pArray = pTrans->undoActions;
×
1923
  } else {
1924
    TAOS_RETURN(TSDB_CODE_MND_TRANS_INVALID_STAGE);
×
1925
  }
1926

1927
  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
×
1928
    STransAction *pAction = taosArrayGet(pArray, i);
×
1929
    mInfo("trans:%d, %s:%d set processed for kill msg received, errCode from %s to success", pTrans->id,
×
1930
          mndTransStr(pAction->stage), i, tstrerror(pAction->errCode));
1931
    pAction->msgSent = 1;
×
1932
    pAction->msgReceived = 1;
×
1933
    pAction->errCode = 0;
×
1934
  }
1935

1936
  mndTransExecute(pMnode, pTrans);
×
1937
  return 0;
×
1938
}
1939

1940
static int32_t mndProcessKillTransReq(SRpcMsg *pReq) {
1✔
1941
  SMnode       *pMnode = pReq->info.node;
1✔
1942
  SKillTransReq killReq = {0};
1✔
1943
  int32_t       code = -1;
1✔
1944
  STrans       *pTrans = NULL;
1✔
1945

1946
  if (tDeserializeSKillTransReq(pReq->pCont, pReq->contLen, &killReq) != 0) {
1!
1947
    code = TSDB_CODE_INVALID_MSG;
×
1948
    goto _OVER;
×
1949
  }
1950

1951
  mInfo("trans:%d, start to kill", killReq.transId);
1!
1952
  if ((code = mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_KILL_TRANS)) != 0) {
1!
1953
    goto _OVER;
1✔
1954
  }
1955

1956
  pTrans = mndAcquireTrans(pMnode, killReq.transId);
×
1957
  if (pTrans == NULL) {
×
1958
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
1959
    if (terrno != 0) code = terrno;
×
1960
    goto _OVER;
×
1961
  }
1962

1963
  code = mndKillTrans(pMnode, pTrans);
×
1964

1965
_OVER:
1✔
1966
  if (code != 0) {
1!
1967
    mError("trans:%d, failed to kill since %s", killReq.transId, terrstr());
1!
1968
  }
1969

1970
  mndReleaseTrans(pMnode, pTrans);
1✔
1971
  TAOS_RETURN(code);
1✔
1972
}
1973

1974
static int32_t mndCompareTransId(int32_t *pTransId1, int32_t *pTransId2) { return *pTransId1 >= *pTransId2 ? 1 : 0; }
191✔
1975

1976
void mndTransPullup(SMnode *pMnode) {
33,524✔
1977
  SSdb   *pSdb = pMnode->pSdb;
33,524✔
1978
  SArray *pArray = taosArrayInit(sdbGetSize(pSdb, SDB_TRANS), sizeof(int32_t));
33,524✔
1979
  if (pArray == NULL) return;
33,524!
1980

1981
  void *pIter = NULL;
33,524✔
1982
  while (1) {
5,330✔
1983
    STrans *pTrans = NULL;
38,854✔
1984
    pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
38,854✔
1985
    if (pIter == NULL) break;
38,854✔
1986
    if (taosArrayPush(pArray, &pTrans->id) == NULL) {
10,660!
1987
      mError("failed to put trans into array, trans:%d, but pull up will continute", pTrans->id);
×
1988
    }
1989
    sdbRelease(pSdb, pTrans);
5,330✔
1990
  }
1991

1992
  taosArraySort(pArray, (__compar_fn_t)mndCompareTransId);
33,524✔
1993

1994
  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
38,854✔
1995
    int32_t *pTransId = taosArrayGet(pArray, i);
5,330✔
1996
    STrans  *pTrans = mndAcquireTrans(pMnode, *pTransId);
5,330✔
1997
    if (pTrans != NULL) {
5,330!
1998
      mndTransExecute(pMnode, pTrans);
5,330✔
1999
    }
2000
    mndReleaseTrans(pMnode, pTrans);
5,330✔
2001
  }
2002
  taosArrayDestroy(pArray);
33,524✔
2003
}
2004

2005
static int32_t mndRetrieveTrans(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
7,653✔
2006
  SMnode *pMnode = pReq->info.node;
7,653✔
2007
  SSdb   *pSdb = pMnode->pSdb;
7,653✔
2008
  int32_t numOfRows = 0;
7,653✔
2009
  STrans *pTrans = NULL;
7,653✔
2010
  int32_t cols = 0;
7,653✔
2011
  int32_t code = 0;
7,653✔
2012
  int32_t lino = 0;
7,653✔
2013

2014
  while (numOfRows < rows) {
8,690!
2015
    pShow->pIter = sdbFetch(pSdb, SDB_TRANS, pShow->pIter, (void **)&pTrans);
8,691✔
2016
    if (pShow->pIter == NULL) break;
8,700✔
2017

2018
    cols = 0;
1,036✔
2019

2020
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,036✔
2021
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->id, false), pTrans, &lino, _OVER);
1,037!
2022

2023
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,037✔
2024
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->createdTime, false), pTrans, &lino,
1,037!
2025
                        _OVER);
2026

2027
    char stage[TSDB_TRANS_STAGE_LEN + VARSTR_HEADER_SIZE] = {0};
1,037✔
2028
    STR_WITH_MAXSIZE_TO_VARSTR(stage, mndTransStr(pTrans->stage), pShow->pMeta->pSchemas[cols].bytes);
1,037✔
2029
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,037✔
2030
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stage, false), pTrans, &lino, _OVER);
1,037!
2031

2032
    char opername[TSDB_TRANS_OPER_LEN + VARSTR_HEADER_SIZE] = {0};
1,037✔
2033
    STR_WITH_MAXSIZE_TO_VARSTR(opername, pTrans->opername, pShow->pMeta->pSchemas[cols].bytes);
1,037✔
2034
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,037✔
2035
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)opername, false), pTrans, &lino, _OVER);
1,037!
2036

2037
    char dbname[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
1,037✔
2038
    STR_WITH_MAXSIZE_TO_VARSTR(dbname, mndGetDbStr(pTrans->dbname), pShow->pMeta->pSchemas[cols].bytes);
1,037✔
2039
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,037✔
2040
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)dbname, false), pTrans, &lino, _OVER);
1,037!
2041

2042
    char stbname[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
1,037✔
2043
    STR_WITH_MAXSIZE_TO_VARSTR(stbname, mndGetDbStr(pTrans->stbname), pShow->pMeta->pSchemas[cols].bytes);
1,037✔
2044
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,037✔
2045
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbname, false), pTrans, &lino, _OVER);
1,037!
2046

2047
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,037✔
2048
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->failedTimes, false), pTrans, &lino,
1,037!
2049
                        _OVER);
2050

2051
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,037✔
2052
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pTrans->lastExecTime, false), pTrans, &lino,
1,037!
2053
                        _OVER);
2054

2055
    char    lastInfo[TSDB_TRANS_ERROR_LEN + VARSTR_HEADER_SIZE] = {0};
1,037✔
2056
    char    detail[TSDB_TRANS_ERROR_LEN + 1] = {0};
1,037✔
2057
    int32_t len = tsnprintf(detail, sizeof(detail), "action:%d code:0x%x(%s) ", pTrans->lastAction,
3,111✔
2058
                            pTrans->lastErrorNo & 0xFFFF, tstrerror(pTrans->lastErrorNo));
1,037✔
2059
    SEpSet  epset = pTrans->lastEpset;
1,037✔
2060
    if (epset.numOfEps > 0) {
1,037!
2061
      len += tsnprintf(detail + len, sizeof(detail) - len, "msgType:%s numOfEps:%d inUse:%d ",
2,074!
2062
                       TMSG_INFO(pTrans->lastMsgType), epset.numOfEps, epset.inUse);
2,074✔
2063
      for (int32_t i = 0; i < pTrans->lastEpset.numOfEps; ++i) {
2,906✔
2064
        len += tsnprintf(detail + len, sizeof(detail) - len, "ep:%d-%s:%u ", i, epset.eps[i].fqdn, epset.eps[i].port);
1,869✔
2065
      }
2066
    }
2067
    STR_WITH_MAXSIZE_TO_VARSTR(lastInfo, detail, pShow->pMeta->pSchemas[cols].bytes);
1,037✔
2068
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,037✔
2069
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)lastInfo, false), pTrans, &lino, _OVER);
1,037!
2070

2071
    numOfRows++;
1,037✔
2072
    sdbRelease(pSdb, pTrans);
1,037✔
2073
  }
2074

2075
_OVER:
×
2076
  if (code != 0) mError("failed to retrieve at line:%d, since %s", lino, tstrerror(code));
7,663!
2077
  pShow->numOfRows += numOfRows;
7,665✔
2078
  return numOfRows;
7,665✔
2079
}
2080

2081
static void mndCancelGetNextTrans(SMnode *pMnode, void *pIter) {
×
2082
  SSdb *pSdb = pMnode->pSdb;
×
2083
  sdbCancelFetchByType(pSdb, pIter, SDB_TRANS);
×
2084
}
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc