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

taosdata / TDengine / #3564

24 Dec 2024 05:40AM UTC coverage: 62.21% (+1.2%) from 61.045%
#3564

push

travis-ci

web-flow
Merge pull request #29289 from taosdata/fix/TD-33270-2

ci(stream):add stream unit test

138331 of 285924 branches covered (48.38%)

Branch coverage included in aggregate %.

215800 of 283329 relevant lines covered (76.17%)

19198660.97 hits per line

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

66.56
/source/dnode/mnode/impl/src/mndStb.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 "audit.h"
18
#include "mndDb.h"
19
#include "mndDnode.h"
20
#include "mndIndex.h"
21
#include "mndIndexComm.h"
22
#include "mndInfoSchema.h"
23
#include "mndMnode.h"
24
#include "mndPerfSchema.h"
25
#include "mndPrivilege.h"
26
#include "mndScheduler.h"
27
#include "mndShow.h"
28
#include "mndSma.h"
29
#include "mndStb.h"
30
#include "mndTopic.h"
31
#include "mndTrans.h"
32
#include "mndUser.h"
33
#include "mndVgroup.h"
34
#include "tname.h"
35

36
#define STB_VER_NUMBER   2
37
#define STB_RESERVE_SIZE 64
38

39
static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw);
40
static int32_t  mndStbActionInsert(SSdb *pSdb, SStbObj *pStb);
41
static int32_t  mndStbActionDelete(SSdb *pSdb, SStbObj *pStb);
42
static int32_t  mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew);
43
static int32_t  mndProcessTtlTimer(SRpcMsg *pReq);
44
static int32_t  mndProcessTrimDbTimer(SRpcMsg *pReq);
45
static int32_t  mndProcessS3MigrateDbTimer(SRpcMsg *pReq);
46
static int32_t  mndProcessS3MigrateDbRsp(SRpcMsg *pReq);
47
static int32_t  mndProcessCreateStbReq(SRpcMsg *pReq);
48
static int32_t  mndProcessAlterStbReq(SRpcMsg *pReq);
49
static int32_t  mndProcessDropStbReq(SRpcMsg *pReq);
50
static int32_t  mndProcessDropTtltbRsp(SRpcMsg *pReq);
51
static int32_t  mndProcessTrimDbRsp(SRpcMsg *pReq);
52
static int32_t  mndProcessTableMetaReq(SRpcMsg *pReq);
53
static int32_t  mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
54
static int32_t  mndRetrieveStbCol(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
55
static void     mndCancelGetNextStb(SMnode *pMnode, void *pIter);
56
static int32_t  mndProcessTableCfgReq(SRpcMsg *pReq);
57
static int32_t  mndAlterStbImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp,
58
                               void *alterOriData, int32_t alterOriDataLen);
59
static int32_t  mndAlterStbAndUpdateTagIdxImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp,
60
                                              void *alterOriData, int32_t alterOriDataLen, const SMAlterStbReq *pAlter);
61

62
static int32_t mndProcessCreateIndexReq(SRpcMsg *pReq);
63
static int32_t mndProcessDropIndexReq(SRpcMsg *pReq);
64

65
static int32_t mndProcessDropStbReqFromMNode(SRpcMsg *pReq);
66
static int32_t mndProcessDropTbWithTsma(SRpcMsg *pReq);
67
static int32_t mndProcessFetchTtlExpiredTbs(SRpcMsg *pReq);
68

69
int32_t mndInitStb(SMnode *pMnode) {
1,774✔
70
  SSdbTable table = {
1,774✔
71
      .sdbType = SDB_STB,
72
      .keyType = SDB_KEY_BINARY,
73
      .encodeFp = (SdbEncodeFp)mndStbActionEncode,
74
      .decodeFp = (SdbDecodeFp)mndStbActionDecode,
75
      .insertFp = (SdbInsertFp)mndStbActionInsert,
76
      .updateFp = (SdbUpdateFp)mndStbActionUpdate,
77
      .deleteFp = (SdbDeleteFp)mndStbActionDelete,
78
  };
79

80
  mndSetMsgHandle(pMnode, TDMT_MND_CREATE_STB, mndProcessCreateStbReq);
1,774✔
81
  mndSetMsgHandle(pMnode, TDMT_MND_ALTER_STB, mndProcessAlterStbReq);
1,774✔
82
  mndSetMsgHandle(pMnode, TDMT_MND_DROP_STB, mndProcessDropStbReq);
1,774✔
83
  mndSetMsgHandle(pMnode, TDMT_VND_CREATE_STB_RSP, mndTransProcessRsp);
1,774✔
84
  mndSetMsgHandle(pMnode, TDMT_VND_DROP_TTL_TABLE_RSP, mndProcessDropTtltbRsp);
1,774✔
85
  mndSetMsgHandle(pMnode, TDMT_VND_TRIM_RSP, mndProcessTrimDbRsp);
1,774✔
86
  mndSetMsgHandle(pMnode, TDMT_VND_ALTER_STB_RSP, mndTransProcessRsp);
1,774✔
87
  mndSetMsgHandle(pMnode, TDMT_VND_DROP_STB_RSP, mndTransProcessRsp);
1,774✔
88
  mndSetMsgHandle(pMnode, TDMT_MND_TABLE_META, mndProcessTableMetaReq);
1,774✔
89
  mndSetMsgHandle(pMnode, TDMT_MND_TTL_TIMER, mndProcessTtlTimer);
1,774✔
90
  mndSetMsgHandle(pMnode, TDMT_MND_TRIM_DB_TIMER, mndProcessTrimDbTimer);
1,774✔
91
  mndSetMsgHandle(pMnode, TDMT_VND_S3MIGRATE_RSP, mndProcessS3MigrateDbRsp);
1,774✔
92
  mndSetMsgHandle(pMnode, TDMT_MND_S3MIGRATE_DB_TIMER, mndProcessS3MigrateDbTimer);
1,774✔
93
  mndSetMsgHandle(pMnode, TDMT_MND_TABLE_CFG, mndProcessTableCfgReq);
1,774✔
94
  mndSetMsgHandle(pMnode, TDMT_MND_STB_DROP, mndProcessDropStbReqFromMNode);
1,774✔
95
  mndSetMsgHandle(pMnode, TDMT_MND_STB_DROP_RSP, mndTransProcessRsp);
1,774✔
96
  mndSetMsgHandle(pMnode, TDMT_MND_DROP_TB_WITH_TSMA, mndProcessDropTbWithTsma);
1,774✔
97
  mndSetMsgHandle(pMnode, TDMT_VND_FETCH_TTL_EXPIRED_TBS_RSP, mndProcessFetchTtlExpiredTbs);
1,774✔
98
  mndSetMsgHandle(pMnode, TDMT_VND_DROP_TABLE_RSP, mndTransProcessRsp);
1,774✔
99
  //  mndSetMsgHandle(pMnode, TDMT_MND_SYSTABLE_RETRIEVE, mndProcessRetrieveStbReq);
100

101
  // mndSetMsgHandle(pMnode, TDMT_MND_CREATE_INDEX, mndProcessCreateIndexReq);
102
  // mndSetMsgHandle(pMnode, TDMT_MND_DROP_INDEX, mndProcessDropIndexReq);
103
  // mndSetMsgHandle(pMnode, TDMT_VND_CREATE_INDEX_RSP, mndTransProcessRsp);
104
  // mndSetMsgHandle(pMnode, TDMT_VND_DROP_INDEX_RSP, mndTransProcessRsp);
105

106
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_STB, mndRetrieveStb);
1,774✔
107
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_STB, mndCancelGetNextStb);
1,774✔
108

109
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_COL, mndRetrieveStbCol);
1,774✔
110
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_COL, mndCancelGetNextStb);
1,774✔
111

112
  return sdbSetTable(pMnode->pSdb, table);
1,774✔
113
}
114

115
void mndCleanupStb(SMnode *pMnode) {}
1,773✔
116

117
SSdbRaw *mndStbActionEncode(SStbObj *pStb) {
36,470✔
118
  int32_t code = 0;
36,470✔
119
  int32_t lino = 0;
36,470✔
120
  terrno = TSDB_CODE_OUT_OF_MEMORY;
36,470✔
121

122
  int32_t size = sizeof(SStbObj) + (pStb->numOfColumns + pStb->numOfTags) * sizeof(SSchema) + pStb->commentLen +
72,940✔
123
                 pStb->ast1Len + pStb->ast2Len + pStb->numOfColumns * sizeof(SColCmpr) + STB_RESERVE_SIZE +
72,940✔
124
                 taosArrayGetSize(pStb->pFuncs) * TSDB_FUNC_NAME_LEN;
36,470✔
125
  SSdbRaw *pRaw = sdbAllocRaw(SDB_STB, STB_VER_NUMBER, size);
36,470✔
126
  if (pRaw == NULL) goto _OVER;
36,470!
127

128
  int32_t dataPos = 0;
36,470✔
129
  SDB_SET_BINARY(pRaw, dataPos, pStb->name, TSDB_TABLE_FNAME_LEN, _OVER)
36,470!
130
  SDB_SET_BINARY(pRaw, dataPos, pStb->db, TSDB_DB_FNAME_LEN, _OVER)
36,470!
131
  SDB_SET_INT64(pRaw, dataPos, pStb->createdTime, _OVER)
36,470!
132
  SDB_SET_INT64(pRaw, dataPos, pStb->updateTime, _OVER)
36,470!
133
  SDB_SET_INT64(pRaw, dataPos, pStb->uid, _OVER)
36,470!
134
  SDB_SET_INT64(pRaw, dataPos, pStb->dbUid, _OVER)
36,470!
135
  SDB_SET_INT32(pRaw, dataPos, pStb->tagVer, _OVER)
36,470!
136
  SDB_SET_INT32(pRaw, dataPos, pStb->colVer, _OVER)
36,470!
137
  SDB_SET_INT32(pRaw, dataPos, pStb->smaVer, _OVER)
36,470!
138
  SDB_SET_INT32(pRaw, dataPos, pStb->nextColId, _OVER)
36,470!
139
  SDB_SET_INT64(pRaw, dataPos, pStb->maxdelay[0], _OVER)
36,470!
140
  SDB_SET_INT64(pRaw, dataPos, pStb->maxdelay[1], _OVER)
36,470!
141
  SDB_SET_INT64(pRaw, dataPos, pStb->watermark[0], _OVER)
36,470!
142
  SDB_SET_INT64(pRaw, dataPos, pStb->watermark[1], _OVER)
36,470!
143
  SDB_SET_INT32(pRaw, dataPos, pStb->ttl, _OVER)
36,470!
144
  SDB_SET_INT32(pRaw, dataPos, pStb->numOfColumns, _OVER)
36,470!
145
  SDB_SET_INT32(pRaw, dataPos, pStb->numOfTags, _OVER)
36,470!
146
  SDB_SET_INT32(pRaw, dataPos, pStb->numOfFuncs, _OVER)
36,470!
147
  SDB_SET_INT32(pRaw, dataPos, pStb->commentLen, _OVER)
36,470!
148
  SDB_SET_INT32(pRaw, dataPos, pStb->ast1Len, _OVER)
36,470!
149
  SDB_SET_INT32(pRaw, dataPos, pStb->ast2Len, _OVER)
36,470!
150

151
  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
2,907,645✔
152
    SSchema *pSchema = &pStb->pColumns[i];
2,871,175✔
153
    SDB_SET_INT8(pRaw, dataPos, pSchema->type, _OVER)
2,871,175!
154
    SDB_SET_INT8(pRaw, dataPos, pSchema->flags, _OVER)
2,871,175!
155
    SDB_SET_INT16(pRaw, dataPos, pSchema->colId, _OVER)
2,871,175!
156
    SDB_SET_INT32(pRaw, dataPos, pSchema->bytes, _OVER)
2,871,175!
157
    SDB_SET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, _OVER)
2,871,175!
158
  }
159

160
  for (int32_t i = 0; i < pStb->numOfTags; ++i) {
444,334✔
161
    SSchema *pSchema = &pStb->pTags[i];
407,864✔
162
    SDB_SET_INT8(pRaw, dataPos, pSchema->type, _OVER)
407,864!
163
    SDB_SET_INT8(pRaw, dataPos, pSchema->flags, _OVER)
407,864!
164
    SDB_SET_INT16(pRaw, dataPos, pSchema->colId, _OVER)
407,864!
165
    SDB_SET_INT32(pRaw, dataPos, pSchema->bytes, _OVER)
407,864!
166
    SDB_SET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, _OVER)
407,864!
167
  }
168

169
  for (int32_t i = 0; i < pStb->numOfFuncs; ++i) {
36,476✔
170
    char *func = taosArrayGet(pStb->pFuncs, i);
6✔
171
    SDB_SET_BINARY(pRaw, dataPos, func, TSDB_FUNC_NAME_LEN, _OVER)
6!
172
  }
173

174
  if (pStb->commentLen > 0) {
36,470✔
175
    SDB_SET_BINARY(pRaw, dataPos, pStb->comment, pStb->commentLen + 1, _OVER)
172!
176
  }
177

178
  if (pStb->ast1Len > 0) {
36,470✔
179
    SDB_SET_BINARY(pRaw, dataPos, pStb->pAst1, pStb->ast1Len, _OVER)
6!
180
  }
181

182
  if (pStb->ast2Len > 0) {
36,470✔
183
    SDB_SET_BINARY(pRaw, dataPos, pStb->pAst2, pStb->ast2Len, _OVER)
6!
184
  }
185

186
  if (pStb->pCmpr != NULL) {
36,470!
187
    for (int i = 0; i < pStb->numOfColumns; i++) {
2,907,645✔
188
      SColCmpr *p = &pStb->pCmpr[i];
2,871,175✔
189
      SDB_SET_INT16(pRaw, dataPos, p->id, _OVER)
2,871,175!
190
      SDB_SET_INT32(pRaw, dataPos, p->alg, _OVER)
2,871,175!
191
    }
192
  }
193
  SDB_SET_RESERVE(pRaw, dataPos, STB_RESERVE_SIZE, _OVER)
36,470!
194
  SDB_SET_DATALEN(pRaw, dataPos, _OVER)
36,470!
195

196
  terrno = 0;
36,470✔
197

198
_OVER:
36,470✔
199
  if (terrno != 0) {
36,470!
200
    mError("stb:%s, failed to encode to raw:%p since %s", pStb->name, pRaw, terrstr());
×
201
    sdbFreeRaw(pRaw);
×
202
    return NULL;
×
203
  }
204

205
  mTrace("stb:%s, encode to raw:%p, row:%p", pStb->name, pRaw, pStb);
36,470✔
206
  return pRaw;
36,470✔
207
}
208

209
static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw) {
33,511✔
210
  int32_t code = 0;
33,511✔
211
  int32_t lino = 0;
33,511✔
212
  terrno = TSDB_CODE_OUT_OF_MEMORY;
33,511✔
213
  SSdbRow *pRow = NULL;
33,511✔
214
  SStbObj *pStb = NULL;
33,511✔
215

216
  int8_t sver = 0;
33,511✔
217
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto _OVER;
33,511!
218

219
  if (sver > STB_VER_NUMBER) {
33,511!
220
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
×
221
    goto _OVER;
×
222
  }
223

224
  pRow = sdbAllocRow(sizeof(SStbObj));
33,511✔
225
  if (pRow == NULL) goto _OVER;
33,511!
226

227
  pStb = sdbGetRowObj(pRow);
33,511✔
228
  if (pStb == NULL) goto _OVER;
33,511!
229

230
  int32_t dataPos = 0;
33,511✔
231
  SDB_GET_BINARY(pRaw, dataPos, pStb->name, TSDB_TABLE_FNAME_LEN, _OVER)
33,511!
232
  SDB_GET_BINARY(pRaw, dataPos, pStb->db, TSDB_DB_FNAME_LEN, _OVER)
33,511!
233
  SDB_GET_INT64(pRaw, dataPos, &pStb->createdTime, _OVER)
33,511!
234
  SDB_GET_INT64(pRaw, dataPos, &pStb->updateTime, _OVER)
33,511!
235
  SDB_GET_INT64(pRaw, dataPos, &pStb->uid, _OVER)
33,511!
236
  SDB_GET_INT64(pRaw, dataPos, &pStb->dbUid, _OVER)
33,511!
237
  SDB_GET_INT32(pRaw, dataPos, &pStb->tagVer, _OVER)
33,511!
238
  SDB_GET_INT32(pRaw, dataPos, &pStb->colVer, _OVER)
33,511!
239
  SDB_GET_INT32(pRaw, dataPos, &pStb->smaVer, _OVER)
33,511!
240
  SDB_GET_INT32(pRaw, dataPos, &pStb->nextColId, _OVER)
33,511!
241
  SDB_GET_INT64(pRaw, dataPos, &pStb->maxdelay[0], _OVER)
33,511!
242
  SDB_GET_INT64(pRaw, dataPos, &pStb->maxdelay[1], _OVER)
33,511!
243
  SDB_GET_INT64(pRaw, dataPos, &pStb->watermark[0], _OVER)
33,511!
244
  SDB_GET_INT64(pRaw, dataPos, &pStb->watermark[1], _OVER)
33,511!
245
  SDB_GET_INT32(pRaw, dataPos, &pStb->ttl, _OVER)
33,511!
246
  SDB_GET_INT32(pRaw, dataPos, &pStb->numOfColumns, _OVER)
33,511!
247
  SDB_GET_INT32(pRaw, dataPos, &pStb->numOfTags, _OVER)
33,511!
248
  SDB_GET_INT32(pRaw, dataPos, &pStb->numOfFuncs, _OVER)
33,511!
249
  SDB_GET_INT32(pRaw, dataPos, &pStb->commentLen, _OVER)
33,511!
250
  SDB_GET_INT32(pRaw, dataPos, &pStb->ast1Len, _OVER)
33,511!
251
  SDB_GET_INT32(pRaw, dataPos, &pStb->ast2Len, _OVER)
33,511!
252

253
  pStb->pColumns = taosMemoryCalloc(pStb->numOfColumns, sizeof(SSchema));
33,511!
254
  pStb->pTags = taosMemoryCalloc(pStb->numOfTags, sizeof(SSchema));
33,511!
255
  pStb->pFuncs = taosArrayInit(pStb->numOfFuncs, TSDB_FUNC_NAME_LEN);
33,511✔
256
  if (pStb->pColumns == NULL || pStb->pTags == NULL || pStb->pFuncs == NULL) {
33,511!
257
    goto _OVER;
×
258
  }
259

260
  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
6,832,090✔
261
    SSchema *pSchema = &pStb->pColumns[i];
6,798,579✔
262
    SDB_GET_INT8(pRaw, dataPos, &pSchema->type, _OVER)
6,798,579!
263
    SDB_GET_INT8(pRaw, dataPos, &pSchema->flags, _OVER)
6,798,579!
264
    SDB_GET_INT16(pRaw, dataPos, &pSchema->colId, _OVER)
6,798,579!
265
    SDB_GET_INT32(pRaw, dataPos, &pSchema->bytes, _OVER)
6,798,579!
266
    SDB_GET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, _OVER)
6,798,579!
267
  }
268

269
  for (int32_t i = 0; i < pStb->numOfTags; ++i) {
643,772✔
270
    SSchema *pSchema = &pStb->pTags[i];
610,261✔
271
    SDB_GET_INT8(pRaw, dataPos, &pSchema->type, _OVER)
610,261!
272
    SDB_GET_INT8(pRaw, dataPos, &pSchema->flags, _OVER)
610,261!
273
    SDB_GET_INT16(pRaw, dataPos, &pSchema->colId, _OVER)
610,261!
274
    SDB_GET_INT32(pRaw, dataPos, &pSchema->bytes, _OVER)
610,261!
275
    SDB_GET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, _OVER)
610,261!
276
  }
277

278
  for (int32_t i = 0; i < pStb->numOfFuncs; ++i) {
33,521✔
279
    char funcName[TSDB_FUNC_NAME_LEN] = {0};
10✔
280
    SDB_GET_BINARY(pRaw, dataPos, funcName, TSDB_FUNC_NAME_LEN, _OVER)
10!
281
    if (taosArrayPush(pStb->pFuncs, funcName) == NULL) goto _OVER;
20!
282
  }
283

284
  if (pStb->commentLen > 0) {
33,511✔
285
    pStb->comment = taosMemoryCalloc(pStb->commentLen + 1, 1);
153!
286
    if (pStb->comment == NULL) goto _OVER;
153!
287
    SDB_GET_BINARY(pRaw, dataPos, pStb->comment, pStb->commentLen + 1, _OVER)
153!
288
  }
289

290
  if (pStb->ast1Len > 0) {
33,511✔
291
    pStb->pAst1 = taosMemoryCalloc(pStb->ast1Len, 1);
10!
292
    if (pStb->pAst1 == NULL) goto _OVER;
10!
293
    SDB_GET_BINARY(pRaw, dataPos, pStb->pAst1, pStb->ast1Len, _OVER)
10!
294
  }
295

296
  if (pStb->ast2Len > 0) {
33,511✔
297
    pStb->pAst2 = taosMemoryCalloc(pStb->ast2Len, 1);
10!
298
    if (pStb->pAst2 == NULL) goto _OVER;
10!
299
    SDB_GET_BINARY(pRaw, dataPos, pStb->pAst2, pStb->ast2Len, _OVER)
10!
300
  }
301

302
  pStb->pCmpr = taosMemoryCalloc(pStb->numOfColumns, sizeof(SColCmpr));
33,511!
303
  if (sver < STB_VER_NUMBER) {
33,511!
304
    // compatible with old data, setup default compress value
305
    // impl later
306
    for (int i = 0; i < pStb->numOfColumns; i++) {
×
307
      SSchema  *pSchema = &pStb->pColumns[i];
×
308
      SColCmpr *pCmpr = &pStb->pCmpr[i];
×
309
      pCmpr->id = pSchema->colId;
×
310
      pCmpr->alg = createDefaultColCmprByType(pSchema->type);
×
311
    }
312
  } else {
313
    for (int i = 0; i < pStb->numOfColumns; i++) {
6,832,090✔
314
      SColCmpr *pCmpr = &pStb->pCmpr[i];
6,798,579✔
315
      SDB_GET_INT16(pRaw, dataPos, &pCmpr->id, _OVER)
6,798,579!
316
      SDB_GET_INT32(pRaw, dataPos, (int32_t *)&pCmpr->alg, _OVER)  // compatiable
6,798,579!
317
    }
318
  }
319

320
  SDB_GET_RESERVE(pRaw, dataPos, STB_RESERVE_SIZE, _OVER)
33,511!
321

322
  terrno = 0;
33,511✔
323

324
_OVER:
33,511✔
325
  if (terrno != 0) {
33,511!
326
    mError("stb:%s, failed to decode from raw:%p since %s", pStb == NULL ? "null" : pStb->name, pRaw, terrstr());
×
327
    if (pStb != NULL) {
×
328
      taosMemoryFreeClear(pStb->pColumns);
×
329
      taosMemoryFreeClear(pStb->pTags);
×
330
      taosMemoryFreeClear(pStb->comment);
×
331
      taosMemoryFree(pStb->pCmpr);
×
332
    }
333
    taosMemoryFreeClear(pRow);
×
334
    return NULL;
×
335
  }
336

337
  mTrace("stb:%s, decode from raw:%p, row:%p", pStb->name, pRaw, pStb);
33,511✔
338
  return pRow;
33,511✔
339
}
340

341
void mndFreeStb(SStbObj *pStb) {
42,156✔
342
  taosArrayDestroy(pStb->pFuncs);
42,156✔
343
  taosMemoryFreeClear(pStb->pColumns);
42,156!
344
  taosMemoryFreeClear(pStb->pTags);
42,156!
345
  taosMemoryFreeClear(pStb->comment);
42,156!
346
  taosMemoryFreeClear(pStb->pAst1);
42,156!
347
  taosMemoryFreeClear(pStb->pAst2);
42,156!
348
  taosMemoryFreeClear(pStb->pCmpr);
42,156!
349
}
42,156✔
350

351
static int32_t mndStbActionInsert(SSdb *pSdb, SStbObj *pStb) {
11,693✔
352
  mTrace("stb:%s, perform insert action, row:%p", pStb->name, pStb);
11,693✔
353
  return 0;
11,693✔
354
}
355

356
static int32_t mndStbActionDelete(SSdb *pSdb, SStbObj *pStb) {
40,578✔
357
  mTrace("stb:%s, perform delete action, row:%p", pStb->name, pStb);
40,578✔
358
  mndFreeStb(pStb);
40,578✔
359
  return 0;
40,578✔
360
}
361

362
static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew) {
16,637✔
363
  mTrace("stb:%s, perform update action, old row:%p new row:%p", pOld->name, pOld, pNew);
16,637✔
364

365
  taosWLockLatch(&pOld->lock);
16,637✔
366
  int32_t numOfColumns = pOld->numOfColumns;
16,637✔
367
  if (pOld->numOfColumns < pNew->numOfColumns) {
16,637✔
368
    void *pColumns = taosMemoryMalloc(pNew->numOfColumns * sizeof(SSchema));
3,210!
369
    if (pColumns != NULL) {
3,210!
370
      taosMemoryFree(pOld->pColumns);
3,210!
371
      pOld->pColumns = pColumns;
3,210✔
372
    } else {
373
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
374
      mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());
×
375
      taosWUnLockLatch(&pOld->lock);
×
376
    }
377
  }
378

379
  if (pOld->numOfTags < pNew->numOfTags) {
16,637✔
380
    void *pTags = taosMemoryMalloc(pNew->numOfTags * sizeof(SSchema));
963!
381
    if (pTags != NULL) {
963!
382
      taosMemoryFree(pOld->pTags);
963!
383
      pOld->pTags = pTags;
963✔
384
    } else {
385
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
386
      mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());
×
387
      taosWUnLockLatch(&pOld->lock);
×
388
    }
389
  }
390

391
  if (pOld->commentLen < pNew->commentLen && pNew->commentLen > 0) {
16,637!
392
    void *comment = taosMemoryMalloc(pNew->commentLen + 1);
17!
393
    if (comment != NULL) {
17!
394
      taosMemoryFree(pOld->comment);
17!
395
      pOld->comment = comment;
17✔
396
    } else {
397
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
398
      mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());
×
399
      taosWUnLockLatch(&pOld->lock);
×
400
    }
401
  }
402
  pOld->commentLen = pNew->commentLen;
16,637✔
403

404
  if (pOld->ast1Len < pNew->ast1Len) {
16,637!
405
    void *pAst1 = taosMemoryMalloc(pNew->ast1Len + 1);
×
406
    if (pAst1 != NULL) {
×
407
      taosMemoryFree(pOld->pAst1);
×
408
      pOld->pAst1 = pAst1;
×
409
    } else {
410
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
411
      mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());
×
412
      taosWUnLockLatch(&pOld->lock);
×
413
    }
414
  }
415

416
  if (pOld->ast2Len < pNew->ast2Len) {
16,637!
417
    void *pAst2 = taosMemoryMalloc(pNew->ast2Len + 1);
×
418
    if (pAst2 != NULL) {
×
419
      taosMemoryFree(pOld->pAst2);
×
420
      pOld->pAst2 = pAst2;
×
421
    } else {
422
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
423
      mTrace("stb:%s, failed to perform update action since %s", pOld->name, terrstr());
×
424
      taosWUnLockLatch(&pOld->lock);
×
425
    }
426
  }
427

428
  pOld->updateTime = pNew->updateTime;
16,637✔
429
  pOld->tagVer = pNew->tagVer;
16,637✔
430
  pOld->colVer = pNew->colVer;
16,637✔
431
  pOld->smaVer = pNew->smaVer;
16,637✔
432
  pOld->nextColId = pNew->nextColId;
16,637✔
433
  pOld->ttl = pNew->ttl;
16,637✔
434
  if (pNew->numOfColumns > 0) {
16,637✔
435
    pOld->numOfColumns = pNew->numOfColumns;
16,593✔
436
    memcpy(pOld->pColumns, pNew->pColumns, pOld->numOfColumns * sizeof(SSchema));
16,593✔
437
  }
438
  if (pNew->numOfTags > 0) {
16,637✔
439
    pOld->numOfTags = pNew->numOfTags;
16,593✔
440
    memcpy(pOld->pTags, pNew->pTags, pOld->numOfTags * sizeof(SSchema));
16,593✔
441
  }
442
  if (pNew->commentLen > 0) {
16,637✔
443
    memcpy(pOld->comment, pNew->comment, pNew->commentLen + 1);
131✔
444
    pOld->commentLen = pNew->commentLen;
131✔
445
  }
446
  if (pNew->ast1Len != 0) {
16,637!
447
    memcpy(pOld->pAst1, pNew->pAst1, pNew->ast1Len);
×
448
    pOld->ast1Len = pNew->ast1Len;
×
449
  }
450
  if (pNew->ast2Len != 0) {
16,637!
451
    memcpy(pOld->pAst2, pNew->pAst2, pNew->ast2Len);
×
452
    pOld->ast2Len = pNew->ast2Len;
×
453
  }
454
  if (numOfColumns < pNew->numOfColumns) {
16,637✔
455
    taosMemoryFree(pOld->pCmpr);
3,210!
456
    pOld->pCmpr = taosMemoryCalloc(pNew->numOfColumns, sizeof(SColCmpr));
3,210!
457
    memcpy(pOld->pCmpr, pNew->pCmpr, pNew->numOfColumns * sizeof(SColCmpr));
3,210✔
458
  } else {
459
    memcpy(pOld->pCmpr, pNew->pCmpr, pNew->numOfColumns * sizeof(SColCmpr));
13,427✔
460
  }
461

462
  taosWUnLockLatch(&pOld->lock);
16,637✔
463
  return 0;
16,637✔
464
}
465

466
SStbObj *mndAcquireStb(SMnode *pMnode, char *stbName) {
935,525✔
467
  SSdb    *pSdb = pMnode->pSdb;
935,525✔
468
  SStbObj *pStb = sdbAcquire(pSdb, SDB_STB, stbName);
935,525✔
469
  if (pStb == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
935,541!
470
    terrno = TSDB_CODE_MND_STB_NOT_EXIST;
17,108✔
471
  }
472
  return pStb;
935,541✔
473
}
474

475
void mndReleaseStb(SMnode *pMnode, SStbObj *pStb) {
927,372✔
476
  SSdb *pSdb = pMnode->pSdb;
927,372✔
477
  sdbRelease(pSdb, pStb);
927,372✔
478
}
927,372✔
479

480
SDbObj *mndAcquireDbByStb(SMnode *pMnode, const char *stbName) {
14,958✔
481
  SName name = {0};
14,958✔
482
  if ((terrno = tNameFromString(&name, stbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) return NULL;
14,958!
483

484
  char db[TSDB_TABLE_FNAME_LEN] = {0};
14,958✔
485
  if ((terrno = tNameGetFullDbName(&name, db)) != 0) return NULL;
14,958!
486

487
  return mndAcquireDb(pMnode, db);
14,958✔
488
}
489

490
static FORCE_INLINE int32_t schemaExColIdCompare(const void *colId, const void *pSchema) {
491
  if (*(col_id_t *)colId < ((SSchema *)pSchema)->colId) {
492
    return -1;
493
  } else if (*(col_id_t *)colId > ((SSchema *)pSchema)->colId) {
494
    return 1;
495
  }
496
  return 0;
497
}
498

499
void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen, void *alterOriData,
28,567✔
500
                            int32_t alterOriDataLen) {
501
  SEncoder       encoder = {0};
28,567✔
502
  int32_t        contLen;
503
  SName          name = {0};
28,567✔
504
  SVCreateStbReq req = {0};
28,567✔
505

506
  if ((terrno = tNameFromString(&name, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) {
28,567!
507
    goto _err;
×
508
  }
509
  char dbFName[TSDB_DB_FNAME_LEN] = {0};
28,567✔
510
  if ((terrno = tNameGetFullDbName(&name, dbFName)) != 0) {
28,567!
511
    goto _err;
×
512
  };
513

514
  req.name = (char *)tNameGetTableName(&name);
28,567✔
515
  req.suid = pStb->uid;
28,567✔
516
  req.rollup = pStb->ast1Len > 0 ? 1 : 0;
28,567✔
517
  req.alterOriData = alterOriData;
28,567✔
518
  req.alterOriDataLen = alterOriDataLen;
28,567✔
519
  req.source = pStb->source;
28,567✔
520
  // todo
521
  req.schemaRow.nCols = pStb->numOfColumns;
28,567✔
522
  req.schemaRow.version = pStb->colVer;
28,567✔
523
  req.schemaRow.pSchema = pStb->pColumns;
28,567✔
524
  req.schemaTag.nCols = pStb->numOfTags;
28,567✔
525
  req.schemaTag.version = pStb->tagVer;
28,567✔
526
  req.schemaTag.pSchema = pStb->pTags;
28,567✔
527

528
  req.colCmpred = 1;
28,567✔
529
  SColCmprWrapper *pCmpr = &req.colCmpr;
28,567✔
530
  pCmpr->version = pStb->colVer;
28,567✔
531
  pCmpr->nCols = pStb->numOfColumns;
28,567✔
532

533
  req.colCmpr.pColCmpr = taosMemoryCalloc(pCmpr->nCols, sizeof(SColCmpr));
28,567!
534
  for (int32_t i = 0; i < pStb->numOfColumns; i++) {
2,594,101✔
535
    SColCmpr *p = &pCmpr->pColCmpr[i];
2,565,534✔
536
    p->alg = pStb->pCmpr[i].alg;
2,565,534✔
537
    p->id = pStb->pCmpr[i].id;
2,565,534✔
538
  }
539

540
  if (req.rollup) {
28,567✔
541
    req.rsmaParam.maxdelay[0] = pStb->maxdelay[0];
4✔
542
    req.rsmaParam.maxdelay[1] = pStb->maxdelay[1];
4✔
543
    req.rsmaParam.watermark[0] = pStb->watermark[0];
4✔
544
    req.rsmaParam.watermark[1] = pStb->watermark[1];
4✔
545
    if (pStb->ast1Len > 0) {
4!
546
      if (mndConvertRsmaTask(&req.rsmaParam.qmsg[0], &req.rsmaParam.qmsgLen[0], pStb->pAst1, pStb->uid,
4!
547
                             STREAM_TRIGGER_WINDOW_CLOSE, req.rsmaParam.watermark[0],
548
                             req.rsmaParam.deleteMark[0]) < 0) {
549
        goto _err;
×
550
      }
551
    }
552
    if (pStb->ast2Len > 0) {
4!
553
      if (mndConvertRsmaTask(&req.rsmaParam.qmsg[1], &req.rsmaParam.qmsgLen[1], pStb->pAst2, pStb->uid,
4!
554
                             STREAM_TRIGGER_WINDOW_CLOSE, req.rsmaParam.watermark[1],
555
                             req.rsmaParam.deleteMark[1]) < 0) {
556
        goto _err;
×
557
      }
558
    }
559
  }
560
  // get length
561
  int32_t ret = 0;
28,567✔
562
  tEncodeSize(tEncodeSVCreateStbReq, &req, contLen, ret);
28,567!
563
  if (ret < 0) {
28,567!
564
    goto _err;
×
565
  }
566

567
  contLen += sizeof(SMsgHead);
28,567✔
568

569
  SMsgHead *pHead = taosMemoryCalloc(1, contLen);
28,567!
570
  if (pHead == NULL) {
28,567!
571
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
572
    goto _err;
×
573
  }
574

575
  pHead->contLen = htonl(contLen);
28,567✔
576
  pHead->vgId = htonl(pVgroup->vgId);
28,567✔
577

578
  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
28,567✔
579
  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
28,567✔
580
  if (tEncodeSVCreateStbReq(&encoder, &req) < 0) {
28,567!
581
    taosMemoryFreeClear(pHead);
×
582
    tEncoderClear(&encoder);
×
583
    goto _err;
×
584
  }
585
  tEncoderClear(&encoder);
28,567✔
586

587
  *pContLen = contLen;
28,567✔
588
  taosMemoryFreeClear(req.rsmaParam.qmsg[0]);
28,567!
589
  taosMemoryFreeClear(req.rsmaParam.qmsg[1]);
28,567!
590
  taosMemoryFreeClear(req.colCmpr.pColCmpr);
28,567!
591
  return pHead;
28,567✔
592
_err:
×
593
  taosMemoryFreeClear(req.rsmaParam.qmsg[0]);
×
594
  taosMemoryFreeClear(req.rsmaParam.qmsg[1]);
×
595
  taosMemoryFreeClear(req.colCmpr.pColCmpr);
×
596
  return NULL;
×
597
}
598

599
static void *mndBuildVDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen) {
23,680✔
600
  SName        name = {0};
23,680✔
601
  SVDropStbReq req = {0};
23,680✔
602
  int32_t      contLen = 0;
23,680✔
603
  int32_t      ret = 0;
23,680✔
604
  SMsgHead    *pHead = NULL;
23,680✔
605
  SEncoder     encoder = {0};
23,680✔
606

607
  if ((terrno = tNameFromString(&name, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) {
23,680!
608
    return NULL;
×
609
  }
610

611
  req.name = (char *)tNameGetTableName(&name);
23,680✔
612
  req.suid = pStb->uid;
23,680✔
613

614
  tEncodeSize(tEncodeSVDropStbReq, &req, contLen, ret);
23,680!
615
  if (ret < 0) return NULL;
23,680!
616

617
  contLen += sizeof(SMsgHead);
23,680✔
618
  pHead = taosMemoryMalloc(contLen);
23,680!
619
  if (pHead == NULL) {
23,680!
620
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
621
    return NULL;
×
622
  }
623

624
  pHead->contLen = htonl(contLen);
23,680✔
625
  pHead->vgId = htonl(pVgroup->vgId);
23,680✔
626

627
  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
23,680✔
628

629
  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
23,680✔
630
  int32_t code = tEncodeSVDropStbReq(&encoder, &req);
23,680✔
631
  tEncoderClear(&encoder);
23,680✔
632
  if (code != 0) {
23,680!
633
    terrno = code;
×
634
    return NULL;
×
635
  }
636

637
  *pContLen = contLen;
23,680✔
638
  return pHead;
23,680✔
639
}
640

641
int32_t mndCheckCreateStbReq(SMCreateStbReq *pCreate) {
8,845✔
642
  int32_t code = 0;
8,845✔
643
  if (pCreate->igExists < 0 || pCreate->igExists > 1) {
8,845!
644
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
645
    TAOS_RETURN(code);
×
646
  }
647

648
  if (pCreate->numOfColumns < TSDB_MIN_COLUMNS || pCreate->numOfTags + pCreate->numOfColumns > TSDB_MAX_COLUMNS) {
8,845!
649
    code = TSDB_CODE_PAR_INVALID_COLUMNS_NUM;
×
650
    TAOS_RETURN(code);
×
651
  }
652

653
  if (pCreate->numOfTags <= 0 || pCreate->numOfTags > TSDB_MAX_TAGS) {
8,845!
654
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
655
    TAOS_RETURN(code);
×
656
  }
657

658
  SField *pField = taosArrayGet(pCreate->pColumns, 0);
8,845✔
659
  if (pField->type != TSDB_DATA_TYPE_TIMESTAMP) {
8,845!
660
    code = TSDB_CODE_PAR_INVALID_FIRST_COLUMN;
×
661
    TAOS_RETURN(code);
×
662
  }
663

664
  for (int32_t i = 0; i < pCreate->numOfColumns; ++i) {
192,980✔
665
    SFieldWithOptions *pField1 = taosArrayGet(pCreate->pColumns, i);
184,135✔
666
    if (pField1->type >= TSDB_DATA_TYPE_MAX) {
184,135!
667
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
668
      TAOS_RETURN(code);
×
669
    }
670
    if (pField1->bytes <= 0) {
184,135!
671
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
672
      TAOS_RETURN(code);
×
673
    }
674
    if (pField1->name[0] == 0) {
184,135!
675
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
676
      TAOS_RETURN(code);
×
677
    }
678
  }
679

680
  for (int32_t i = 0; i < pCreate->numOfTags; ++i) {
61,626✔
681
    SField *pField1 = taosArrayGet(pCreate->pTags, i);
52,781✔
682
    if (pField1->type >= TSDB_DATA_TYPE_MAX) {
52,781!
683
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
684
      TAOS_RETURN(code);
×
685
    }
686
    if (pField1->bytes <= 0) {
52,781!
687
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
688
      TAOS_RETURN(code);
×
689
    }
690
    if (pField1->name[0] == 0) {
52,781!
691
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
692
      TAOS_RETURN(code);
×
693
    }
694
  }
695

696
  TAOS_RETURN(code);
8,845✔
697
}
698

699
static int32_t mndSetCreateStbPrepareLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
×
700
  int32_t  code = 0;
×
701
  SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);
×
702
  if (pRedoRaw == NULL) {
×
703
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
704
    if (terrno != 0) code = terrno;
×
705
    TAOS_RETURN(code);
×
706
  }
707
  if ((code = mndTransAppendPrepareLog(pTrans, pRedoRaw)) != 0) {
×
708
    sdbFreeRaw(pRedoRaw);
×
709
    TAOS_RETURN(code);
×
710
  }
711
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING));
×
712

713
  TAOS_RETURN(code);
×
714
}
715

716
static int32_t mndSetCreateStbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
8,630✔
717
  int32_t  code = 0;
8,630✔
718
  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);
8,630✔
719
  if (pCommitRaw == NULL) {
8,630!
720
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
721
    if (terrno != 0) code = terrno;
×
722
    TAOS_RETURN(code);
×
723
  }
724
  if ((code = mndTransAppendCommitlog(pTrans, pCommitRaw)) != 0) {
8,630!
725
    sdbFreeRaw(pCommitRaw);
×
726
    TAOS_RETURN(code);
×
727
  }
728
  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));
8,630!
729

730
  TAOS_RETURN(code);
8,630✔
731
}
732

733
static int32_t mndSetCreateStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
8,630✔
734
  int32_t code = 0;
8,630✔
735
  SSdb   *pSdb = pMnode->pSdb;
8,630✔
736
  SVgObj *pVgroup = NULL;
8,630✔
737
  void   *pIter = NULL;
8,630✔
738
  int32_t contLen;
739

740
  while (1) {
72,427✔
741
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
81,057✔
742
    if (pIter == NULL) break;
81,057✔
743
    if (!mndVgroupInDb(pVgroup, pDb->uid)) {
72,427✔
744
      sdbRelease(pSdb, pVgroup);
51,037✔
745
      continue;
51,037✔
746
    }
747

748
    void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, NULL, 0);
21,390✔
749
    if (pReq == NULL) {
21,390!
750
      sdbCancelFetch(pSdb, pIter);
×
751
      sdbRelease(pSdb, pVgroup);
×
752
      code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
753
      if (terrno != 0) code = terrno;
×
754
      TAOS_RETURN(code);
×
755
    }
756

757
    STransAction action = {0};
21,390✔
758
    action.mTraceId = pTrans->mTraceId;
21,390✔
759
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
21,390✔
760
    action.pCont = pReq;
21,390✔
761
    action.contLen = contLen;
21,390✔
762
    action.msgType = TDMT_VND_CREATE_STB;
21,390✔
763
    action.acceptableCode = TSDB_CODE_TDB_STB_ALREADY_EXIST;
21,390✔
764
    action.retryCode = TSDB_CODE_TDB_STB_NOT_EXIST;
21,390✔
765
    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
21,390!
766
      taosMemoryFree(pReq);
×
767
      sdbCancelFetch(pSdb, pIter);
×
768
      sdbRelease(pSdb, pVgroup);
×
769
      TAOS_RETURN(code);
×
770
    }
771
    sdbRelease(pSdb, pVgroup);
21,390✔
772
  }
773

774
  TAOS_RETURN(code);
8,630✔
775
}
776

777
int32_t mndSetForceDropCreateStbRedoActions(SMnode *pMnode, STrans *pTrans, SVgObj *pVgroup, SStbObj *pStb) {
2✔
778
  int32_t code = 0;
2✔
779
  SSdb   *pSdb = pMnode->pSdb;
2✔
780
  int32_t contLen;
781

782
  void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, NULL, 0);
2✔
783
  if (pReq == NULL) {
2!
784
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
785
    if (terrno != 0) code = terrno;
×
786
    TAOS_RETURN(code);
×
787
  }
788

789
  STransAction action = {0};
2✔
790
  action.mTraceId = pTrans->mTraceId;
2✔
791
  action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
2✔
792
  action.pCont = pReq;
2✔
793
  action.contLen = contLen;
2✔
794
  action.msgType = TDMT_VND_CREATE_STB;
2✔
795
  action.acceptableCode = TSDB_CODE_TDB_STB_ALREADY_EXIST;
2✔
796
  action.retryCode = TSDB_CODE_TDB_STB_NOT_EXIST;
2✔
797
  if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
2!
798
    taosMemoryFree(pReq);
×
799
    TAOS_RETURN(code);
×
800
  }
801

802
  TAOS_RETURN(code);
2✔
803
}
804

805
static int32_t mndSetCreateStbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
8,630✔
806
  int32_t code = 0;
8,630✔
807
  SSdb   *pSdb = pMnode->pSdb;
8,630✔
808
  SVgObj *pVgroup = NULL;
8,630✔
809
  void   *pIter = NULL;
8,630✔
810

811
  while (1) {
72,427✔
812
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
81,057✔
813
    if (pIter == NULL) break;
81,057✔
814
    if (!mndVgroupInDb(pVgroup, pDb->uid)) {
72,427✔
815
      sdbRelease(pSdb, pVgroup);
51,037✔
816
      continue;
51,037✔
817
    }
818

819
    int32_t contLen = 0;
21,390✔
820
    void   *pReq = mndBuildVDropStbReq(pMnode, pVgroup, pStb, &contLen);
21,390✔
821
    if (pReq == NULL) {
21,390!
822
      sdbCancelFetch(pSdb, pIter);
×
823
      sdbRelease(pSdb, pVgroup);
×
824
      code = TSDB_CODE_OUT_OF_MEMORY;
×
825
      TAOS_RETURN(code);
×
826
    }
827

828
    STransAction action = {0};
21,390✔
829
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
21,390✔
830
    action.pCont = pReq;
21,390✔
831
    action.contLen = contLen;
21,390✔
832
    action.msgType = TDMT_VND_DROP_STB;
21,390✔
833
    action.acceptableCode = TSDB_CODE_TDB_STB_NOT_EXIST;
21,390✔
834
    if ((code = mndTransAppendUndoAction(pTrans, &action)) != 0) {
21,390!
835
      taosMemoryFree(pReq);
×
836
      sdbCancelFetch(pSdb, pIter);
×
837
      sdbRelease(pSdb, pVgroup);
×
838
      TAOS_RETURN(code);
×
839
    }
840
    sdbRelease(pSdb, pVgroup);
21,390✔
841
  }
842

843
  TAOS_RETURN(code);
8,630✔
844
}
845

846
static SSchema *mndFindStbColumns(const SStbObj *pStb, const char *colName) {
×
847
  for (int32_t col = 0; col < pStb->numOfColumns; ++col) {
×
848
    SSchema *pSchema = &pStb->pColumns[col];
×
849
    if (strncasecmp(pSchema->name, colName, TSDB_COL_NAME_LEN) == 0) {
×
850
      return pSchema;
×
851
    }
852
  }
853
  return NULL;
×
854
}
855

856
int32_t mndBuildStbFromReq(SMnode *pMnode, SStbObj *pDst, SMCreateStbReq *pCreate, SDbObj *pDb) {
8,645✔
857
  int32_t code = 0;
8,645✔
858
  memcpy(pDst->name, pCreate->name, TSDB_TABLE_FNAME_LEN);
8,645✔
859
  memcpy(pDst->db, pDb->name, TSDB_DB_FNAME_LEN);
8,645✔
860
  pDst->createdTime = taosGetTimestampMs();
8,645✔
861
  pDst->updateTime = pDst->createdTime;
8,645✔
862
  pDst->uid = (pCreate->source == TD_REQ_FROM_TAOX_OLD || pCreate->source == TD_REQ_FROM_TAOX)
8,645✔
863
                  ? pCreate->suid
864
                  : mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN);
17,290!
865
  pDst->dbUid = pDb->uid;
8,645✔
866
  pDst->tagVer = 1;
8,645✔
867
  pDst->colVer = 1;
8,645✔
868
  pDst->smaVer = 1;
8,645✔
869
  pDst->nextColId = 1;
8,645✔
870
  pDst->maxdelay[0] = pCreate->delay1;
8,645✔
871
  pDst->maxdelay[1] = pCreate->delay2;
8,645✔
872
  pDst->watermark[0] = pCreate->watermark1;
8,645✔
873
  pDst->watermark[1] = pCreate->watermark2;
8,645✔
874
  pDst->ttl = pCreate->ttl;
8,645✔
875
  pDst->numOfColumns = pCreate->numOfColumns;
8,645✔
876
  pDst->numOfTags = pCreate->numOfTags;
8,645✔
877
  pDst->numOfFuncs = pCreate->numOfFuncs;
8,645✔
878
  pDst->commentLen = pCreate->commentLen;
8,645✔
879
  pDst->pFuncs = pCreate->pFuncs;
8,645✔
880
  pDst->source = pCreate->source;
8,645✔
881
  pCreate->pFuncs = NULL;
8,645✔
882

883
  if (pDst->commentLen > 0) {
8,645✔
884
    pDst->comment = taosMemoryCalloc(pDst->commentLen + 1, 1);
18!
885
    if (pDst->comment == NULL) {
18!
886
      code = terrno;
×
887
      TAOS_RETURN(code);
×
888
    }
889
    memcpy(pDst->comment, pCreate->pComment, pDst->commentLen + 1);
18✔
890
  }
891

892
  pDst->ast1Len = pCreate->ast1Len;
8,645✔
893
  if (pDst->ast1Len > 0) {
8,645✔
894
    pDst->pAst1 = taosMemoryCalloc(pDst->ast1Len, 1);
3!
895
    if (pDst->pAst1 == NULL) {
3!
896
      code = terrno;
×
897
      TAOS_RETURN(code);
×
898
    }
899
    memcpy(pDst->pAst1, pCreate->pAst1, pDst->ast1Len);
3✔
900
  }
901

902
  pDst->ast2Len = pCreate->ast2Len;
8,645✔
903
  if (pDst->ast2Len > 0) {
8,645✔
904
    pDst->pAst2 = taosMemoryCalloc(pDst->ast2Len, 1);
3!
905
    if (pDst->pAst2 == NULL) {
3!
906
      code = terrno;
×
907
      TAOS_RETURN(code);
×
908
    }
909
    memcpy(pDst->pAst2, pCreate->pAst2, pDst->ast2Len);
3✔
910
  }
911

912
  pDst->pColumns = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SSchema));
8,645!
913
  pDst->pTags = taosMemoryCalloc(1, pDst->numOfTags * sizeof(SSchema));
8,645!
914
  if (pDst->pColumns == NULL || pDst->pTags == NULL) {
8,645!
915
    code = terrno;
×
916
    TAOS_RETURN(code);
×
917
  }
918

919
  if (pDst->nextColId < 0 || pDst->nextColId >= 0x7fff - pDst->numOfColumns - pDst->numOfTags) {
8,645!
920
    code = TSDB_CODE_OUT_OF_RANGE;
×
921
    TAOS_RETURN(code);
×
922
  }
923

924
  for (int32_t i = 0; i < pDst->numOfColumns; ++i) {
191,522✔
925
    SFieldWithOptions *pField = taosArrayGet(pCreate->pColumns, i);
182,877✔
926
    SSchema           *pSchema = &pDst->pColumns[i];
182,877✔
927
    pSchema->type = pField->type;
182,877✔
928
    pSchema->bytes = pField->bytes;
182,877✔
929
    pSchema->flags = pField->flags;
182,877✔
930
    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
182,877✔
931
    pSchema->colId = pDst->nextColId;
182,877✔
932
    pDst->nextColId++;
182,877✔
933
  }
934

935
  for (int32_t i = 0; i < pDst->numOfTags; ++i) {
60,253✔
936
    SField  *pField = taosArrayGet(pCreate->pTags, i);
51,608✔
937
    SSchema *pSchema = &pDst->pTags[i];
51,608✔
938
    pSchema->type = pField->type;
51,608✔
939
    pSchema->bytes = pField->bytes;
51,608✔
940
    if (i == 0) {
51,608✔
941
      SSCHMEA_SET_IDX_ON(pSchema);
8,645✔
942
    }
943
    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
51,608✔
944
    pSchema->colId = pDst->nextColId;
51,608✔
945
    pDst->nextColId++;
51,608✔
946
  }
947
  // set col compress
948
  pDst->pCmpr = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SCmprObj));
8,645!
949
  for (int32_t i = 0; i < pDst->numOfColumns; i++) {
191,522✔
950
    SFieldWithOptions *pField = taosArrayGet(pCreate->pColumns, i);
182,877✔
951
    SSchema           *pSchema = &pDst->pColumns[i];
182,877✔
952

953
    SColCmpr *pColCmpr = &pDst->pCmpr[i];
182,877✔
954
    pColCmpr->id = pSchema->colId;
182,877✔
955
    pColCmpr->alg = pField->compress;
182,877✔
956
  }
957
  TAOS_RETURN(code);
8,645✔
958
}
959
static int32_t mndGenIdxNameForFirstTag(char *fullname, char *dbname, char *stbname, char *tagname) {
7,067✔
960
  SName name = {0};
7,067✔
961
  if ((terrno = tNameFromString(&name, stbname, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) {
7,067!
962
    return -1;
×
963
  }
964
  return snprintf(fullname, TSDB_INDEX_FNAME_LEN, "%s.%s_%s", dbname, tagname, tNameGetTableName(&name));
7,067✔
965
}
966

967
static int32_t mndCreateStb(SMnode *pMnode, SRpcMsg *pReq, SMCreateStbReq *pCreate, SDbObj *pDb) {
7,067✔
968
  SStbObj stbObj = {0};
7,067✔
969
  int32_t code = -1;
7,067✔
970

971
  char fullIdxName[TSDB_INDEX_FNAME_LEN * 2] = {0};
7,067✔
972

973
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pReq, "create-stb");
7,067✔
974
  if (pTrans == NULL) {
7,067!
975
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
976
    if (terrno != 0) code = terrno;
×
977
    goto _OVER;
×
978
  }
979

980
  mInfo("trans:%d, used to create stb:%s", pTrans->id, pCreate->name);
7,067!
981
  TAOS_CHECK_GOTO(mndBuildStbFromReq(pMnode, &stbObj, pCreate, pDb), NULL, _OVER);
7,067!
982

983
  SSchema *pSchema = &(stbObj.pTags[0]);
7,067✔
984
  if (mndGenIdxNameForFirstTag(fullIdxName, pDb->name, stbObj.name, pSchema->name) < 0) {
7,067!
985
    goto _OVER;
×
986
  }
987
  SSIdx idx = {0};
7,067✔
988
  if (mndAcquireGlobalIdx(pMnode, fullIdxName, SDB_IDX, &idx) == 0 && idx.pIdx != NULL) {
7,067!
989
    code = TSDB_CODE_MND_TAG_INDEX_ALREADY_EXIST;
×
990
    mndReleaseIdx(pMnode, idx.pIdx);
×
991
    goto _OVER;
×
992
  }
993

994
  SIdxObj idxObj = {0};
7,067✔
995
  memcpy(idxObj.name, fullIdxName, TSDB_INDEX_FNAME_LEN);
7,067✔
996
  memcpy(idxObj.stb, stbObj.name, TSDB_TABLE_FNAME_LEN);
7,067✔
997
  memcpy(idxObj.db, stbObj.db, TSDB_DB_FNAME_LEN);
7,067✔
998
  memcpy(idxObj.colName, pSchema->name, TSDB_COL_NAME_LEN);
7,067✔
999
  idxObj.createdTime = taosGetTimestampMs();
7,067✔
1000
  idxObj.uid = mndGenerateUid(fullIdxName, strlen(fullIdxName));
7,067✔
1001
  idxObj.stbUid = stbObj.uid;
7,067✔
1002
  idxObj.dbUid = stbObj.dbUid;
7,067✔
1003

1004
  TAOS_CHECK_GOTO(mndSetCreateIdxCommitLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
7,067!
1005

1006
  TAOS_CHECK_GOTO(mndAddStbToTrans(pMnode, pTrans, pDb, &stbObj), NULL, _OVER);
7,067✔
1007
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
7,052✔
1008
  code = 0;
7,051✔
1009

1010
_OVER:
7,067✔
1011
  mndTransDrop(pTrans);
7,067✔
1012
  if (mndStbActionDelete(pMnode->pSdb, &stbObj) != 0) mError("failed to mndStbActionDelete");
7,067!
1013
  TAOS_RETURN(code);
7,067✔
1014
}
1015

1016
int32_t mndAddStbToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
8,645✔
1017
  mndTransSetDbName(pTrans, pDb->name, pStb->name);
8,645✔
1018
  TAOS_CHECK_RETURN(mndTransCheckConflict(pMnode, pTrans));
8,645✔
1019
  TAOS_CHECK_RETURN(mndSetCreateStbCommitLogs(pMnode, pTrans, pDb, pStb));
8,630!
1020
  TAOS_CHECK_RETURN(mndSetCreateStbRedoActions(pMnode, pTrans, pDb, pStb));
8,630!
1021
  TAOS_CHECK_RETURN(mndSetCreateStbUndoActions(pMnode, pTrans, pDb, pStb));
8,630!
1022
  return 0;
8,630✔
1023
}
1024

1025
static int32_t mndProcessTtlTimer(SRpcMsg *pReq) {
6,080✔
1026
  SMnode           *pMnode = pReq->info.node;
6,080✔
1027
  SSdb             *pSdb = pMnode->pSdb;
6,080✔
1028
  SVgObj           *pVgroup = NULL;
6,080✔
1029
  void             *pIter = NULL;
6,080✔
1030
  SVDropTtlTableReq ttlReq = {
6,080✔
1031
      .timestampSec = taosGetTimestampSec(), .ttlDropMaxCount = tsTtlBatchDropNum, .nUids = 0, .pTbUids = NULL};
6,080✔
1032
  int32_t reqLen = tSerializeSVDropTtlTableReq(NULL, 0, &ttlReq);
6,080✔
1033
  int32_t contLen = reqLen + sizeof(SMsgHead);
6,080✔
1034

1035
  mDebug("start to process ttl timer");
6,080✔
1036

1037
  while (1) {
98,549✔
1038
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
104,629✔
1039
    if (pIter == NULL) break;
104,629✔
1040

1041
    int32_t   code = 0;
98,549✔
1042
    SMsgHead *pHead = rpcMallocCont(contLen);
98,549✔
1043
    if (pHead == NULL) {
98,549!
1044
      sdbRelease(pSdb, pVgroup);
×
1045
      continue;
×
1046
    }
1047
    pHead->contLen = htonl(contLen);
98,549✔
1048
    pHead->vgId = htonl(pVgroup->vgId);
98,549✔
1049
    if ((code = tSerializeSVDropTtlTableReq((char *)pHead + sizeof(SMsgHead), reqLen, &ttlReq)) < 0) {
98,549!
1050
      mError("vgId:%d, failed to serialize drop ttl table request since %s", pVgroup->vgId, tstrerror(code));
×
1051
      sdbRelease(pSdb, pVgroup);
×
1052
      continue;
×
1053
    }
1054

1055
    SRpcMsg rpcMsg = {
98,549✔
1056
        .msgType = TDMT_VND_FETCH_TTL_EXPIRED_TBS, .pCont = pHead, .contLen = contLen, .info = pReq->info};
1057
    SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup);
98,549✔
1058
    code = tmsgSendReq(&epSet, &rpcMsg);
98,549✔
1059
    if (code != 0) {
98,549✔
1060
      mError("vgId:%d, failed to send drop ttl table request to vnode since 0x%x", pVgroup->vgId, code);
43!
1061
    } else {
1062
      mDebug("vgId:%d, send drop ttl table request to vnode, time:%" PRId32, pVgroup->vgId, ttlReq.timestampSec);
98,506✔
1063
    }
1064
    sdbRelease(pSdb, pVgroup);
98,549✔
1065
  }
1066

1067
  return 0;
6,080✔
1068
}
1069

1070
static int32_t mndProcessTrimDbTimer(SRpcMsg *pReq) {
4✔
1071
  SMnode     *pMnode = pReq->info.node;
4✔
1072
  SSdb       *pSdb = pMnode->pSdb;
4✔
1073
  SVgObj     *pVgroup = NULL;
4✔
1074
  void       *pIter = NULL;
4✔
1075
  SVTrimDbReq trimReq = {.timestamp = taosGetTimestampSec()};
4✔
1076
  int32_t     reqLen = tSerializeSVTrimDbReq(NULL, 0, &trimReq);
4✔
1077
  int32_t     contLen = reqLen + sizeof(SMsgHead);
4✔
1078

1079
  while (1) {
200✔
1080
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
204✔
1081
    if (pIter == NULL) break;
204✔
1082

1083
    int32_t code = 0;
200✔
1084

1085
    SMsgHead *pHead = rpcMallocCont(contLen);
200✔
1086
    if (pHead == NULL) {
200!
1087
      sdbCancelFetch(pSdb, pVgroup);
×
1088
      sdbRelease(pSdb, pVgroup);
×
1089
      continue;
×
1090
    }
1091
    pHead->contLen = htonl(contLen);
200✔
1092
    pHead->vgId = htonl(pVgroup->vgId);
200✔
1093
    if ((code = tSerializeSVTrimDbReq((char *)pHead + sizeof(SMsgHead), reqLen, &trimReq)) < 0) {
200!
1094
      mError("vgId:%d, failed to serialize trim db request since %s", pVgroup->vgId, tstrerror(code));
×
1095
    }
1096

1097
    SRpcMsg rpcMsg = {.msgType = TDMT_VND_TRIM, .pCont = pHead, .contLen = contLen};
200✔
1098
    SEpSet  epSet = mndGetVgroupEpset(pMnode, pVgroup);
200✔
1099
    code = tmsgSendReq(&epSet, &rpcMsg);
200✔
1100
    if (code != 0) {
200!
1101
      mError("vgId:%d, timer failed to send vnode-trim request to vnode since 0x%x", pVgroup->vgId, code);
×
1102
    } else {
1103
      mInfo("vgId:%d, timer send vnode-trim request to vnode, time:%d", pVgroup->vgId, trimReq.timestamp);
200!
1104
    }
1105
    sdbRelease(pSdb, pVgroup);
200✔
1106
  }
1107

1108
  return 0;
4✔
1109
}
1110

1111
static int32_t mndProcessS3MigrateDbTimer(SRpcMsg *pReq) {
×
1112
  SMnode          *pMnode = pReq->info.node;
×
1113
  SSdb            *pSdb = pMnode->pSdb;
×
1114
  SVgObj          *pVgroup = NULL;
×
1115
  void            *pIter = NULL;
×
1116
  SVS3MigrateDbReq s3migrateReq = {.timestamp = taosGetTimestampSec()};
×
1117
  int32_t          reqLen = tSerializeSVS3MigrateDbReq(NULL, 0, &s3migrateReq);
×
1118
  int32_t          contLen = reqLen + sizeof(SMsgHead);
×
1119

1120
  while (1) {
×
1121
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
×
1122
    if (pIter == NULL) break;
×
1123

1124
    int32_t code = 0;
×
1125

1126
    SMsgHead *pHead = rpcMallocCont(contLen);
×
1127
    if (pHead == NULL) {
×
1128
      sdbRelease(pSdb, pVgroup);
×
1129
      continue;
×
1130
    }
1131
    pHead->contLen = htonl(contLen);
×
1132
    pHead->vgId = htonl(pVgroup->vgId);
×
1133
    if ((code = tSerializeSVS3MigrateDbReq((char *)pHead + sizeof(SMsgHead), reqLen, &s3migrateReq)) < 0) {
×
1134
      mError("vgId:%d, failed to serialize s3migrate db request since %s", pVgroup->vgId, tstrerror(code));
×
1135
      sdbRelease(pSdb, pVgroup);
×
1136
      continue;
×
1137
    }
1138

1139
    SRpcMsg rpcMsg = {.msgType = TDMT_VND_S3MIGRATE, .pCont = pHead, .contLen = contLen};
×
1140
    SEpSet  epSet = mndGetVgroupEpset(pMnode, pVgroup);
×
1141
    code = tmsgSendReq(&epSet, &rpcMsg);
×
1142
    if (code != 0) {
×
1143
      mError("vgId:%d, timer failed to send vnode-s3migrate request to vnode since 0x%x", pVgroup->vgId, code);
×
1144
    } else {
1145
      mInfo("vgId:%d, timer send vnode-s3migrate request to vnode, time:%d", pVgroup->vgId, s3migrateReq.timestamp);
×
1146
    }
1147
    sdbRelease(pSdb, pVgroup);
×
1148
  }
1149

1150
  return 0;
×
1151
}
1152

1153
static int32_t mndFindSuperTableTagIndex(const SStbObj *pStb, const char *tagName) {
3,889✔
1154
  for (int32_t tag = 0; tag < pStb->numOfTags; tag++) {
93,091✔
1155
    if (strcmp(pStb->pTags[tag].name, tagName) == 0) {
90,777✔
1156
      return tag;
1,575✔
1157
    }
1158
  }
1159

1160
  return -1;
2,314✔
1161
}
1162

1163
static int32_t mndFindSuperTableColumnIndex(const SStbObj *pStb, const char *colName) {
3,144✔
1164
  for (int32_t col = 0; col < pStb->numOfColumns; col++) {
951,333✔
1165
    if (strcmp(pStb->pColumns[col].name, colName) == 0) {
948,997✔
1166
      return col;
808✔
1167
    }
1168
  }
1169

1170
  return -1;
2,336✔
1171
}
1172

1173
static bool mndValidateSchema(SSchema *pSchemas, int32_t nSchema, SArray *pFields, int32_t maxLen) {
1,639✔
1174
  int32_t rowLen = 0;
1,639✔
1175
  for (int32_t i = 0; i < nSchema; ++i) {
530,958✔
1176
    rowLen += (pSchemas + i)->bytes;
529,319✔
1177
  }
1178

1179
  int32_t nField = taosArrayGetSize(pFields);
1,639✔
1180
  for (int32_t i = 0; i < nField; ++i) {
3,278✔
1181
    rowLen += ((SField *)TARRAY_GET_ELEM(pFields, i))->bytes;
1,639✔
1182
  }
1183

1184
  return rowLen <= maxLen;
1,639✔
1185
}
1186

1187
static int32_t mndBuildStbFromAlter(SStbObj *pStb, SStbObj *pDst, SMCreateStbReq *createReq) {
93✔
1188
  int32_t code = 0;
93✔
1189
  taosRLockLatch(&pStb->lock);
93✔
1190
  memcpy(pDst, pStb, sizeof(SStbObj));
93✔
1191
  taosRUnLockLatch(&pStb->lock);
93✔
1192

1193
  pDst->source = createReq->source;
93✔
1194
  pDst->updateTime = taosGetTimestampMs();
93✔
1195
  pDst->numOfColumns = createReq->numOfColumns;
93✔
1196
  pDst->numOfTags = createReq->numOfTags;
93✔
1197
  pDst->pColumns = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SSchema));
93!
1198
  pDst->pTags = taosMemoryCalloc(1, pDst->numOfTags * sizeof(SSchema));
93!
1199
  pDst->pCmpr = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SColCmpr));
93!
1200

1201
  if (pDst->pColumns == NULL || pDst->pTags == NULL || pDst->pCmpr == NULL) {
93!
1202
    code = terrno;
×
1203
    TAOS_RETURN(code);
×
1204
  }
1205

1206
  if (pDst->nextColId < 0 || pDst->nextColId >= 0x7fff - pDst->numOfColumns - pDst->numOfTags) {
93!
1207
    code = TSDB_CODE_OUT_OF_RANGE;
×
1208
    TAOS_RETURN(code);
×
1209
  }
1210

1211
  for (int32_t i = 0; i < pDst->numOfColumns; ++i) {
674✔
1212
    SFieldWithOptions *pField = taosArrayGet(createReq->pColumns, i);
581✔
1213
    SSchema           *pSchema = &pDst->pColumns[i];
581✔
1214
    pSchema->type = pField->type;
581✔
1215
    pSchema->bytes = pField->bytes;
581✔
1216
    pSchema->flags = pField->flags;
581✔
1217
    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
581✔
1218
    int32_t cIndex = mndFindSuperTableColumnIndex(pStb, pField->name);
581✔
1219
    if (cIndex >= 0) {
581✔
1220
      pSchema->colId = pStb->pColumns[cIndex].colId;
514✔
1221
    } else {
1222
      pSchema->colId = pDst->nextColId++;
67✔
1223
    }
1224
  }
1225

1226
  for (int32_t i = 0; i < pDst->numOfTags; ++i) {
614✔
1227
    SField  *pField = taosArrayGet(createReq->pTags, i);
521✔
1228
    SSchema *pSchema = &pDst->pTags[i];
521✔
1229
    pSchema->type = pField->type;
521✔
1230
    pSchema->bytes = pField->bytes;
521✔
1231
    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
521✔
1232
    int32_t cIndex = mndFindSuperTableTagIndex(pStb, pField->name);
521✔
1233
    if (cIndex >= 0) {
521✔
1234
      pSchema->colId = pStb->pTags[cIndex].colId;
466✔
1235
    } else {
1236
      pSchema->colId = pDst->nextColId++;
55✔
1237
    }
1238
  }
1239
  for (int32_t i = 0; i < pDst->numOfColumns; i++) {
674✔
1240
    SColCmpr          *p = pDst->pCmpr + i;
581✔
1241
    SFieldWithOptions *pField = taosArrayGet(createReq->pColumns, i);
581✔
1242
    SSchema           *pSchema = &pDst->pColumns[i];
581✔
1243
    p->id = pSchema->colId;
581✔
1244
    if (pField->compress == 0) {
581!
1245
      p->alg = createDefaultColCmprByType(pSchema->type);
×
1246
    } else {
1247
      p->alg = pField->compress;
581✔
1248
    }
1249
  }
1250
  pDst->tagVer = createReq->tagVer;
93✔
1251
  pDst->colVer = createReq->colVer;
93✔
1252
  return TSDB_CODE_SUCCESS;
93✔
1253
}
1254

1255
static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) {
7,267✔
1256
  SMnode        *pMnode = pReq->info.node;
7,267✔
1257
  int32_t        code = -1;
7,267✔
1258
  SStbObj       *pStb = NULL;
7,267✔
1259
  SDbObj        *pDb = NULL;
7,267✔
1260
  SMCreateStbReq createReq = {0};
7,267✔
1261
  bool           isAlter = false;
7,267✔
1262

1263
  if (tDeserializeSMCreateStbReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
7,267!
1264
    code = TSDB_CODE_INVALID_MSG;
×
1265
    goto _OVER;
×
1266
  }
1267

1268
  mInfo("stb:%s, start to create", createReq.name);
7,267!
1269
  if (mndCheckCreateStbReq(&createReq) != 0) {
7,267!
1270
    code = TSDB_CODE_INVALID_MSG;
×
1271
    goto _OVER;
×
1272
  }
1273

1274
  pStb = mndAcquireStb(pMnode, createReq.name);
7,267✔
1275
  if (pStb != NULL) {
7,267✔
1276
    if (createReq.igExists) {
198✔
1277
      if (createReq.source == TD_REQ_FROM_APP) {
195✔
1278
        mInfo("stb:%s, already exist, ignore exist is set", createReq.name);
9!
1279
        code = 0;
9✔
1280
        goto _OVER;
9✔
1281
      } else if (pStb->uid != createReq.suid) {
186!
1282
        mInfo("stb:%s, alter table does not need to be done, because table is deleted", createReq.name);
×
1283
        code = 0;
×
1284
        goto _OVER;
×
1285
      } else if (createReq.tagVer > 0 || createReq.colVer > 0) {
279!
1286
        int32_t tagDelta = createReq.tagVer - pStb->tagVer;
186✔
1287
        int32_t colDelta = createReq.colVer - pStb->colVer;
186✔
1288
        mInfo("stb:%s, already exist while create, input tagVer:%d colVer:%d, exist tagVer:%d colVer:%d",
186!
1289
              createReq.name, createReq.tagVer, createReq.colVer, pStb->tagVer, pStb->colVer);
1290
        if (tagDelta <= 0 && colDelta <= 0) {
186✔
1291
          mInfo("stb:%s, schema version is not incremented and nothing needs to be done", createReq.name);
93!
1292
          code = 0;
93✔
1293
          goto _OVER;
93✔
1294
        } else if ((tagDelta == 1 && colDelta == 0) || (tagDelta == 0 && colDelta == 1) ||
93!
1295
                   (pStb->colVer == 1 && createReq.colVer > 1) || (pStb->tagVer == 1 && createReq.tagVer > 1)) {
×
1296
          isAlter = true;
93✔
1297
          mInfo("stb:%s, schema version is only increased by 1 number, do alter operation", createReq.name);
93!
1298
        } else {
1299
          mError("stb:%s, schema version increase more than 1 number, error is returned", createReq.name);
×
1300
          code = TSDB_CODE_MND_INVALID_SCHEMA_VER;
×
1301
          goto _OVER;
×
1302
        }
1303
      } else {
1304
        mError("stb:%s, already exist while create, input tagVer:%d colVer:%d is invalid, origin tagVer:%d colVer:%d",
×
1305
               createReq.name, createReq.tagVer, createReq.colVer, pStb->tagVer, pStb->colVer);
1306
        code = TSDB_CODE_MND_INVALID_SCHEMA_VER;
×
1307
        goto _OVER;
×
1308
      }
1309
    } else {
1310
      code = TSDB_CODE_MND_STB_ALREADY_EXIST;
3✔
1311
      goto _OVER;
3✔
1312
    }
1313
  } else if (terrno != TSDB_CODE_MND_STB_NOT_EXIST) {
7,069!
1314
    goto _OVER;
×
1315
  } else if ((createReq.source == TD_REQ_FROM_TAOX_OLD || createReq.source == TD_REQ_FROM_TAOX) &&
7,069!
1316
             (createReq.tagVer != 1 || createReq.colVer != 1)) {
49!
1317
    mInfo("stb:%s, alter table does not need to be done, because table is deleted", createReq.name);
2!
1318
    code = 0;
2✔
1319
    goto _OVER;
2✔
1320
  }
1321

1322
  pDb = mndAcquireDbByStb(pMnode, createReq.name);
7,160✔
1323
  if (pDb == NULL) {
7,160!
1324
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
1325
    goto _OVER;
×
1326
  }
1327

1328
  if ((code = mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pDb)) != 0) {
7,160!
1329
    goto _OVER;
×
1330
  }
1331

1332
  int32_t numOfStbs = -1;
7,160✔
1333
  if ((code = mndGetNumOfStbs(pMnode, pDb->name, &numOfStbs)) != 0) {
7,160!
1334
    goto _OVER;
×
1335
  }
1336

1337
  if (pDb->cfg.numOfStables == 1 && numOfStbs != 0) {
7,160!
1338
    code = TSDB_CODE_MND_SINGLE_STB_MODE_DB;
×
1339
    goto _OVER;
×
1340
  }
1341

1342
  if ((code = grantCheck(TSDB_GRANT_STABLE)) < 0) {
7,160!
1343
    goto _OVER;
×
1344
  }
1345

1346
  if (isAlter) {
7,160✔
1347
    bool    needRsp = false;
93✔
1348
    SStbObj pDst = {0};
93✔
1349
    if ((code = mndBuildStbFromAlter(pStb, &pDst, &createReq)) != 0) {
93!
1350
      taosMemoryFreeClear(pDst.pTags);
×
1351
      taosMemoryFreeClear(pDst.pColumns);
×
1352
      taosMemoryFreeClear(pDst.pCmpr);
×
1353
      goto _OVER;
×
1354
    }
1355

1356
    code = mndAlterStbImp(pMnode, pReq, pDb, &pDst, needRsp, NULL, 0);
93✔
1357
    taosMemoryFreeClear(pDst.pTags);
93!
1358
    taosMemoryFreeClear(pDst.pColumns);
93!
1359
    taosMemoryFreeClear(pDst.pCmpr);
93!
1360
  } else {
1361
    code = mndCreateStb(pMnode, pReq, &createReq, pDb);
7,067✔
1362
  }
1363
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
7,160✔
1364

1365
  SName name = {0};
7,160✔
1366
  TAOS_CHECK_RETURN(tNameFromString(&name, createReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
7,160!
1367

1368
  if (createReq.sql == NULL && createReq.sqlLen == 0) {
8,133!
1369
    char detail[1000] = {0};
973✔
1370

1371
    (void)tsnprintf(detail, sizeof(detail), "dbname:%s, stable name:%s", name.dbname, name.tname);
973✔
1372

1373
    auditRecord(pReq, pMnode->clusterId, "createStb", name.dbname, name.tname, detail, strlen(detail));
973✔
1374
  } else {
1375
    auditRecord(pReq, pMnode->clusterId, "createStb", name.dbname, name.tname, createReq.sql, createReq.sqlLen);
6,187✔
1376
  }
1377
_OVER:
7,267✔
1378
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
7,267✔
1379
    mError("stb:%s, failed to create since %s", createReq.name, tstrerror(code));
19!
1380
  }
1381

1382
  mndReleaseStb(pMnode, pStb);
7,267✔
1383
  mndReleaseDb(pMnode, pDb);
7,267✔
1384
  tFreeSMCreateStbReq(&createReq);
7,267✔
1385

1386
  TAOS_RETURN(code);
7,267✔
1387
}
1388

1389
static int32_t mndCheckAlterStbReq(SMAlterStbReq *pAlter) {
3,049✔
1390
  int32_t code = 0;
3,049✔
1391
  if (pAlter->commentLen >= 0) return 0;
3,049!
1392
  if (pAlter->ttl != 0) return 0;
×
1393

1394
  if (pAlter->numOfFields < 1 || pAlter->numOfFields != (int32_t)taosArrayGetSize(pAlter->pFields)) {
×
1395
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
1396
    TAOS_RETURN(code);
×
1397
  }
1398

1399
  for (int32_t i = 0; i < pAlter->numOfFields; ++i) {
×
1400
    SField *pField = taosArrayGet(pAlter->pFields, i);
×
1401
    if (pField->name[0] == 0) {
×
1402
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
1403
      TAOS_RETURN(code);
×
1404
    }
1405
  }
1406

1407
  TAOS_RETURN(code);
×
1408
}
1409

1410
int32_t mndAllocStbSchemas(const SStbObj *pOld, SStbObj *pNew) {
4,438✔
1411
  pNew->pTags = taosMemoryCalloc(pNew->numOfTags, sizeof(SSchema));
4,438!
1412
  pNew->pColumns = taosMemoryCalloc(pNew->numOfColumns, sizeof(SSchema));
4,438!
1413
  pNew->pCmpr = taosMemoryCalloc(pNew->numOfColumns, sizeof(SColCmpr));
4,438!
1414
  if (pNew->pTags == NULL || pNew->pColumns == NULL || pNew->pCmpr == NULL) {
4,438!
1415
    TAOS_RETURN(terrno);
×
1416
  }
1417

1418
  memcpy(pNew->pColumns, pOld->pColumns, sizeof(SSchema) * pOld->numOfColumns);
4,438✔
1419
  memcpy(pNew->pTags, pOld->pTags, sizeof(SSchema) * pOld->numOfTags);
4,438✔
1420
  memcpy(pNew->pCmpr, pOld->pCmpr, sizeof(SColCmpr) * pOld->numOfColumns);
4,438✔
1421

1422
  TAOS_RETURN(0);
4,438✔
1423
}
1424

1425
static int32_t mndUpdateStbCommentAndTTL(const SStbObj *pOld, SStbObj *pNew, char *pComment, int32_t commentLen,
22✔
1426
                                         int32_t ttl) {
1427
  int32_t code = 0;
22✔
1428
  if (commentLen > 0) {
22✔
1429
    pNew->commentLen = commentLen;
17✔
1430
    pNew->comment = taosMemoryCalloc(1, commentLen + 1);
17!
1431
    if (pNew->comment == NULL) {
17!
1432
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
1433
      return -1;
×
1434
    }
1435
    memcpy(pNew->comment, pComment, commentLen + 1);
17✔
1436
  } else if (commentLen == 0) {
5!
1437
    pNew->commentLen = 0;
5✔
1438
  } else {
1439
  }
1440

1441
  if (ttl >= 0) {
22!
1442
    pNew->ttl = ttl;
22✔
1443
  }
1444

1445
  if ((code = mndAllocStbSchemas(pOld, pNew)) != 0) {
22!
1446
    TAOS_RETURN(code);
×
1447
  }
1448
  TAOS_RETURN(code);
22✔
1449
}
1450

1451
static int32_t mndAddSuperTableTag(const SStbObj *pOld, SStbObj *pNew, SArray *pFields, int32_t ntags) {
453✔
1452
  int32_t code = 0;
453✔
1453
  if (pOld->numOfTags + ntags > TSDB_MAX_TAGS) {
453!
1454
    code = TSDB_CODE_MND_TOO_MANY_TAGS;
×
1455
    TAOS_RETURN(code);
×
1456
  }
1457

1458
  if (pOld->numOfColumns + ntags + pOld->numOfTags > TSDB_MAX_COLUMNS) {
453!
1459
    code = TSDB_CODE_MND_TOO_MANY_COLUMNS;
×
1460
    TAOS_RETURN(code);
×
1461
  }
1462

1463
  if (!mndValidateSchema(pOld->pTags, pOld->numOfTags, pFields, TSDB_MAX_TAGS_LEN)) {
453!
1464
    code = TSDB_CODE_PAR_INVALID_TAGS_LENGTH;
×
1465
    TAOS_RETURN(code);
×
1466
  }
1467

1468
  pNew->numOfTags = pNew->numOfTags + ntags;
453✔
1469
  if ((code = mndAllocStbSchemas(pOld, pNew)) != 0) {
453!
1470
    TAOS_RETURN(code);
×
1471
  }
1472

1473
  if (pNew->nextColId < 0 || pNew->nextColId >= 0x7fff - ntags) {
453!
1474
    code = TSDB_CODE_OUT_OF_RANGE;
×
1475
    TAOS_RETURN(code);
×
1476
  }
1477

1478
  for (int32_t i = 0; i < ntags; i++) {
892✔
1479
    SField *pField = taosArrayGet(pFields, i);
453✔
1480
    if (mndFindSuperTableColumnIndex(pOld, pField->name) >= 0) {
453✔
1481
      code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
6✔
1482
      TAOS_RETURN(code);
6✔
1483
    }
1484

1485
    if (mndFindSuperTableTagIndex(pOld, pField->name) >= 0) {
447✔
1486
      code = TSDB_CODE_MND_TAG_ALREADY_EXIST;
8✔
1487
      TAOS_RETURN(code);
8✔
1488
    }
1489

1490
    SSchema *pSchema = &pNew->pTags[pOld->numOfTags + i];
439✔
1491
    pSchema->bytes = pField->bytes;
439✔
1492
    pSchema->type = pField->type;
439✔
1493
    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
439✔
1494
    pSchema->colId = pNew->nextColId;
439✔
1495
    pNew->nextColId++;
439✔
1496

1497
    mInfo("stb:%s, start to add tag %s", pNew->name, pSchema->name);
439!
1498
  }
1499

1500
  pNew->tagVer++;
439✔
1501
  TAOS_RETURN(code);
439✔
1502
}
1503

1504
static int32_t mndCheckAlterColForTopic(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {
1,327✔
1505
  int32_t code = 0;
1,327✔
1506
  SSdb   *pSdb = pMnode->pSdb;
1,327✔
1507
  void   *pIter = NULL;
1,327✔
1508
  while (1) {
1,017✔
1509
    SMqTopicObj *pTopic = NULL;
2,344✔
1510
    pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);
2,344✔
1511
    if (pIter == NULL) break;
2,344✔
1512

1513
    mInfo("topic:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d, subType:%d sql:%s",
1,095!
1514
          pTopic->name, stbFullName, suid, colId, pTopic->subType, pTopic->sql);
1515
    if (pTopic->ast == NULL) {
1,095✔
1516
      sdbRelease(pSdb, pTopic);
103✔
1517
      continue;
103✔
1518
    }
1519

1520
    SNode *pAst = NULL;
992✔
1521
    if (nodesStringToNode(pTopic->ast, &pAst) != 0) {
992!
1522
      code = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC;
×
1523
      mError("topic:%s, create ast error", pTopic->name);
×
1524
      sdbRelease(pSdb, pTopic);
×
1525
      sdbCancelFetch(pSdb, pIter);
×
1526
      TAOS_RETURN(code);
78✔
1527
    }
1528

1529
    SNodeList *pNodeList = NULL;
992✔
1530
    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=
992!
1531
        0) {
1532
      sdbRelease(pSdb, pTopic);
×
1533
      sdbCancelFetch(pSdb, pIter);
×
1534
      TAOS_RETURN(code);
×
1535
    }
1536
    SNode *pNode = NULL;
992✔
1537
    FOREACH(pNode, pNodeList) {
1,633✔
1538
      SColumnNode *pCol = (SColumnNode *)pNode;
1,556✔
1539
      mInfo("topic:%s, check colId:%d tableId:%" PRId64 " ctbStbUid:%" PRId64, pTopic->name, pCol->colId, pCol->tableId,
1,556!
1540
            pTopic->ctbStbUid);
1541

1542
      if (pCol->tableId != suid && pTopic->ctbStbUid != suid) {
1,556✔
1543
        mInfo("topic:%s, check colId:%d passed", pTopic->name, pCol->colId);
837!
1544
        goto NEXT;
837✔
1545
      }
1546
      if (pCol->colId > 0 && pCol->colId == colId) {
719!
1547
        code = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC;
78✔
1548
        mError("topic:%s, check colId:%d conflicted", pTopic->name, pCol->colId);
78!
1549
        nodesDestroyNode(pAst);
78✔
1550
        nodesDestroyList(pNodeList);
78✔
1551
        sdbCancelFetch(pSdb, pIter);
78✔
1552
        sdbRelease(pSdb, pTopic);
78✔
1553
        TAOS_RETURN(code);
78✔
1554
      }
1555
      mInfo("topic:%s, check colId:%d passed", pTopic->name, pCol->colId);
641!
1556
    }
1557

1558
  NEXT:
77✔
1559
    sdbRelease(pSdb, pTopic);
914✔
1560
    nodesDestroyNode(pAst);
914✔
1561
    nodesDestroyList(pNodeList);
914✔
1562
  }
1563
  TAOS_RETURN(code);
1,249✔
1564
}
1565

1566
static int32_t mndCheckAlterColForStream(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {
1,249✔
1567
  int32_t code = 0;
1,249✔
1568
  SSdb   *pSdb = pMnode->pSdb;
1,249✔
1569
  void   *pIter = NULL;
1,249✔
1570
  while (1) {
10✔
1571
    SStreamObj *pStream = NULL;
1,259✔
1572
    pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);
1,259✔
1573
    if (pIter == NULL) break;
1,259✔
1574

1575
    SNode *pAst = NULL;
30✔
1576
    if (nodesStringToNode(pStream->ast, &pAst) != 0) {
30!
1577
      code = TSDB_CODE_MND_INVALID_STREAM_OPTION;
×
1578
      mError("stream:%s, create ast error", pStream->name);
×
1579
      sdbRelease(pSdb, pStream);
×
1580
      sdbCancelFetch(pSdb, pIter);
×
1581
      TAOS_RETURN(code);
20✔
1582
    }
1583

1584
    SNodeList *pNodeList = NULL;
30✔
1585
    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=
30!
1586
        0) {
1587
      sdbRelease(pSdb, pStream);
×
1588
      sdbCancelFetch(pSdb, pIter);
×
1589
      TAOS_RETURN(code);
×
1590
    }
1591

1592
    SNode *pNode = NULL;
30✔
1593
    FOREACH(pNode, pNodeList) {
95!
1594
      SColumnNode *pCol = (SColumnNode *)pNode;
90✔
1595

1596
      if (pCol->tableId != suid) {
90✔
1597
        mInfo("stream:%s, check colId:%d passed", pStream->name, pCol->colId);
5!
1598
        goto NEXT;
5✔
1599
      }
1600
      if (pCol->colId > 0 && pCol->colId == colId) {
85!
1601
        code = TSDB_CODE_MND_STREAM_MUST_BE_DELETED;
20✔
1602
        mError("stream:%s, check colId:%d conflicted", pStream->name, pCol->colId);
20!
1603
        nodesDestroyNode(pAst);
20✔
1604
        nodesDestroyList(pNodeList);
20✔
1605
        sdbRelease(pSdb, pStream);
20✔
1606
        sdbCancelFetch(pSdb, pIter);
20✔
1607
        TAOS_RETURN(code);
20✔
1608
      }
1609
      mInfo("stream:%s, check colId:%d passed", pStream->name, pCol->colId);
65!
1610
    }
1611

1612
  NEXT:
5✔
1613
    sdbRelease(pSdb, pStream);
10✔
1614
    nodesDestroyNode(pAst);
10✔
1615
    nodesDestroyList(pNodeList);
10✔
1616
  }
1617
  TAOS_RETURN(code);
1,229✔
1618
}
1619

1620
static int32_t mndCheckAlterColForTSma(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {
1,229✔
1621
  int32_t code = 0;
1,229✔
1622
  SSdb   *pSdb = pMnode->pSdb;
1,229✔
1623
  void   *pIter = NULL;
1,229✔
1624
  while (1) {
10✔
1625
    SSmaObj *pSma = NULL;
1,239✔
1626
    pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);
1,239✔
1627
    if (pIter == NULL) break;
1,239✔
1628

1629
    mInfo("tsma:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d, sql:%s", pSma->name, stbFullName,
10!
1630
          suid, colId, pSma->sql);
1631

1632
    SNode *pAst = NULL;
10✔
1633
    if (nodesStringToNode(pSma->ast, &pAst) != 0) {
10!
1634
      code = TSDB_CODE_SDB_INVALID_DATA_CONTENT;
×
1635
      mError("tsma:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d failed since parse AST err",
×
1636
             pSma->name, stbFullName, suid, colId);
1637
      sdbCancelFetch(pSdb, pIter);
×
1638
      TAOS_RETURN(code);
×
1639
    }
1640

1641
    SNodeList *pNodeList = NULL;
10✔
1642
    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=
10!
1643
        0) {
1644
      sdbCancelFetch(pSdb, pIter);
×
1645
      TAOS_RETURN(code);
×
1646
    }
1647
    SNode *pNode = NULL;
10✔
1648
    FOREACH(pNode, pNodeList) {
55!
1649
      SColumnNode *pCol = (SColumnNode *)pNode;
50✔
1650
      mInfo("tsma:%s, check colId:%d tableId:%" PRId64, pSma->name, pCol->colId, pCol->tableId);
50!
1651

1652
      if ((pCol->tableId != suid) && (pSma->stbUid != suid)) {
50!
1653
        mInfo("tsma:%s, check colId:%d passed", pSma->name, pCol->colId);
5!
1654
        goto NEXT;
5✔
1655
      }
1656
      if ((pCol->colId) > 0 && (pCol->colId == colId)) {
45!
1657
        code = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TSMA;
×
1658
        mError("tsma:%s, check colId:%d conflicted", pSma->name, pCol->colId);
×
1659
        nodesDestroyNode(pAst);
×
1660
        nodesDestroyList(pNodeList);
×
1661
        sdbRelease(pSdb, pSma);
×
1662
        sdbCancelFetch(pSdb, pIter);
×
1663
        TAOS_RETURN(code);
×
1664
      }
1665
      mInfo("tsma:%s, check colId:%d passed", pSma->name, pCol->colId);
45!
1666
    }
1667

1668
  NEXT:
5✔
1669
    sdbRelease(pSdb, pSma);
10✔
1670
    nodesDestroyNode(pAst);
10✔
1671
    nodesDestroyList(pNodeList);
10✔
1672
  }
1673
  TAOS_RETURN(code);
1,229✔
1674
}
1675

1676
int32_t mndCheckColAndTagModifiable(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {
1,327✔
1677
  TAOS_CHECK_RETURN(mndCheckAlterColForTopic(pMnode, stbFullName, suid, colId));
1,327✔
1678
  TAOS_CHECK_RETURN(mndCheckAlterColForStream(pMnode, stbFullName, suid, colId));
1,249✔
1679
  TAOS_CHECK_RETURN(mndCheckAlterColForTSma(pMnode, stbFullName, suid, colId));
1,229!
1680
  TAOS_RETURN(0);
1,229✔
1681
}
1682

1683
static int32_t mndDropSuperTableTag(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const char *tagName) {
376✔
1684
  int32_t code = 0;
376✔
1685
  int32_t tag = mndFindSuperTableTagIndex(pOld, tagName);
376✔
1686
  if (tag < 0) {
376✔
1687
    code = TSDB_CODE_MND_TAG_NOT_EXIST;
4✔
1688
    TAOS_RETURN(code);
4✔
1689
  }
1690

1691
  col_id_t colId = pOld->pTags[tag].colId;
372✔
1692
  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));
372✔
1693

1694
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
344!
1695

1696
  memmove(pNew->pTags + tag, pNew->pTags + tag + 1, sizeof(SSchema) * (pNew->numOfTags - tag - 1));
344✔
1697
  pNew->numOfTags--;
344✔
1698

1699
  pNew->tagVer++;
344✔
1700

1701
  // if (mndDropIndexByTag(pMnode, pOld, tagName) != 0) {
1702
  //   return -1;
1703
  // }
1704
  mInfo("stb:%s, start to drop tag %s", pNew->name, tagName);
344!
1705
  TAOS_RETURN(code);
344✔
1706
}
1707

1708
static int32_t mndAlterStbTagName(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, SArray *pFields) {
679✔
1709
  int32_t code = 0;
679✔
1710
  if ((int32_t)taosArrayGetSize(pFields) != 2) {
679!
1711
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
1712
    TAOS_RETURN(code);
×
1713
  }
1714

1715
  SField *pField0 = taosArrayGet(pFields, 0);
679✔
1716
  SField *pField1 = taosArrayGet(pFields, 1);
679✔
1717

1718
  const char *oldTagName = pField0->name;
679✔
1719
  const char *newTagName = pField1->name;
679✔
1720

1721
  int32_t tag = mndFindSuperTableTagIndex(pOld, oldTagName);
679✔
1722
  if (tag < 0) {
679✔
1723
    code = TSDB_CODE_MND_TAG_NOT_EXIST;
9✔
1724
    TAOS_RETURN(code);
9✔
1725
  }
1726

1727
  col_id_t colId = pOld->pTags[tag].colId;
670✔
1728
  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));
670✔
1729

1730
  if (mndFindSuperTableTagIndex(pOld, newTagName) >= 0) {
648✔
1731
    code = TSDB_CODE_MND_TAG_ALREADY_EXIST;
11✔
1732
    TAOS_RETURN(code);
11✔
1733
  }
1734

1735
  if (mndFindSuperTableColumnIndex(pOld, newTagName) >= 0) {
637✔
1736
    code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
1✔
1737
    TAOS_RETURN(code);
1✔
1738
  }
1739

1740
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
636!
1741

1742
  SSchema *pSchema = (SSchema *)(pNew->pTags + tag);
636✔
1743
  memcpy(pSchema->name, newTagName, TSDB_COL_NAME_LEN);
636✔
1744

1745
  pNew->tagVer++;
636✔
1746
  mInfo("stb:%s, start to modify tag %s to %s", pNew->name, oldTagName, newTagName);
636!
1747
  TAOS_RETURN(code);
636✔
1748
}
1749

1750
static int32_t mndAlterStbTagBytes(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const SField *pField) {
42✔
1751
  int32_t code = 0;
42✔
1752
  int32_t tag = mndFindSuperTableTagIndex(pOld, pField->name);
42✔
1753
  if (tag < 0) {
42✔
1754
    code = TSDB_CODE_MND_TAG_NOT_EXIST;
1✔
1755
    TAOS_RETURN(code);
1✔
1756
  }
1757

1758
  col_id_t colId = pOld->pTags[tag].colId;
41✔
1759
  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));
41✔
1760

1761
  uint32_t nLen = 0;
32✔
1762
  for (int32_t i = 0; i < pOld->numOfTags; ++i) {
145✔
1763
    nLen += (pOld->pTags[i].colId == colId) ? pField->bytes : pOld->pTags[i].bytes;
113✔
1764
  }
1765

1766
  if (nLen > TSDB_MAX_TAGS_LEN) {
32!
1767
    code = TSDB_CODE_PAR_INVALID_TAGS_LENGTH;
×
1768
    TAOS_RETURN(code);
×
1769
  }
1770

1771
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
32!
1772

1773
  SSchema *pTag = pNew->pTags + tag;
32✔
1774

1775
  if (!(pTag->type == TSDB_DATA_TYPE_BINARY || pTag->type == TSDB_DATA_TYPE_VARBINARY ||
32!
1776
        pTag->type == TSDB_DATA_TYPE_NCHAR || pTag->type == TSDB_DATA_TYPE_GEOMETRY)) {
9!
1777
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
1✔
1778
    TAOS_RETURN(code);
1✔
1779
  }
1780

1781
  if (pField->bytes <= pTag->bytes) {
31✔
1782
    code = TSDB_CODE_MND_INVALID_ROW_BYTES;
1✔
1783
    TAOS_RETURN(code);
1✔
1784
  }
1785

1786
  pTag->bytes = pField->bytes;
30✔
1787
  pNew->tagVer++;
30✔
1788

1789
  mInfo("stb:%s, start to modify tag len %s to %d", pNew->name, pField->name, pField->bytes);
30!
1790
  TAOS_RETURN(code);
30✔
1791
}
1792

1793
static int32_t mndUpdateSuperTableColumnCompress(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, SArray *pField,
26✔
1794
                                                 int32_t nCols) {
1795
  // if (pColCmpr == NULL || colName == NULL) return -1;
1796

1797
  if (taosArrayGetSize(pField) != nCols) return TSDB_CODE_FAILED;
26!
1798
  TAOS_FIELD *p = taosArrayGet(pField, 0);
26✔
1799

1800
  int32_t code = 0;
26✔
1801
  int32_t idx = mndFindSuperTableColumnIndex(pOld, p->name);
26✔
1802
  if (idx == -1) {
26!
1803
    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;
×
1804
    TAOS_RETURN(code);
×
1805
  }
1806
  SSchema *pTarget = &pOld->pColumns[idx];
26✔
1807
  col_id_t colId = pTarget->colId;
26✔
1808

1809
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
26!
1810
  code = validColCmprByType(pTarget->type, p->bytes);
26✔
1811
  if (code != TSDB_CODE_SUCCESS) {
26✔
1812
    TAOS_RETURN(code);
1✔
1813
  }
1814

1815
  int8_t updated = 0;
25✔
1816
  for (int i = 0; i < pNew->numOfColumns; i++) {
161!
1817
    SColCmpr *pCmpr = &pNew->pCmpr[i];
161✔
1818
    if (pCmpr->id == colId) {
161✔
1819
      uint32_t dst = 0;
25✔
1820
      updated = tUpdateCompress(pCmpr->alg, p->bytes, TSDB_COLVAL_COMPRESS_DISABLED, TSDB_COLVAL_LEVEL_DISABLED,
25✔
1821
                                TSDB_COLVAL_LEVEL_MEDIUM, &dst);
1822
      if (updated > 0) pCmpr->alg = dst;
25✔
1823
      break;
25✔
1824
    }
1825
  }
1826

1827
  if (updated == 0) {
25✔
1828
    code = TSDB_CODE_MND_COLUMN_COMPRESS_ALREADY_EXIST;
7✔
1829
    TAOS_RETURN(code);
7✔
1830
  } else if (updated == -1) {
18!
1831
    code = TSDB_CODE_TSC_COMPRESS_LEVEL_ERROR;
×
1832
    TAOS_RETURN(code);
×
1833
  }
1834

1835
  pNew->colVer++;
18✔
1836

1837
  TAOS_RETURN(code);
18✔
1838
}
1839
static int32_t mndAddSuperTableColumn(const SStbObj *pOld, SStbObj *pNew, SArray *pFields, int32_t ncols,
1,186✔
1840
                                      int8_t withCompress) {
1841
  int32_t code = 0;
1,186✔
1842
  if (pOld->numOfColumns + ncols + pOld->numOfTags > TSDB_MAX_COLUMNS) {
1,186!
1843
    code = TSDB_CODE_MND_TOO_MANY_COLUMNS;
×
1844
    TAOS_RETURN(code);
×
1845
  }
1846

1847
  if ((code = grantCheck(TSDB_GRANT_TIMESERIES)) != 0) {
1,186!
1848
    TAOS_RETURN(code);
×
1849
  }
1850

1851
  if (!mndValidateSchema(pOld->pColumns, pOld->numOfColumns, pFields, TSDB_MAX_BYTES_PER_ROW)) {
1,186!
1852
    code = TSDB_CODE_PAR_INVALID_ROW_LENGTH;
×
1853
    TAOS_RETURN(code);
×
1854
  }
1855

1856
  pNew->numOfColumns = pNew->numOfColumns + ncols;
1,186✔
1857

1858
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
1,186!
1859

1860
  if (pNew->nextColId < 0 || pNew->nextColId >= 0x7fff - ncols) {
1,186!
1861
    code = TSDB_CODE_OUT_OF_RANGE;
×
1862
    TAOS_RETURN(code);
×
1863
  }
1864

1865
  for (int32_t i = 0; i < ncols; i++) {
2,355✔
1866
    if (withCompress) {
1,186✔
1867
      SFieldWithOptions *pField = taosArrayGet(pFields, i);
2✔
1868
      if (mndFindSuperTableColumnIndex(pOld, pField->name) >= 0) {
2!
1869
        code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
×
1870
        TAOS_RETURN(code);
×
1871
      }
1872

1873
      if (mndFindSuperTableTagIndex(pOld, pField->name) >= 0) {
2!
1874
        code = TSDB_CODE_MND_TAG_ALREADY_EXIST;
×
1875
        TAOS_RETURN(code);
×
1876
      }
1877

1878
      SSchema *pSchema = &pNew->pColumns[pOld->numOfColumns + i];
2✔
1879
      pSchema->bytes = pField->bytes;
2✔
1880
      pSchema->type = pField->type;
2✔
1881
      memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
2✔
1882
      pSchema->colId = pNew->nextColId;
2✔
1883
      pNew->nextColId++;
2✔
1884

1885
      SColCmpr *pCmpr = &pNew->pCmpr[pOld->numOfColumns + i];
2✔
1886
      pCmpr->id = pSchema->colId;
2✔
1887
      pCmpr->alg = pField->compress;
2✔
1888
      mInfo("stb:%s, start to add column %s", pNew->name, pSchema->name);
2!
1889
    } else {
1890
      SField *pField = taosArrayGet(pFields, i);
1,184✔
1891
      if (mndFindSuperTableColumnIndex(pOld, pField->name) >= 0) {
1,184✔
1892
        code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
10✔
1893
        TAOS_RETURN(code);
10✔
1894
      }
1895

1896
      if (mndFindSuperTableTagIndex(pOld, pField->name) >= 0) {
1,174✔
1897
        code = TSDB_CODE_MND_TAG_ALREADY_EXIST;
7✔
1898
        TAOS_RETURN(code);
7✔
1899
      }
1900

1901
      SSchema *pSchema = &pNew->pColumns[pOld->numOfColumns + i];
1,167✔
1902
      pSchema->bytes = pField->bytes;
1,167✔
1903
      pSchema->type = pField->type;
1,167✔
1904
      memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
1,167✔
1905
      pSchema->colId = pNew->nextColId;
1,167✔
1906
      pNew->nextColId++;
1,167✔
1907

1908
      SColCmpr *pCmpr = &pNew->pCmpr[pOld->numOfColumns + i];
1,167✔
1909
      pCmpr->id = pSchema->colId;
1,167✔
1910
      pCmpr->alg = createDefaultColCmprByType(pSchema->type);
1,167✔
1911
      mInfo("stb:%s, start to add column %s", pNew->name, pSchema->name);
1,167!
1912
    }
1913
  }
1914

1915
  pNew->colVer++;
1,169✔
1916
  TAOS_RETURN(code);
1,169✔
1917
}
1918

1919
static int32_t mndDropSuperTableColumn(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const char *colName) {
199✔
1920
  int32_t code = 0;
199✔
1921
  int32_t col = mndFindSuperTableColumnIndex(pOld, colName);
199✔
1922
  if (col < 0) {
199✔
1923
    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;
7✔
1924
    TAOS_RETURN(code);
7✔
1925
  }
1926

1927
  if (col == 0) {
192✔
1928
    code = TSDB_CODE_MND_INVALID_STB_ALTER_OPTION;
5✔
1929
    TAOS_RETURN(code);
5✔
1930
  }
1931

1932
  if (pOld->numOfColumns == 2) {
187✔
1933
    code = TSDB_CODE_PAR_INVALID_DROP_COL;
1✔
1934
    TAOS_RETURN(code);
1✔
1935
  }
1936

1937
  col_id_t colId = pOld->pColumns[col].colId;
186✔
1938
  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));
186✔
1939

1940
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
159!
1941

1942
  int32_t sz = pNew->numOfColumns - col - 1;
159✔
1943
  memmove(pNew->pColumns + col, pNew->pColumns + col + 1, sizeof(SSchema) * sz);
159✔
1944
  memmove(pNew->pCmpr + col, pNew->pCmpr + col + 1, sizeof(SColCmpr) * sz);
159✔
1945
  pNew->numOfColumns--;
159✔
1946

1947
  pNew->colVer++;
159✔
1948
  mInfo("stb:%s, start to drop col %s", pNew->name, colName);
159!
1949
  TAOS_RETURN(code);
159✔
1950
}
1951

1952
static int32_t mndAlterStbColumnBytes(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const SField *pField) {
62✔
1953
  int32_t code = 0;
62✔
1954
  int32_t col = mndFindSuperTableColumnIndex(pOld, pField->name);
62✔
1955
  if (col < 0) {
62✔
1956
    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;
3✔
1957
    TAOS_RETURN(code);
3✔
1958
  }
1959

1960
  col_id_t colId = pOld->pColumns[col].colId;
59✔
1961

1962
  uint32_t nLen = 0;
59✔
1963
  for (int32_t i = 0; i < pOld->numOfColumns; ++i) {
336✔
1964
    nLen += (pOld->pColumns[i].colId == colId) ? pField->bytes : pOld->pColumns[i].bytes;
277✔
1965
  }
1966

1967
  if (nLen > TSDB_MAX_BYTES_PER_ROW) {
59✔
1968
    code = TSDB_CODE_MND_INVALID_ROW_BYTES;
1✔
1969
    TAOS_RETURN(code);
1✔
1970
  }
1971

1972
  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));
58✔
1973

1974
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
46!
1975

1976
  SSchema *pCol = pNew->pColumns + col;
46✔
1977
  if (!(pCol->type == TSDB_DATA_TYPE_BINARY || pCol->type == TSDB_DATA_TYPE_VARBINARY ||
46✔
1978
        pCol->type == TSDB_DATA_TYPE_NCHAR || pCol->type == TSDB_DATA_TYPE_GEOMETRY)) {
9!
1979
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
1✔
1980
    TAOS_RETURN(code);
1✔
1981
  }
1982

1983
  if (pField->bytes <= pCol->bytes) {
45✔
1984
    code = TSDB_CODE_MND_INVALID_ROW_BYTES;
1✔
1985
    TAOS_RETURN(code);
1✔
1986
  }
1987

1988
  pCol->bytes = pField->bytes;
44✔
1989
  pNew->colVer++;
44✔
1990

1991
  mInfo("stb:%s, start to modify col len %s to %d", pNew->name, pField->name, pField->bytes);
44!
1992
  TAOS_RETURN(code);
44✔
1993
}
1994

1995
static int32_t mndSetAlterStbPrepareLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
2,954✔
1996
  int32_t  code = 0;
2,954✔
1997
  SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);
2,954✔
1998
  if (pRedoRaw == NULL) {
2,954!
1999
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2000
    if (terrno != 0) code = terrno;
×
2001
    TAOS_RETURN(code);
×
2002
  }
2003
  if ((code = mndTransAppendPrepareLog(pTrans, pRedoRaw)) != 0) {
2,954!
2004
    sdbFreeRaw(pRedoRaw);
×
2005
    TAOS_RETURN(code);
×
2006
  }
2007
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY));
2,954!
2008

2009
  TAOS_RETURN(code);
2,954✔
2010
}
2011

2012
static int32_t mndSetAlterStbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
2,954✔
2013
  int32_t  code = 0;
2,954✔
2014
  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);
2,954✔
2015
  if (pCommitRaw == NULL) {
2,954!
2016
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2017
    if (terrno != 0) code = terrno;
×
2018
    TAOS_RETURN(code);
×
2019
  }
2020
  if ((code = mndTransAppendCommitlog(pTrans, pCommitRaw)) != 0) {
2,954!
2021
    sdbFreeRaw(pCommitRaw);
×
2022
    TAOS_RETURN(code);
×
2023
  }
2024
  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));
2,954!
2025

2026
  TAOS_RETURN(code);
2,954✔
2027
}
2028

2029
static int32_t mndSetAlterStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb, void *alterOriData,
2,954✔
2030
                                         int32_t alterOriDataLen) {
2031
  int32_t code = 0;
2,954✔
2032
  SSdb   *pSdb = pMnode->pSdb;
2,954✔
2033
  SVgObj *pVgroup = NULL;
2,954✔
2034
  void   *pIter = NULL;
2,954✔
2035
  int32_t contLen;
2036

2037
  while (1) {
9,747✔
2038
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
12,701✔
2039
    if (pIter == NULL) break;
12,701✔
2040
    if (!mndVgroupInDb(pVgroup, pDb->uid)) {
9,747✔
2041
      sdbRelease(pSdb, pVgroup);
3,521✔
2042
      continue;
3,521✔
2043
    }
2044

2045
    void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, alterOriData, alterOriDataLen);
6,226✔
2046
    if (pReq == NULL) {
6,226!
2047
      sdbCancelFetch(pSdb, pIter);
×
2048
      sdbRelease(pSdb, pVgroup);
×
2049
      code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2050
      if (terrno != 0) code = terrno;
×
2051
      TAOS_RETURN(code);
×
2052
    }
2053
    STransAction action = {0};
6,226✔
2054
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
6,226✔
2055
    action.pCont = pReq;
6,226✔
2056
    action.contLen = contLen;
6,226✔
2057
    action.msgType = TDMT_VND_ALTER_STB;
6,226✔
2058
    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
6,226!
2059
      taosMemoryFree(pReq);
×
2060
      sdbCancelFetch(pSdb, pIter);
×
2061
      sdbRelease(pSdb, pVgroup);
×
2062
      TAOS_RETURN(code);
×
2063
    }
2064
    sdbRelease(pSdb, pVgroup);
6,226✔
2065
  }
2066

2067
  TAOS_RETURN(code);
2,954✔
2068
}
2069

2070
static int32_t mndSetAlterStbRedoActions2(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb,
×
2071
                                          void *alterOriData, int32_t alterOriDataLen) {
2072
  int32_t code = 0;
×
2073
  SSdb   *pSdb = pMnode->pSdb;
×
2074
  SVgObj *pVgroup = NULL;
×
2075
  void   *pIter = NULL;
×
2076
  int32_t contLen;
2077

2078
  while (1) {
×
2079
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
×
2080
    if (pIter == NULL) break;
×
2081
    if (!mndVgroupInDb(pVgroup, pDb->uid)) {
×
2082
      sdbRelease(pSdb, pVgroup);
×
2083
      continue;
×
2084
    }
2085

2086
    void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, alterOriData, alterOriDataLen);
×
2087
    if (pReq == NULL) {
×
2088
      sdbCancelFetch(pSdb, pIter);
×
2089
      sdbRelease(pSdb, pVgroup);
×
2090
      code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2091
      if (terrno != 0) code = terrno;
×
2092
      TAOS_RETURN(code);
×
2093
    }
2094
    STransAction action = {0};
×
2095
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
×
2096
    action.pCont = pReq;
×
2097
    action.contLen = contLen;
×
2098
    action.msgType = TDMT_VND_CREATE_INDEX;
×
2099
    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
×
2100
      taosMemoryFree(pReq);
×
2101
      sdbCancelFetch(pSdb, pIter);
×
2102
      sdbRelease(pSdb, pVgroup);
×
2103
      TAOS_RETURN(code);
×
2104
    }
2105
    sdbRelease(pSdb, pVgroup);
×
2106
  }
2107

2108
  TAOS_RETURN(code);
×
2109
}
2110
static int32_t mndBuildStbSchemaImp(SDbObj *pDb, SStbObj *pStb, const char *tbName, STableMetaRsp *pRsp) {
830,352✔
2111
  int32_t code = 0;
830,352✔
2112
  taosRLockLatch(&pStb->lock);
830,352✔
2113

2114
  int32_t totalCols = pStb->numOfColumns + pStb->numOfTags;
830,365✔
2115
  pRsp->pSchemas = taosMemoryCalloc(totalCols, sizeof(SSchema));
830,365!
2116
  if (pRsp->pSchemas == NULL) {
830,351!
2117
    taosRUnLockLatch(&pStb->lock);
×
2118
    code = terrno;
×
2119
    TAOS_RETURN(code);
×
2120
  }
2121
  pRsp->pSchemaExt = taosMemoryCalloc(pStb->numOfColumns, sizeof(SSchemaExt));
830,351!
2122
  if (pRsp->pSchemaExt == NULL) {
830,356!
2123
    taosRUnLockLatch(&pStb->lock);
×
2124
    code = terrno;
×
2125
    TAOS_RETURN(code);
×
2126
  }
2127

2128
  tstrncpy(pRsp->dbFName, pStb->db, sizeof(pRsp->dbFName));
830,356✔
2129
  tstrncpy(pRsp->tbName, tbName, sizeof(pRsp->tbName));
830,356✔
2130
  tstrncpy(pRsp->stbName, tbName, sizeof(pRsp->stbName));
830,356✔
2131
  pRsp->dbId = pDb->uid;
830,356✔
2132
  pRsp->numOfTags = pStb->numOfTags;
830,356✔
2133
  pRsp->numOfColumns = pStb->numOfColumns;
830,356✔
2134
  pRsp->precision = pDb->cfg.precision;
830,356✔
2135
  pRsp->tableType = TSDB_SUPER_TABLE;
830,356✔
2136
  pRsp->sversion = pStb->colVer;
830,356✔
2137
  pRsp->tversion = pStb->tagVer;
830,356✔
2138
  pRsp->suid = pStb->uid;
830,356✔
2139
  pRsp->tuid = pStb->uid;
830,356✔
2140

2141
  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
22,775,917✔
2142
    SSchema *pSchema = &pRsp->pSchemas[i];
21,945,561✔
2143
    SSchema *pSrcSchema = &pStb->pColumns[i];
21,945,561✔
2144
    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
21,945,561✔
2145
    pSchema->type = pSrcSchema->type;
21,945,561✔
2146
    pSchema->flags = pSrcSchema->flags;
21,945,561✔
2147
    pSchema->colId = pSrcSchema->colId;
21,945,561✔
2148
    pSchema->bytes = pSrcSchema->bytes;
21,945,561✔
2149
  }
2150

2151
  for (int32_t i = 0; i < pStb->numOfTags; ++i) {
13,179,526✔
2152
    SSchema *pSchema = &pRsp->pSchemas[i + pStb->numOfColumns];
12,349,170✔
2153
    SSchema *pSrcSchema = &pStb->pTags[i];
12,349,170✔
2154
    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
12,349,170✔
2155
    pSchema->type = pSrcSchema->type;
12,349,170✔
2156
    pSchema->flags = pSrcSchema->flags;
12,349,170✔
2157
    pSchema->colId = pSrcSchema->colId;
12,349,170✔
2158
    pSchema->bytes = pSrcSchema->bytes;
12,349,170✔
2159
  }
2160
  for (int32_t i = 0; i < pStb->numOfColumns; i++) {
22,776,639✔
2161
    SColCmpr   *pCmpr = &pStb->pCmpr[i];
21,946,283✔
2162
    SSchemaExt *pSchEx = &pRsp->pSchemaExt[i];
21,946,283✔
2163
    pSchEx->colId = pCmpr->id;
21,946,283✔
2164
    pSchEx->compress = pCmpr->alg;
21,946,283✔
2165
  }
2166

2167
  taosRUnLockLatch(&pStb->lock);
830,356✔
2168
  TAOS_RETURN(code);
830,363✔
2169
}
2170

2171
static int32_t mndBuildStbCfgImp(SDbObj *pDb, SStbObj *pStb, const char *tbName, STableCfgRsp *pRsp) {
31✔
2172
  int32_t code = 0;
31✔
2173
  taosRLockLatch(&pStb->lock);
31✔
2174

2175
  int32_t totalCols = pStb->numOfColumns + pStb->numOfTags;
31✔
2176
  pRsp->pSchemas = taosMemoryCalloc(totalCols, sizeof(SSchema));
31!
2177
  if (pRsp->pSchemas == NULL) {
31!
2178
    taosRUnLockLatch(&pStb->lock);
×
2179
    code = terrno;
×
2180
    TAOS_RETURN(code);
×
2181
  }
2182

2183
  tstrncpy(pRsp->dbFName, pStb->db, sizeof(pRsp->dbFName));
31✔
2184
  tstrncpy(pRsp->tbName, tbName, sizeof(pRsp->tbName));
31✔
2185
  tstrncpy(pRsp->stbName, tbName, sizeof(pRsp->stbName));
31✔
2186
  pRsp->numOfTags = pStb->numOfTags;
31✔
2187
  pRsp->numOfColumns = pStb->numOfColumns;
31✔
2188
  pRsp->tableType = TSDB_SUPER_TABLE;
31✔
2189
  pRsp->delay1 = pStb->maxdelay[0];
31✔
2190
  pRsp->delay2 = pStb->maxdelay[1];
31✔
2191
  pRsp->watermark1 = pStb->watermark[0];
31✔
2192
  pRsp->watermark2 = pStb->watermark[1];
31✔
2193
  pRsp->ttl = pStb->ttl;
31✔
2194
  pRsp->commentLen = pStb->commentLen;
31✔
2195
  if (pStb->commentLen > 0) {
31!
2196
    pRsp->pComment = taosStrdup(pStb->comment);
×
2197
  }
2198

2199
  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
149✔
2200
    SSchema *pSchema = &pRsp->pSchemas[i];
118✔
2201
    SSchema *pSrcSchema = &pStb->pColumns[i];
118✔
2202
    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
118✔
2203
    pSchema->type = pSrcSchema->type;
118✔
2204
    pSchema->flags = pSrcSchema->flags;
118✔
2205
    pSchema->colId = pSrcSchema->colId;
118✔
2206
    pSchema->bytes = pSrcSchema->bytes;
118✔
2207
  }
2208

2209
  for (int32_t i = 0; i < pStb->numOfTags; ++i) {
90✔
2210
    SSchema *pSchema = &pRsp->pSchemas[i + pStb->numOfColumns];
59✔
2211
    SSchema *pSrcSchema = &pStb->pTags[i];
59✔
2212
    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
59✔
2213
    pSchema->type = pSrcSchema->type;
59✔
2214
    pSchema->flags = pSrcSchema->flags;
59✔
2215
    pSchema->colId = pSrcSchema->colId;
59✔
2216
    pSchema->bytes = pSrcSchema->bytes;
59✔
2217
  }
2218

2219
  if (pStb->numOfFuncs > 0) {
31!
2220
    pRsp->pFuncs = taosArrayDup(pStb->pFuncs, NULL);
×
2221
  }
2222

2223
  pRsp->pSchemaExt = taosMemoryCalloc(pStb->numOfColumns, sizeof(SSchemaExt));
31!
2224
  for (int32_t i = 0; i < pStb->numOfColumns; i++) {
149✔
2225
    SColCmpr *pCmpr = &pStb->pCmpr[i];
118✔
2226

2227
    SSchemaExt *pSchExt = &pRsp->pSchemaExt[i];
118✔
2228
    pSchExt->colId = pCmpr->id;
118✔
2229
    pSchExt->compress = pCmpr->alg;
118✔
2230
  }
2231

2232
  taosRUnLockLatch(&pStb->lock);
31✔
2233
  TAOS_RETURN(code);
31✔
2234
}
2235

2236
static int32_t mndValidateStbVersion(SMnode *pMnode, SSTableVersion *pStbVer, bool *schema, bool *sma) {
79,915✔
2237
  int32_t code = 0;
79,915✔
2238
  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};
79,915✔
2239
  snprintf(tbFName, sizeof(tbFName), "%s.%s", pStbVer->dbFName, pStbVer->stbName);
79,915✔
2240

2241
  SDbObj *pDb = mndAcquireDb(pMnode, pStbVer->dbFName);
79,915✔
2242
  if (pDb == NULL) {
79,915✔
2243
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
352✔
2244
    TAOS_RETURN(code);
352✔
2245
  }
2246

2247
  if (pDb->uid != pStbVer->dbId) {
79,563✔
2248
    mndReleaseDb(pMnode, pDb);
184✔
2249
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
184✔
2250
    TAOS_RETURN(code);
184✔
2251
  }
2252

2253
  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);
79,379✔
2254
  if (pStb == NULL) {
79,379✔
2255
    mndReleaseDb(pMnode, pDb);
159✔
2256
    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
159✔
2257
    TAOS_RETURN(code);
159✔
2258
  }
2259

2260
  taosRLockLatch(&pStb->lock);
79,220✔
2261

2262
  if (pStbVer->sversion != pStb->colVer || pStbVer->tversion != pStb->tagVer) {
79,220✔
2263
    *schema = true;
321✔
2264
  } else {
2265
    *schema = false;
78,899✔
2266
  }
2267

2268
  if (pStbVer->smaVer && pStbVer->smaVer != pStb->smaVer) {
79,220!
2269
    *sma = true;
×
2270
  } else {
2271
    *sma = false;
79,220✔
2272
  }
2273

2274
  taosRUnLockLatch(&pStb->lock);
79,220✔
2275

2276
  mndReleaseDb(pMnode, pDb);
79,219✔
2277
  mndReleaseStb(pMnode, pStb);
79,220✔
2278
  return TSDB_CODE_SUCCESS;
79,220✔
2279
}
2280

2281
static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp) {
821,675✔
2282
  int32_t code = 0;
821,675✔
2283
  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};
821,675✔
2284
  snprintf(tbFName, sizeof(tbFName), "%s.%s", dbFName, tbName);
821,675✔
2285

2286
  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);
821,675✔
2287
  if (pDb == NULL) {
821,666!
2288
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
2289
    TAOS_RETURN(code);
×
2290
  }
2291

2292
  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);
821,666✔
2293
  if (pStb == NULL) {
821,666✔
2294
    mndReleaseDb(pMnode, pDb);
1,760✔
2295
    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
1,760✔
2296
    TAOS_RETURN(code);
1,760✔
2297
  }
2298

2299
  code = mndBuildStbSchemaImp(pDb, pStb, tbName, pRsp);
819,906✔
2300
  mndReleaseDb(pMnode, pDb);
819,910✔
2301
  mndReleaseStb(pMnode, pStb);
819,915✔
2302
  TAOS_RETURN(code);
819,915✔
2303
}
2304

2305
static int32_t mndBuildStbCfg(SMnode *pMnode, const char *dbFName, const char *tbName, STableCfgRsp *pRsp) {
31✔
2306
  int32_t code = 0;
31✔
2307
  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};
31✔
2308
  snprintf(tbFName, sizeof(tbFName), "%s.%s", dbFName, tbName);
31✔
2309

2310
  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);
31✔
2311
  if (pDb == NULL) {
31!
2312
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
2313
    TAOS_RETURN(code);
×
2314
  }
2315

2316
  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);
31✔
2317
  if (pStb == NULL) {
31!
2318
    mndReleaseDb(pMnode, pDb);
×
2319
    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
×
2320
    TAOS_RETURN(code);
×
2321
  }
2322

2323
  code = mndBuildStbCfgImp(pDb, pStb, tbName, pRsp);
31✔
2324

2325
  mndReleaseDb(pMnode, pDb);
31✔
2326
  mndReleaseStb(pMnode, pStb);
31✔
2327
  TAOS_RETURN(code);
31✔
2328
}
2329

2330
static int32_t mndBuildSMAlterStbRsp(SDbObj *pDb, SStbObj *pObj, void **pCont, int32_t *pLen) {
2,839✔
2331
  int32_t       code = 0;
2,839✔
2332
  SEncoder      ec = {0};
2,839✔
2333
  uint32_t      contLen = 0;
2,839✔
2334
  SMAlterStbRsp alterRsp = {0};
2,839✔
2335
  SName         name = {0};
2,839✔
2336
  TAOS_CHECK_RETURN(tNameFromString(&name, pObj->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
2,839!
2337

2338
  alterRsp.pMeta = taosMemoryCalloc(1, sizeof(STableMetaRsp));
2,839!
2339
  if (NULL == alterRsp.pMeta) {
2,839!
2340
    code = terrno;
×
2341
    TAOS_RETURN(code);
×
2342
  }
2343

2344
  code = mndBuildStbSchemaImp(pDb, pObj, name.tname, alterRsp.pMeta);
2,839✔
2345
  if (code) {
2,839!
2346
    tFreeSMAlterStbRsp(&alterRsp);
×
2347
    return code;
×
2348
  }
2349

2350
  tEncodeSize(tEncodeSMAlterStbRsp, &alterRsp, contLen, code);
2,839!
2351
  if (code) {
2,839!
2352
    tFreeSMAlterStbRsp(&alterRsp);
×
2353
    return code;
×
2354
  }
2355

2356
  void *cont = taosMemoryMalloc(contLen);
2,839!
2357
  if (NULL == cont) {
2,839!
2358
    code = terrno;
×
2359
    tFreeSMAlterStbRsp(&alterRsp);
×
2360
    TAOS_RETURN(code);
×
2361
  }
2362
  tEncoderInit(&ec, cont, contLen);
2,839✔
2363
  code = tEncodeSMAlterStbRsp(&ec, &alterRsp);
2,839✔
2364
  tEncoderClear(&ec);
2,839✔
2365

2366
  tFreeSMAlterStbRsp(&alterRsp);
2,839✔
2367

2368
  if (code < 0) TAOS_RETURN(code);
2,839!
2369

2370
  *pCont = cont;
2,839✔
2371
  *pLen = contLen;
2,839✔
2372

2373
  TAOS_RETURN(code);
2,839✔
2374
}
2375

2376
int32_t mndBuildSMCreateStbRsp(SMnode *pMnode, char *dbFName, char *stbFName, void **pCont, int32_t *pLen) {
7,615✔
2377
  int32_t code = -1;
7,615✔
2378
  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);
7,615✔
2379
  if (NULL == pDb) {
7,615!
2380
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2381
    if (terrno != 0) code = terrno;
×
2382
    TAOS_RETURN(code);
×
2383
  }
2384

2385
  SStbObj *pObj = mndAcquireStb(pMnode, stbFName);
7,615✔
2386
  if (NULL == pObj) {
7,615✔
2387
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
3✔
2388
    if (terrno != 0) code = terrno;
3!
2389
    goto _OVER;
3✔
2390
  }
2391

2392
  SEncoder       ec = {0};
7,612✔
2393
  uint32_t       contLen = 0;
7,612✔
2394
  SMCreateStbRsp stbRsp = {0};
7,612✔
2395
  SName          name = {0};
7,612✔
2396
  TAOS_CHECK_GOTO(tNameFromString(&name, pObj->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE), NULL, _OVER);
7,612!
2397

2398
  stbRsp.pMeta = taosMemoryCalloc(1, sizeof(STableMetaRsp));
7,612!
2399
  if (NULL == stbRsp.pMeta) {
7,612!
2400
    code = terrno;
×
2401
    goto _OVER;
×
2402
  }
2403

2404
  code = mndBuildStbSchemaImp(pDb, pObj, name.tname, stbRsp.pMeta);
7,612✔
2405
  if (code) {
7,612!
2406
    tFreeSMCreateStbRsp(&stbRsp);
×
2407
    goto _OVER;
×
2408
  }
2409

2410
  tEncodeSize(tEncodeSMCreateStbRsp, &stbRsp, contLen, code);
7,612!
2411
  if (code) {
7,612!
2412
    tFreeSMCreateStbRsp(&stbRsp);
×
2413
    goto _OVER;
×
2414
  }
2415

2416
  void *cont = taosMemoryMalloc(contLen);
7,612!
2417
  if (NULL == cont) {
7,612!
2418
    code = terrno;
×
2419
    tFreeSMCreateStbRsp(&stbRsp);
×
2420
    goto _OVER;
×
2421
  }
2422
  tEncoderInit(&ec, cont, contLen);
7,612✔
2423
  TAOS_CHECK_GOTO(tEncodeSMCreateStbRsp(&ec, &stbRsp), NULL, _OVER);
7,612!
2424
  tEncoderClear(&ec);
7,612✔
2425

2426
  tFreeSMCreateStbRsp(&stbRsp);
7,612✔
2427

2428
  *pCont = cont;
7,612✔
2429
  *pLen = contLen;
7,612✔
2430

2431
  code = 0;
7,612✔
2432

2433
_OVER:
7,615✔
2434
  if (pObj) {
7,615✔
2435
    mndReleaseStb(pMnode, pObj);
7,612✔
2436
  }
2437

2438
  if (pDb) {
7,615!
2439
    mndReleaseDb(pMnode, pDb);
7,615✔
2440
  }
2441

2442
  TAOS_RETURN(code);
7,615✔
2443
}
2444

2445
static int32_t mndAlterStbImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp,
1,974✔
2446
                              void *alterOriData, int32_t alterOriDataLen) {
2447
  int32_t code = -1;
1,974✔
2448
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "alter-stb");
1,974✔
2449
  if (pTrans == NULL) {
1,974!
2450
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2451
    if (terrno != 0) code = terrno;
×
2452
    goto _OVER;
×
2453
  }
2454

2455
  mInfo("trans:%d, used to alter stb:%s", pTrans->id, pStb->name);
1,974!
2456
  mndTransSetDbName(pTrans, pDb->name, pStb->name);
1,974✔
2457
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
1,974!
2458

2459
  if (needRsp) {
1,974✔
2460
    void   *pCont = NULL;
1,859✔
2461
    int32_t contLen = 0;
1,859✔
2462
    TAOS_CHECK_GOTO(mndBuildSMAlterStbRsp(pDb, pStb, &pCont, &contLen), NULL, _OVER);
1,859!
2463
    mndTransSetRpcRsp(pTrans, pCont, contLen);
1,859✔
2464
  }
2465

2466
  TAOS_CHECK_GOTO(mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
1,974!
2467
  TAOS_CHECK_GOTO(mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
1,974!
2468
  TAOS_CHECK_GOTO(mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb, alterOriData, alterOriDataLen), NULL, _OVER);
1,974!
2469
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
1,974!
2470

2471
  code = 0;
1,974✔
2472

2473
_OVER:
1,974✔
2474
  mndTransDrop(pTrans);
1,974✔
2475
  TAOS_RETURN(code);
1,974✔
2476
}
2477

2478
static int32_t mndAlterStbAndUpdateTagIdxImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp,
980✔
2479
                                             void *alterOriData, int32_t alterOriDataLen, const SMAlterStbReq *pAlter) {
2480
  int32_t code = -1;
980✔
2481
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "alter-stb");
980✔
2482
  if (pTrans == NULL) {
980!
2483
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2484
    if (terrno != 0) code = terrno;
×
2485
    goto _OVER;
×
2486
  }
2487

2488
  mInfo("trans:%d, used to alter stb:%s", pTrans->id, pStb->name);
980!
2489
  mndTransSetDbName(pTrans, pDb->name, pStb->name);
980✔
2490

2491
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
980!
2492

2493
  if (needRsp) {
980!
2494
    void   *pCont = NULL;
980✔
2495
    int32_t contLen = 0;
980✔
2496
    TAOS_CHECK_GOTO(mndBuildSMAlterStbRsp(pDb, pStb, &pCont, &contLen), NULL, _OVER);
980!
2497
    mndTransSetRpcRsp(pTrans, pCont, contLen);
980✔
2498
  }
2499

2500
  if (pAlter->alterType == TSDB_ALTER_TABLE_DROP_TAG) {
980✔
2501
    SIdxObj idxObj = {0};
344✔
2502
    SField *pField0 = taosArrayGet(pAlter->pFields, 0);
344✔
2503
    bool    exist = false;
344✔
2504
    if (mndGetIdxsByTagName(pMnode, pStb, pField0->name, &idxObj) == 0) {
344✔
2505
      exist = true;
6✔
2506
    }
2507
    TAOS_CHECK_GOTO(mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
344!
2508
    TAOS_CHECK_GOTO(mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
344!
2509

2510
    if (exist == true) {
344✔
2511
      TAOS_CHECK_GOTO(mndSetDropIdxPrepareLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
6!
2512
      TAOS_CHECK_GOTO(mndSetDropIdxCommitLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
6!
2513
    }
2514

2515
    TAOS_CHECK_GOTO(mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb, alterOriData, alterOriDataLen), NULL, _OVER);
344!
2516
    TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
344!
2517

2518
  } else if (pAlter->alterType == TSDB_ALTER_TABLE_UPDATE_TAG_NAME) {
636!
2519
    SIdxObj     idxObj = {0};
636✔
2520
    SField     *pField0 = taosArrayGet(pAlter->pFields, 0);
636✔
2521
    SField     *pField1 = taosArrayGet(pAlter->pFields, 1);
636✔
2522
    const char *oTagName = pField0->name;
636✔
2523
    const char *nTagName = pField1->name;
636✔
2524
    bool        exist = false;
636✔
2525

2526
    if (mndGetIdxsByTagName(pMnode, pStb, pField0->name, &idxObj) == 0) {
636✔
2527
      exist = true;
33✔
2528
    }
2529

2530
    TAOS_CHECK_GOTO(mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
636!
2531
    TAOS_CHECK_GOTO(mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
636!
2532

2533
    if (exist == true) {
636✔
2534
      memcpy(idxObj.colName, nTagName, strlen(nTagName));
33✔
2535
      idxObj.colName[strlen(nTagName)] = 0;
33✔
2536
      TAOS_CHECK_GOTO(mndSetAlterIdxPrepareLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
33!
2537
      TAOS_CHECK_GOTO(mndSetAlterIdxCommitLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
33!
2538
    }
2539

2540
    TAOS_CHECK_GOTO(mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb, alterOriData, alterOriDataLen), NULL, _OVER);
636!
2541
    TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
636!
2542
  }
2543
  code = 0;
980✔
2544

2545
_OVER:
980✔
2546
  mndTransDrop(pTrans);
980✔
2547
  TAOS_RETURN(code);
980✔
2548
}
2549

2550
static int32_t mndAlterStb(SMnode *pMnode, SRpcMsg *pReq, const SMAlterStbReq *pAlter, SDbObj *pDb, SStbObj *pOld) {
3,045✔
2551
  bool    needRsp = true;
3,045✔
2552
  int32_t code = -1;
3,045✔
2553
  SField *pField0 = NULL;
3,045✔
2554

2555
  SStbObj stbObj = {0};
3,045✔
2556
  taosRLockLatch(&pOld->lock);
3,045✔
2557
  memcpy(&stbObj, pOld, sizeof(SStbObj));
3,045✔
2558
  taosRUnLockLatch(&pOld->lock);
3,045✔
2559
  stbObj.pColumns = NULL;
3,045✔
2560
  stbObj.pTags = NULL;
3,045✔
2561
  stbObj.pFuncs = NULL;
3,045✔
2562
  stbObj.pCmpr = NULL;
3,045✔
2563
  stbObj.updateTime = taosGetTimestampMs();
3,045✔
2564
  stbObj.lock = 0;
3,045✔
2565
  bool updateTagIndex = false;
3,045✔
2566
  switch (pAlter->alterType) {
3,045!
2567
    case TSDB_ALTER_TABLE_ADD_TAG:
453✔
2568
      code = mndAddSuperTableTag(pOld, &stbObj, pAlter->pFields, pAlter->numOfFields);
453✔
2569
      break;
453✔
2570
    case TSDB_ALTER_TABLE_DROP_TAG:
376✔
2571
      pField0 = taosArrayGet(pAlter->pFields, 0);
376✔
2572
      code = mndDropSuperTableTag(pMnode, pOld, &stbObj, pField0->name);
376✔
2573
      updateTagIndex = true;
376✔
2574
      break;
376✔
2575
    case TSDB_ALTER_TABLE_UPDATE_TAG_NAME:
679✔
2576
      code = mndAlterStbTagName(pMnode, pOld, &stbObj, pAlter->pFields);
679✔
2577
      updateTagIndex = true;
679✔
2578
      break;
679✔
2579
    case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES:
42✔
2580
      pField0 = taosArrayGet(pAlter->pFields, 0);
42✔
2581
      code = mndAlterStbTagBytes(pMnode, pOld, &stbObj, pField0);
42✔
2582
      break;
42✔
2583
    case TSDB_ALTER_TABLE_ADD_COLUMN:
1,184✔
2584
      code = mndAddSuperTableColumn(pOld, &stbObj, pAlter->pFields, pAlter->numOfFields, 0);
1,184✔
2585
      break;
1,184✔
2586
    case TSDB_ALTER_TABLE_DROP_COLUMN:
199✔
2587
      pField0 = taosArrayGet(pAlter->pFields, 0);
199✔
2588
      code = mndDropSuperTableColumn(pMnode, pOld, &stbObj, pField0->name);
199✔
2589
      break;
199✔
2590
    case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES:
62✔
2591
      pField0 = taosArrayGet(pAlter->pFields, 0);
62✔
2592
      code = mndAlterStbColumnBytes(pMnode, pOld, &stbObj, pField0);
62✔
2593
      break;
62✔
2594
    case TSDB_ALTER_TABLE_UPDATE_OPTIONS:
22✔
2595
      needRsp = false;
22✔
2596
      code = mndUpdateStbCommentAndTTL(pOld, &stbObj, pAlter->comment, pAlter->commentLen, pAlter->ttl);
22✔
2597
      break;
22✔
2598
    case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS:
26✔
2599
      code = mndUpdateSuperTableColumnCompress(pMnode, pOld, &stbObj, pAlter->pFields, pAlter->numOfFields);
26✔
2600
      break;
26✔
2601
    case TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COMPRESS_OPTION:
2✔
2602
      code = mndAddSuperTableColumn(pOld, &stbObj, pAlter->pFields, pAlter->numOfFields, 1);
2✔
2603
      break;
2✔
2604
    default:
×
2605
      needRsp = false;
×
2606
      terrno = TSDB_CODE_OPS_NOT_SUPPORT;
×
2607
      break;
×
2608
  }
2609

2610
  if (code != 0) goto _OVER;
3,045✔
2611
  if (updateTagIndex == false) {
2,861✔
2612
    code = mndAlterStbImp(pMnode, pReq, pDb, &stbObj, needRsp, pReq->pCont, pReq->contLen);
1,881✔
2613
  } else {
2614
    code = mndAlterStbAndUpdateTagIdxImp(pMnode, pReq, pDb, &stbObj, needRsp, pReq->pCont, pReq->contLen, pAlter);
980✔
2615
  }
2616

2617
_OVER:
3,045✔
2618
  taosMemoryFreeClear(stbObj.pTags);
3,045!
2619
  taosMemoryFreeClear(stbObj.pColumns);
3,045!
2620
  taosMemoryFreeClear(stbObj.pCmpr);
3,045!
2621
  if (pAlter->commentLen > 0) {
3,045✔
2622
    taosMemoryFreeClear(stbObj.comment);
17!
2623
  }
2624
  TAOS_RETURN(code);
3,045✔
2625
}
2626

2627
static int32_t mndProcessAlterStbReq(SRpcMsg *pReq) {
3,049✔
2628
  SMnode       *pMnode = pReq->info.node;
3,049✔
2629
  int32_t       code = -1;
3,049✔
2630
  SDbObj       *pDb = NULL;
3,049✔
2631
  SStbObj      *pStb = NULL;
3,049✔
2632
  SMAlterStbReq alterReq = {0};
3,049✔
2633

2634
  if (tDeserializeSMAlterStbReq(pReq->pCont, pReq->contLen, &alterReq) != 0) {
3,049!
2635
    code = TSDB_CODE_INVALID_MSG;
×
2636
    goto _OVER;
×
2637
  }
2638

2639
  mInfo("stb:%s, start to alter", alterReq.name);
3,049!
2640
  if (mndCheckAlterStbReq(&alterReq) != 0) goto _OVER;
3,049!
2641

2642
  pDb = mndAcquireDbByStb(pMnode, alterReq.name);
3,049✔
2643
  if (pDb == NULL) {
3,049✔
2644
    code = TSDB_CODE_MND_DB_NOT_EXIST;
2✔
2645
    goto _OVER;
2✔
2646
  }
2647

2648
  pStb = mndAcquireStb(pMnode, alterReq.name);
3,047✔
2649
  if (pStb == NULL) {
3,047✔
2650
    code = TSDB_CODE_MND_STB_NOT_EXIST;
2✔
2651
    goto _OVER;
2✔
2652
  }
2653

2654
  if ((code = mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pDb)) != 0) {
3,045!
2655
    goto _OVER;
×
2656
  }
2657

2658
  code = mndAlterStb(pMnode, pReq, &alterReq, pDb, pStb);
3,045✔
2659
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
3,045✔
2660

2661
  SName   name = {0};
3,045✔
2662
  int32_t ret = 0;
3,045✔
2663
  if ((ret = tNameFromString(&name, alterReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0)
3,045!
2664
    mError("stb:%s, failed to tNameFromString since %s", alterReq.name, tstrerror(ret));
×
2665

2666
  auditRecord(pReq, pMnode->clusterId, "alterStb", name.dbname, name.tname, alterReq.sql, alterReq.sqlLen);
3,045✔
2667

2668
_OVER:
3,049✔
2669
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
3,049!
2670
    mError("stb:%s, failed to alter since %s", alterReq.name, tstrerror(code));
188!
2671
  }
2672

2673
  mndReleaseStb(pMnode, pStb);
3,049✔
2674
  mndReleaseDb(pMnode, pDb);
3,049✔
2675
  tFreeSMAltertbReq(&alterReq);
3,049✔
2676

2677
  TAOS_RETURN(code);
3,049✔
2678
}
2679

2680
static int32_t mndSetDropStbPrepareLogs(SMnode *pMnode, STrans *pTrans, SStbObj *pStb) {
955✔
2681
  int32_t  code = 0;
955✔
2682
  SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);
955✔
2683
  if (pRedoRaw == NULL) {
955!
2684
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2685
    if (terrno != 0) code = terrno;
×
2686
    TAOS_RETURN(code);
×
2687
  }
2688
  if ((code = mndTransAppendPrepareLog(pTrans, pRedoRaw)) != 0) {
955!
2689
    sdbFreeRaw(pRedoRaw);
×
2690
    TAOS_RETURN(code);
×
2691
  }
2692
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING));
955!
2693

2694
  TAOS_RETURN(code);
955✔
2695
}
2696

2697
static int32_t mndSetDropStbCommitLogs(SMnode *pMnode, STrans *pTrans, SStbObj *pStb) {
955✔
2698
  int32_t  code = 0;
955✔
2699
  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);
955✔
2700
  if (pCommitRaw == NULL) {
955!
2701
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2702
    if (terrno != 0) code = terrno;
×
2703
    TAOS_RETURN(code);
×
2704
  }
2705
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
955!
2706
    sdbFreeRaw(pCommitRaw);
×
2707
    TAOS_RETURN(code);
×
2708
  }
2709
  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));
955!
2710

2711
  TAOS_RETURN(code);
955✔
2712
}
2713

2714
static int32_t mndSetDropStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
955✔
2715
  int32_t code = 0;
955✔
2716
  SSdb   *pSdb = pMnode->pSdb;
955✔
2717
  SVgObj *pVgroup = NULL;
955✔
2718
  void   *pIter = NULL;
955✔
2719

2720
  while (1) {
4,123✔
2721
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
5,078✔
2722
    if (pIter == NULL) break;
5,078✔
2723
    if (!mndVgroupInDb(pVgroup, pDb->uid)) {
4,123✔
2724
      sdbRelease(pSdb, pVgroup);
1,833✔
2725
      continue;
1,833✔
2726
    }
2727

2728
    int32_t contLen = 0;
2,290✔
2729
    void   *pReq = mndBuildVDropStbReq(pMnode, pVgroup, pStb, &contLen);
2,290✔
2730
    if (pReq == NULL) {
2,290!
2731
      sdbCancelFetch(pSdb, pIter);
×
2732
      sdbRelease(pSdb, pVgroup);
×
2733
      code = TSDB_CODE_OUT_OF_MEMORY;
×
2734
      TAOS_RETURN(code);
×
2735
    }
2736

2737
    STransAction action = {0};
2,290✔
2738
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
2,290✔
2739
    action.pCont = pReq;
2,290✔
2740
    action.contLen = contLen;
2,290✔
2741
    action.msgType = TDMT_VND_DROP_STB;
2,290✔
2742
    action.acceptableCode = TSDB_CODE_TDB_STB_NOT_EXIST;
2,290✔
2743
    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
2,290!
2744
      taosMemoryFree(pReq);
×
2745
      sdbCancelFetch(pSdb, pIter);
×
2746
      sdbRelease(pSdb, pVgroup);
×
2747
      TAOS_RETURN(code);
×
2748
    }
2749
    sdbRelease(pSdb, pVgroup);
2,290✔
2750
  }
2751

2752
  TAOS_RETURN(code);
955✔
2753
}
2754

2755
static int32_t mndDropStb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb) {
955✔
2756
  int32_t code = -1;
955✔
2757
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "drop-stb");
955✔
2758
  if (pTrans == NULL) {
955!
2759
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2760
    if (terrno != 0) code = terrno;
×
2761
    goto _OVER;
×
2762
  }
2763

2764
  mInfo("trans:%d, used to drop stb:%s", pTrans->id, pStb->name);
955!
2765
  mndTransSetDbName(pTrans, pDb->name, pStb->name);
955✔
2766
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
955!
2767

2768
  TAOS_CHECK_GOTO(mndSetDropStbPrepareLogs(pMnode, pTrans, pStb), NULL, _OVER);
955!
2769
  TAOS_CHECK_GOTO(mndSetDropStbCommitLogs(pMnode, pTrans, pStb), NULL, _OVER);
955!
2770
  TAOS_CHECK_GOTO(mndSetDropStbRedoActions(pMnode, pTrans, pDb, pStb), NULL, _OVER);
955!
2771
  TAOS_CHECK_GOTO(mndDropIdxsByStb(pMnode, pTrans, pDb, pStb), NULL, _OVER);
955!
2772
  TAOS_CHECK_GOTO(mndDropSmasByStb(pMnode, pTrans, pDb, pStb), NULL, _OVER);
955!
2773
  TAOS_CHECK_GOTO(mndUserRemoveStb(pMnode, pTrans, pStb->name), NULL, _OVER);
955!
2774
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
955!
2775
  code = 0;
955✔
2776

2777
_OVER:
955✔
2778
  mndTransDrop(pTrans);
955✔
2779
  TAOS_RETURN(code);
955✔
2780
}
2781

2782
static int32_t mndCheckDropStbForTopic(SMnode *pMnode, const char *stbFullName, int64_t suid) {
959✔
2783
  int32_t code = 0;
959✔
2784
  SSdb   *pSdb = pMnode->pSdb;
959✔
2785
  void   *pIter = NULL;
959✔
2786
  while (1) {
80✔
2787
    SMqTopicObj *pTopic = NULL;
1,039✔
2788
    pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);
1,039✔
2789
    if (pIter == NULL) break;
1,039✔
2790

2791
    if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {
80✔
2792
      if (pTopic->stbUid == suid) {
14!
2793
        sdbRelease(pSdb, pTopic);
×
2794
        sdbCancelFetch(pSdb, pIter);
×
2795
        TAOS_RETURN(-1);
×
2796
      }
2797
    }
2798

2799
    if (pTopic->ast == NULL) {
80✔
2800
      sdbRelease(pSdb, pTopic);
38✔
2801
      continue;
38✔
2802
    }
2803

2804
    SNode *pAst = NULL;
42✔
2805
    if (nodesStringToNode(pTopic->ast, &pAst) != 0) {
42!
2806
      code = TSDB_CODE_MND_INVALID_TOPIC_OPTION;
×
2807
      mError("topic:%s, create ast error", pTopic->name);
×
2808
      sdbRelease(pSdb, pTopic);
×
2809
      sdbCancelFetch(pSdb, pIter);
×
2810
      TAOS_RETURN(code);
×
2811
    }
2812

2813
    SNodeList *pNodeList = NULL;
42✔
2814
    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=
42!
2815
        0) {
2816
      sdbRelease(pSdb, pTopic);
×
2817
      sdbCancelFetch(pSdb, pIter);
×
2818
      TAOS_RETURN(code);
×
2819
    }
2820
    SNode *pNode = NULL;
42✔
2821
    FOREACH(pNode, pNodeList) {
42✔
2822
      SColumnNode *pCol = (SColumnNode *)pNode;
28✔
2823

2824
      if (pCol->tableId == suid) {
28!
2825
        sdbRelease(pSdb, pTopic);
×
2826
        nodesDestroyNode(pAst);
×
2827
        nodesDestroyList(pNodeList);
×
2828
        sdbCancelFetch(pSdb, pIter);
×
2829
        TAOS_RETURN(-1);
×
2830
      } else {
2831
        goto NEXT;
28✔
2832
      }
2833
    }
2834
  NEXT:
14✔
2835
    sdbRelease(pSdb, pTopic);
42✔
2836
    nodesDestroyNode(pAst);
42✔
2837
    nodesDestroyList(pNodeList);
42✔
2838
  }
2839
  TAOS_RETURN(code);
959✔
2840
}
2841

2842
static int32_t mndCheckDropStbForStream(SMnode *pMnode, const char *stbFullName, int64_t suid) {
959✔
2843
  int32_t code = 0;
959✔
2844
  SSdb   *pSdb = pMnode->pSdb;
959✔
2845
  void   *pIter = NULL;
959✔
2846
  while (1) {
146✔
2847
    SStreamObj *pStream = NULL;
1,105✔
2848
    pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);
1,105✔
2849
    if (pIter == NULL) break;
1,105✔
2850

2851
    if (pStream->targetStbUid == suid) {
150!
2852
      sdbCancelFetch(pSdb, pIter);
×
2853
      sdbRelease(pSdb, pStream);
×
2854
      TAOS_RETURN(-1);
4✔
2855
    }
2856

2857
    SNode *pAst = NULL;
150✔
2858
    if (nodesStringToNode(pStream->ast, &pAst) != 0) {
150!
2859
      code = TSDB_CODE_MND_INVALID_STREAM_OPTION;
×
2860
      mError("stream:%s, create ast error", pStream->name);
×
2861
      sdbCancelFetch(pSdb, pIter);
×
2862
      sdbRelease(pSdb, pStream);
×
2863
      TAOS_RETURN(code);
×
2864
    }
2865

2866
    SNodeList *pNodeList = NULL;
150✔
2867
    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=
150!
2868
        0) {
2869
      sdbCancelFetch(pSdb, pIter);
×
2870
      sdbRelease(pSdb, pStream);
×
2871
      TAOS_RETURN(code);
×
2872
    }
2873
    SNode *pNode = NULL;
150✔
2874
    FOREACH(pNode, pNodeList) {
150!
2875
      SColumnNode *pCol = (SColumnNode *)pNode;
150✔
2876

2877
      if (pCol->tableId == suid) {
150✔
2878
        sdbCancelFetch(pSdb, pIter);
4✔
2879
        sdbRelease(pSdb, pStream);
4✔
2880
        nodesDestroyNode(pAst);
4✔
2881
        nodesDestroyList(pNodeList);
4✔
2882
        TAOS_RETURN(-1);
4✔
2883
      } else {
2884
        goto NEXT;
146✔
2885
      }
2886
    }
2887
  NEXT:
×
2888
    sdbRelease(pSdb, pStream);
146✔
2889
    nodesDestroyNode(pAst);
146✔
2890
    nodesDestroyList(pNodeList);
146✔
2891
  }
2892
  TAOS_RETURN(code);
955✔
2893
}
2894

2895
static int32_t mndProcessDropTtltbRsp(SRpcMsg *pRsp) { return 0; }
×
2896
static int32_t mndProcessTrimDbRsp(SRpcMsg *pRsp) { return 0; }
202✔
2897
static int32_t mndProcessS3MigrateDbRsp(SRpcMsg *pRsp) { return 0; }
×
2898

2899
static int32_t mndProcessDropStbReq(SRpcMsg *pReq) {
1,248✔
2900
  SMnode      *pMnode = pReq->info.node;
1,248✔
2901
  int32_t      code = -1;
1,248✔
2902
  SDbObj      *pDb = NULL;
1,248✔
2903
  SStbObj     *pStb = NULL;
1,248✔
2904
  SMDropStbReq dropReq = {0};
1,248✔
2905

2906
  TAOS_CHECK_GOTO(tDeserializeSMDropStbReq(pReq->pCont, pReq->contLen, &dropReq), NULL, _OVER);
1,248!
2907

2908
  mInfo("stb:%s, start to drop", dropReq.name);
1,248!
2909

2910
  pStb = mndAcquireStb(pMnode, dropReq.name);
1,248✔
2911
  if (pStb == NULL) {
1,248✔
2912
    if (dropReq.igNotExists) {
289✔
2913
      mInfo("stb:%s, not exist, ignore not exist is set", dropReq.name);
287!
2914
      code = 0;
287✔
2915
      goto _OVER;
287✔
2916
    } else {
2917
      code = TSDB_CODE_MND_STB_NOT_EXIST;
2✔
2918
      goto _OVER;
2✔
2919
    }
2920
  }
2921

2922
  if ((dropReq.source == TD_REQ_FROM_TAOX_OLD || dropReq.source == TD_REQ_FROM_TAOX) && pStb->uid != dropReq.suid) {
959!
2923
    code = 0;
×
2924
    goto _OVER;
×
2925
  }
2926

2927
  pDb = mndAcquireDbByStb(pMnode, dropReq.name);
959✔
2928
  if (pDb == NULL) {
959!
2929
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
2930
    goto _OVER;
×
2931
  }
2932

2933
  if ((code = mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pDb)) != 0) {
959!
2934
    goto _OVER;
×
2935
  }
2936

2937
  if (mndCheckDropStbForTopic(pMnode, dropReq.name, pStb->uid) < 0) {
959!
2938
    code = TSDB_CODE_MND_TOPIC_MUST_BE_DELETED;
×
2939
    goto _OVER;
×
2940
  }
2941

2942
  if (mndCheckDropStbForStream(pMnode, dropReq.name, pStb->uid) < 0) {
959✔
2943
    code = TSDB_CODE_MND_STREAM_MUST_BE_DELETED;
4✔
2944
    goto _OVER;
4✔
2945
  }
2946

2947
  code = mndDropStb(pMnode, pReq, pDb, pStb);
955✔
2948
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
955!
2949

2950
  SName   name = {0};
955✔
2951
  int32_t ret = 0;
955✔
2952
  if ((ret = tNameFromString(&name, dropReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0)
955!
2953
    mError("stb:%s, failed to tNameFromString since %s", dropReq.name, tstrerror(ret));
×
2954

2955
  auditRecord(pReq, pMnode->clusterId, "dropStb", name.dbname, name.tname, dropReq.sql, dropReq.sqlLen);
955✔
2956

2957
_OVER:
1,248✔
2958
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
1,248✔
2959
    mError("stb:%s, failed to drop since %s", dropReq.name, tstrerror(code));
6!
2960
  }
2961

2962
  mndReleaseDb(pMnode, pDb);
1,248✔
2963
  mndReleaseStb(pMnode, pStb);
1,248✔
2964
  tFreeSMDropStbReq(&dropReq);
1,248✔
2965
  TAOS_RETURN(code);
1,248✔
2966
}
2967

2968
static int32_t mndProcessTableMetaReq(SRpcMsg *pReq) {
955,103✔
2969
  SMnode       *pMnode = pReq->info.node;
955,103✔
2970
  int32_t       code = -1;
955,103✔
2971
  STableInfoReq infoReq = {0};
955,103✔
2972
  STableMetaRsp metaRsp = {0};
955,103✔
2973
  SUserObj     *pUser = NULL;
955,103✔
2974

2975
  code = mndAcquireUser(pMnode, pReq->info.conn.user, &pUser);
955,103✔
2976
  if (pUser == NULL) return 0;
955,149!
2977
  bool sysinfo = pUser->sysInfo;
955,149✔
2978

2979
  TAOS_CHECK_GOTO(tDeserializeSTableInfoReq(pReq->pCont, pReq->contLen, &infoReq), NULL, _OVER);
955,149!
2980

2981
  if (0 == strcmp(infoReq.dbFName, TSDB_INFORMATION_SCHEMA_DB)) {
955,137✔
2982
    mInfo("information_schema table:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);
107,203!
2983
    TAOS_CHECK_GOTO(mndBuildInsTableSchema(pMnode, infoReq.dbFName, infoReq.tbName, sysinfo, &metaRsp), NULL, _OVER);
107,203✔
2984
  } else if (0 == strcmp(infoReq.dbFName, TSDB_PERFORMANCE_SCHEMA_DB)) {
847,934✔
2985
    mInfo("performance_schema table:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);
26,600!
2986
    TAOS_CHECK_GOTO(mndBuildPerfsTableSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp), NULL, _OVER);
26,600✔
2987
  } else {
2988
    mInfo("stb:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);
821,334✔
2989
    TAOS_CHECK_GOTO(mndBuildStbSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp), NULL, _OVER);
821,384✔
2990
  }
2991

2992
  int32_t rspLen = tSerializeSTableMetaRsp(NULL, 0, &metaRsp);
953,173✔
2993
  if (rspLen < 0) {
953,151!
2994
    code = TSDB_CODE_INVALID_MSG;
×
2995
    goto _OVER;
×
2996
  }
2997

2998
  void *pRsp = rpcMallocCont(rspLen);
953,151✔
2999
  if (pRsp == NULL) {
953,130!
3000
    code = terrno;
×
3001
    goto _OVER;
×
3002
  }
3003

3004
  if ((rspLen = tSerializeSTableMetaRsp(pRsp, rspLen, &metaRsp)) < 0) {
953,130!
3005
    code = rspLen;
×
3006
    goto _OVER;
×
3007
  }
3008
  pReq->info.rsp = pRsp;
953,155✔
3009
  pReq->info.rspLen = rspLen;
953,155✔
3010
  code = 0;
953,155✔
3011

3012
  mTrace("%s.%s, meta is retrieved", infoReq.dbFName, infoReq.tbName);
953,155✔
3013

3014
_OVER:
951,520✔
3015
  if (code != 0) {
955,139✔
3016
    mError("stb:%s.%s, failed to retrieve meta since %s", infoReq.dbFName, infoReq.tbName, tstrerror(code));
1,984!
3017
  }
3018

3019
  mndReleaseUser(pMnode, pUser);
955,139✔
3020
  tFreeSTableMetaRsp(&metaRsp);
955,156✔
3021
  // TODO change to TAOS_RETURN
3022
  return code;
955,153✔
3023
}
3024

3025
static int32_t mndProcessTableCfgReq(SRpcMsg *pReq) {
31✔
3026
  SMnode      *pMnode = pReq->info.node;
31✔
3027
  int32_t      code = -1;
31✔
3028
  STableCfgReq cfgReq = {0};
31✔
3029
  STableCfgRsp cfgRsp = {0};
31✔
3030

3031
  TAOS_CHECK_GOTO(tDeserializeSTableCfgReq(pReq->pCont, pReq->contLen, &cfgReq), NULL, _OVER);
31!
3032

3033
  char dbName[TSDB_DB_NAME_LEN] = {0};
31✔
3034
  TAOS_CHECK_GOTO(mndExtractShortDbNameFromDbFullName(cfgReq.dbFName, dbName), NULL, _OVER);
31!
3035
  if (0 == strcmp(dbName, TSDB_INFORMATION_SCHEMA_DB)) {
31!
3036
    mInfo("information_schema table:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
×
3037
    TAOS_CHECK_GOTO(mndBuildInsTableCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp), NULL, _OVER);
×
3038
  } else if (0 == strcmp(dbName, TSDB_PERFORMANCE_SCHEMA_DB)) {
31!
3039
    mInfo("performance_schema table:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
×
3040
    TAOS_CHECK_GOTO(mndBuildPerfsTableCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp), NULL, _OVER);
×
3041
  } else {
3042
    mInfo("stb:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
31!
3043
    TAOS_CHECK_GOTO(mndBuildStbCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp), NULL, _OVER);
31!
3044
  }
3045

3046
  int32_t rspLen = tSerializeSTableCfgRsp(NULL, 0, &cfgRsp);
31✔
3047
  if (rspLen < 0) {
31!
3048
    code = TSDB_CODE_INVALID_MSG;
×
3049
    goto _OVER;
×
3050
  }
3051

3052
  void *pRsp = rpcMallocCont(rspLen);
31✔
3053
  if (pRsp == NULL) {
31!
3054
    code = terrno;
×
3055
    goto _OVER;
×
3056
  }
3057

3058
  if ((rspLen = tSerializeSTableCfgRsp(pRsp, rspLen, &cfgRsp)) < 0) {
31!
3059
    code = rspLen;
×
3060
    goto _OVER;
×
3061
  }
3062
  pReq->info.rsp = pRsp;
31✔
3063
  pReq->info.rspLen = rspLen;
31✔
3064
  code = 0;
31✔
3065

3066
  mTrace("%s.%s, cfg is retrieved", cfgReq.dbFName, cfgReq.tbName);
31✔
3067

3068
_OVER:
26✔
3069
  if (code != 0) {
31!
3070
    mError("stb:%s.%s, failed to retrieve cfg since %s", cfgReq.dbFName, cfgReq.tbName, tstrerror(code));
×
3071
  }
3072

3073
  tFreeSTableCfgRsp(&cfgRsp);
31✔
3074
  TAOS_RETURN(code);
31✔
3075
}
3076

3077
int32_t mndValidateStbInfo(SMnode *pMnode, SSTableVersion *pStbVersions, int32_t numOfStbs, void **ppRsp,
64,249✔
3078
                           int32_t *pRspLen) {
3079
  int32_t   code = 0;
64,249✔
3080
  SSTbHbRsp hbRsp = {0};
64,249✔
3081
  hbRsp.pMetaRsp = taosArrayInit(numOfStbs, sizeof(STableMetaRsp));
64,249✔
3082
  if (hbRsp.pMetaRsp == NULL) {
64,248!
3083
    code = terrno;
×
3084
    TAOS_RETURN(code);
×
3085
  }
3086

3087
  hbRsp.pIndexRsp = taosArrayInit(numOfStbs, sizeof(STableIndexRsp));
64,248✔
3088
  if (NULL == hbRsp.pIndexRsp) {
64,249!
3089
    taosArrayDestroy(hbRsp.pMetaRsp);
×
3090
    code = terrno;
×
3091
    TAOS_RETURN(code);
×
3092
  }
3093

3094
  for (int32_t i = 0; i < numOfStbs; ++i) {
144,164✔
3095
    SSTableVersion *pStbVersion = &pStbVersions[i];
79,915✔
3096
    pStbVersion->suid = be64toh(pStbVersion->suid);
79,915✔
3097
    pStbVersion->sversion = ntohl(pStbVersion->sversion);
79,915✔
3098
    pStbVersion->tversion = ntohl(pStbVersion->tversion);
79,915✔
3099
    pStbVersion->smaVer = ntohl(pStbVersion->smaVer);
79,915✔
3100

3101
    bool    schema = false;
79,915✔
3102
    bool    sma = false;
79,915✔
3103
    int32_t code = mndValidateStbVersion(pMnode, pStbVersion, &schema, &sma);
79,915✔
3104
    if (TSDB_CODE_SUCCESS != code) {
79,915✔
3105
      STableMetaRsp metaRsp = {0};
695✔
3106
      metaRsp.numOfColumns = -1;
695✔
3107
      metaRsp.suid = pStbVersion->suid;
695✔
3108
      tstrncpy(metaRsp.dbFName, pStbVersion->dbFName, sizeof(metaRsp.dbFName));
695✔
3109
      tstrncpy(metaRsp.tbName, pStbVersion->stbName, sizeof(metaRsp.tbName));
695✔
3110
      tstrncpy(metaRsp.stbName, pStbVersion->stbName, sizeof(metaRsp.stbName));
695✔
3111
      if (taosArrayPush(hbRsp.pMetaRsp, &metaRsp) == NULL) {
1,390!
3112
        code = terrno;
×
3113
        return code;
×
3114
      }
3115
      continue;
695✔
3116
    }
3117

3118
    if (schema) {
79,220✔
3119
      STableMetaRsp metaRsp = {0};
321✔
3120
      mInfo("stb:%s.%s, start to retrieve meta", pStbVersion->dbFName, pStbVersion->stbName);
321!
3121
      if (mndBuildStbSchema(pMnode, pStbVersion->dbFName, pStbVersion->stbName, &metaRsp) != 0) {
321!
3122
        metaRsp.numOfColumns = -1;
×
3123
        metaRsp.suid = pStbVersion->suid;
×
3124
        tstrncpy(metaRsp.dbFName, pStbVersion->dbFName, sizeof(metaRsp.dbFName));
×
3125
        tstrncpy(metaRsp.tbName, pStbVersion->stbName, sizeof(metaRsp.tbName));
×
3126
        tstrncpy(metaRsp.stbName, pStbVersion->stbName, sizeof(metaRsp.stbName));
×
3127
        if (taosArrayPush(hbRsp.pMetaRsp, &metaRsp) == NULL) {
×
3128
          code = terrno;
×
3129
          return code;
×
3130
        }
3131
        continue;
×
3132
      }
3133

3134
      if (taosArrayPush(hbRsp.pMetaRsp, &metaRsp) == NULL) {
642!
3135
        code = terrno;
×
3136
        return code;
×
3137
      }
3138
    }
3139

3140
    if (sma) {
79,220!
3141
      bool           exist = false;
×
3142
      char           tbFName[TSDB_TABLE_FNAME_LEN];
3143
      STableIndexRsp indexRsp = {0};
×
3144
      indexRsp.pIndex = taosArrayInit(10, sizeof(STableIndexInfo));
×
3145
      if (NULL == indexRsp.pIndex) {
×
3146
        code = terrno;
×
3147
        TAOS_RETURN(code);
×
3148
      }
3149

3150
      (void)tsnprintf(tbFName, sizeof(tbFName), "%s.%s", pStbVersion->dbFName, pStbVersion->stbName);
×
3151
      int32_t code = mndGetTableSma(pMnode, tbFName, &indexRsp, &exist);
×
3152
      if (code || !exist) {
×
3153
        indexRsp.suid = pStbVersion->suid;
×
3154
        indexRsp.version = -1;
×
3155
        indexRsp.pIndex = NULL;
×
3156
      }
3157

3158
      tstrncpy(indexRsp.dbFName, pStbVersion->dbFName, sizeof(indexRsp.dbFName));
×
3159
      tstrncpy(indexRsp.tbName, pStbVersion->stbName, sizeof(indexRsp.tbName));
×
3160

3161
      if (taosArrayPush(hbRsp.pIndexRsp, &indexRsp) == NULL) {
×
3162
        code = terrno;
×
3163
        return code;
×
3164
      }
3165
    }
3166
  }
3167

3168
  int32_t rspLen = tSerializeSSTbHbRsp(NULL, 0, &hbRsp);
64,249✔
3169
  if (rspLen < 0) {
64,249!
3170
    tFreeSSTbHbRsp(&hbRsp);
×
3171
    code = TSDB_CODE_INVALID_MSG;
×
3172
    TAOS_RETURN(code);
×
3173
  }
3174

3175
  void *pRsp = taosMemoryMalloc(rspLen);
64,249!
3176
  if (pRsp == NULL) {
64,249!
3177
    tFreeSSTbHbRsp(&hbRsp);
×
3178
    code = terrno;
×
3179
    TAOS_RETURN(code);
×
3180
  }
3181

3182
  rspLen = tSerializeSSTbHbRsp(pRsp, rspLen, &hbRsp);
64,249✔
3183
  tFreeSSTbHbRsp(&hbRsp);
64,249✔
3184
  if (rspLen < 0) return rspLen;
64,249!
3185
  *ppRsp = pRsp;
64,249✔
3186
  *pRspLen = rspLen;
64,249✔
3187
  TAOS_RETURN(code);
64,249✔
3188
}
3189

3190
int32_t mndGetNumOfStbs(SMnode *pMnode, char *dbName, int32_t *pNumOfStbs) {
8,738✔
3191
  int32_t code = 0;
8,738✔
3192
  SSdb   *pSdb = pMnode->pSdb;
8,738✔
3193
  SDbObj *pDb = mndAcquireDb(pMnode, dbName);
8,738✔
3194
  if (pDb == NULL) {
8,738!
3195
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
3196
    TAOS_RETURN(code);
×
3197
  }
3198

3199
  int32_t numOfStbs = 0;
8,738✔
3200
  void   *pIter = NULL;
8,738✔
3201
  while (1) {
178,805✔
3202
    SStbObj *pStb = NULL;
187,543✔
3203
    pIter = sdbFetch(pSdb, SDB_STB, pIter, (void **)&pStb);
187,543✔
3204
    if (pIter == NULL) break;
187,543✔
3205

3206
    if (pStb->dbUid == pDb->uid) {
178,805✔
3207
      numOfStbs++;
86,486✔
3208
    }
3209

3210
    sdbRelease(pSdb, pStb);
178,805✔
3211
  }
3212

3213
  *pNumOfStbs = numOfStbs;
8,738✔
3214
  mndReleaseDb(pMnode, pDb);
8,738✔
3215
  TAOS_RETURN(code);
8,738✔
3216
}
3217

3218
int32_t mndExtractDbNameFromStbFullName(const char *stbFullName, char *dst) {
×
3219
  SName name = {0};
×
3220
  TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
×
3221

3222
  TAOS_CHECK_RETURN(tNameGetFullDbName(&name, dst));
×
3223

3224
  return 0;
×
3225
}
3226

3227
int32_t mndExtractShortDbNameFromStbFullName(const char *stbFullName, char *dst) {
2✔
3228
  SName name = {0};
2✔
3229
  TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
2!
3230

3231
  TAOS_CHECK_RETURN(tNameGetDbName(&name, dst));
2!
3232

3233
  return 0;
2✔
3234
}
3235

3236
int32_t mndExtractShortDbNameFromDbFullName(const char *stbFullName, char *dst) {
31✔
3237
  SName name = {0};
31✔
3238
  TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB));
31!
3239

3240
  TAOS_CHECK_RETURN(tNameGetDbName(&name, dst));
31!
3241

3242
  return 0;
31✔
3243
}
3244

3245
void mndExtractTbNameFromStbFullName(const char *stbFullName, char *dst, int32_t dstSize) {
1,554,926✔
3246
  int32_t pos = -1;
1,554,926✔
3247
  int32_t num = 0;
1,554,926✔
3248
  for (pos = 0; stbFullName[pos] != 0; ++pos) {
27,196,982✔
3249
    if (stbFullName[pos] == TS_PATH_DELIMITER[0]) num++;
27,196,927✔
3250
    if (num == 2) break;
27,196,927✔
3251
  }
3252

3253
  if (num == 2) {
1,554,926✔
3254
    tstrncpy(dst, stbFullName + pos + 1, dstSize);
1,554,834✔
3255
  }
3256
}
1,554,926✔
3257

3258
static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
16,360✔
3259
  SMnode  *pMnode = pReq->info.node;
16,360✔
3260
  SSdb    *pSdb = pMnode->pSdb;
16,360✔
3261
  int32_t  numOfRows = 0;
16,360✔
3262
  SStbObj *pStb = NULL;
16,360✔
3263
  int32_t  cols = 0;
16,360✔
3264
  int32_t  lino = 0;
16,360✔
3265
  int32_t  code = 0;
16,360✔
3266

3267
  SDbObj *pDb = NULL;
16,360✔
3268
  if (strlen(pShow->db) > 0) {
16,360✔
3269
    pDb = mndAcquireDb(pMnode, pShow->db);
414✔
3270
    if (pDb == NULL) return terrno;
414✔
3271
  }
3272

3273
  while (numOfRows < rows) {
871,904✔
3274
    pShow->pIter = sdbFetch(pSdb, SDB_STB, pShow->pIter, (void **)&pStb);
863,959✔
3275
    if (pShow->pIter == NULL) break;
863,594✔
3276

3277
    if (pDb != NULL && pStb->dbUid != pDb->uid) {
855,162✔
3278
      sdbRelease(pSdb, pStb);
775✔
3279
      continue;
787✔
3280
    }
3281

3282
    if (isTsmaResSTb(pStb->name)) {
854,387✔
3283
      sdbRelease(pSdb, pStb);
12✔
3284
      continue;
12✔
3285
    }
3286

3287
    cols = 0;
854,263✔
3288

3289
    SName name = {0};
854,263✔
3290

3291
    char stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
854,263✔
3292
    mndExtractTbNameFromStbFullName(pStb->name, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN);
854,263✔
3293
    varDataSetLen(stbName, strlen(&stbName[VARSTR_HEADER_SIZE]));
854,370✔
3294
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
854,370✔
3295
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbName, false), pStb, &lino, _ERROR);
853,146!
3296

3297
    char db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
850,337✔
3298
    RETRIEVE_CHECK_GOTO(tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB), pStb, &lino, _ERROR);
850,337!
3299
    RETRIEVE_CHECK_GOTO(tNameGetDbName(&name, varDataVal(db)), pStb, &lino, _ERROR);
852,800!
3300
    varDataSetLen(db, strlen(varDataVal(db)));
853,258✔
3301
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
853,258✔
3302
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)db, false), pStb, &lino, _ERROR);
852,270!
3303

3304
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
848,654✔
3305
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->createdTime, false), pStb, &lino,
847,048!
3306
                        _ERROR);
3307

3308
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
845,897✔
3309
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->numOfColumns, false), pStb, &lino,
844,353!
3310
                        _ERROR);
3311

3312
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
844,173✔
3313
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->numOfTags, false), pStb, &lino, _ERROR);
842,913!
3314

3315
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
843,384✔
3316
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->updateTime, false), pStb, &lino,
842,047!
3317
                        _ERROR);  // number of tables
3318

3319
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
842,519✔
3320
    if (pStb->commentLen > 0) {
841,443!
3321
      char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0};
×
3322
      STR_TO_VARSTR(comment, pStb->comment);
×
3323
      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, comment, false), pStb, &lino, _ERROR);
×
3324
    } else if (pStb->commentLen == 0) {
842,148✔
3325
      char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};
23,802✔
3326
      STR_TO_VARSTR(comment, "");
23,802✔
3327
      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, comment, false), pStb, &lino, _ERROR);
23,802!
3328
    } else {
3329
      colDataSetNULL(pColInfo, numOfRows);
818,346!
3330
    }
3331

3332
    char watermark[64 + VARSTR_HEADER_SIZE] = {0};
842,225✔
3333
    (void)tsnprintf(varDataVal(watermark), sizeof(watermark) - VARSTR_HEADER_SIZE, "%" PRId64 "a,%" PRId64 "a",
842,225✔
3334
              pStb->watermark[0], pStb->watermark[1]);
842,225✔
3335
    varDataSetLen(watermark, strlen(varDataVal(watermark)));
853,959✔
3336

3337
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
853,959✔
3338
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)watermark, false), pStb, &lino, _ERROR);
850,777!
3339

3340
    char maxDelay[64 + VARSTR_HEADER_SIZE] = {0};
846,990✔
3341
    (void)tsnprintf(varDataVal(maxDelay), sizeof(maxDelay) - VARSTR_HEADER_SIZE, "%" PRId64 "a,%" PRId64 "a",
846,990✔
3342
              pStb->maxdelay[0], pStb->maxdelay[1]);
846,990✔
3343
    varDataSetLen(maxDelay, strlen(varDataVal(maxDelay)));
853,723✔
3344

3345
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
853,723✔
3346
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)maxDelay, false), pStb, &lino, _ERROR);
850,736!
3347

3348
    char    rollup[160 + VARSTR_HEADER_SIZE] = {0};
846,849✔
3349
    int32_t rollupNum = (int32_t)taosArrayGetSize(pStb->pFuncs);
846,849✔
3350
    char   *sep = ", ";
848,958✔
3351
    int32_t sepLen = strlen(sep);
848,958✔
3352
    int32_t rollupLen = sizeof(rollup) - VARSTR_HEADER_SIZE - 2;
848,958✔
3353
    for (int32_t i = 0; i < rollupNum; ++i) {
848,961✔
3354
      char *funcName = taosArrayGet(pStb->pFuncs, i);
3✔
3355
      if (i) {
3!
3356
        (void)strncat(varDataVal(rollup), sep, rollupLen);
×
3357
        rollupLen -= sepLen;
×
3358
      }
3359
      (void)strncat(varDataVal(rollup), funcName, rollupLen);
3✔
3360
      rollupLen -= strlen(funcName);
3✔
3361
    }
3362
    varDataSetLen(rollup, strlen(varDataVal(rollup)));
848,958✔
3363

3364
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
848,958✔
3365
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)rollup, false), pStb, &lino, _ERROR);
849,420!
3366

3367
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
845,487✔
3368
    if (pColInfo) {
843,314!
3369
      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)(&pStb->uid), false), pStb, &lino, _ERROR);
843,372!
3370
    }
3371

3372
    numOfRows++;
842,178✔
3373
    sdbRelease(pSdb, pStb);
842,178✔
3374
  }
3375

3376
  if (pDb != NULL) {
16,377✔
3377
    mndReleaseDb(pMnode, pDb);
411✔
3378
  }
3379

3380
  goto _OVER;
16,360✔
3381

3382
_ERROR:
×
3383
  mError("show:0x%" PRIx64 ", failed to retrieve data at %s:%d since %s", pShow->id, __FUNCTION__, lino,
×
3384
         tstrerror(code));
3385

3386
_OVER:
×
3387
  pShow->numOfRows += numOfRows;
16,360✔
3388
  return numOfRows;
16,360✔
3389
}
3390

3391
static int32_t buildDbColsInfoBlock(const SSDataBlock *p, const SSysTableMeta *pSysDbTableMeta, size_t size,
26,607✔
3392
                                    const char *dbName, const char *tbName) {
3393
  char    tName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
26,607✔
3394
  char    dName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
26,607✔
3395
  char    typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
26,607✔
3396
  int32_t numOfRows = p->info.rows;
26,607✔
3397
  int32_t lino = 0;
26,607✔
3398
  int32_t code = 0;
26,607✔
3399

3400
  STR_TO_VARSTR(dName, dbName);
26,607✔
3401
  STR_TO_VARSTR(typeName, "SYSTEM_TABLE");
26,607✔
3402

3403
  for (int32_t i = 0; i < size; ++i) {
556,316!
3404
    const SSysTableMeta *pm = &pSysDbTableMeta[i];
558,666✔
3405
    //    if (pm->sysInfo) {
3406
    //      continue;
3407
    //    }
3408
    if (tbName[0] && strncmp(tbName, pm->name, TSDB_TABLE_NAME_LEN) != 0) {
558,666!
3409
      continue;
×
3410
    }
3411

3412
    STR_TO_VARSTR(tName, pm->name);
558,666✔
3413

3414
    for (int32_t j = 0; j < pm->colNum; j++) {
5,345,732✔
3415
      // table name
3416
      SColumnInfoData *pColInfoData = taosArrayGet(p->pDataBlock, 0);
4,816,023✔
3417
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, tName, false), &lino, _OVER);
4,814,674!
3418

3419
      // database name
3420
      pColInfoData = taosArrayGet(p->pDataBlock, 1);
4,788,937✔
3421
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, dName, false), &lino, _OVER);
4,787,515!
3422

3423
      pColInfoData = taosArrayGet(p->pDataBlock, 2);
4,786,531✔
3424
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, typeName, false), &lino, _OVER);
4,785,266!
3425

3426
      // col name
3427
      char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
4,788,881✔
3428
      STR_TO_VARSTR(colName, pm->schema[j].name);
4,788,881✔
3429
      pColInfoData = taosArrayGet(p->pDataBlock, 3);
4,788,881✔
3430
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, colName, false), &lino, _OVER);
4,816,757!
3431

3432
      // col type
3433
      int8_t colType = pm->schema[j].type;
4,794,059✔
3434
      pColInfoData = taosArrayGet(p->pDataBlock, 4);
4,794,059✔
3435
      char colTypeStr[VARSTR_HEADER_SIZE + 32];
3436
      int  colTypeLen =
4,828,565✔
3437
          tsnprintf(varDataVal(colTypeStr), sizeof(colTypeStr) - VARSTR_HEADER_SIZE, "%s", tDataTypes[colType].name);
4,792,790✔
3438
      if (colType == TSDB_DATA_TYPE_VARCHAR) {
4,828,565✔
3439
        colTypeLen +=
2,567,318✔
3440
            tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE, "(%d)",
2,567,222✔
3441
                      (int32_t)(pm->schema[j].bytes - VARSTR_HEADER_SIZE));
2,567,222✔
3442
      } else if (colType == TSDB_DATA_TYPE_NCHAR) {
2,261,343!
3443
        colTypeLen +=
×
3444
            tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE, "(%d)",
×
3445
                      (int32_t)((pm->schema[j].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
×
3446
      }
3447
      varDataSetLen(colTypeStr, colTypeLen);
4,828,661✔
3448
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, (char *)colTypeStr, false), &lino, _OVER);
4,828,661!
3449

3450
      pColInfoData = taosArrayGet(p->pDataBlock, 5);
4,796,824✔
3451
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, (const char *)&pm->schema[j].bytes, false), &lino, _OVER);
4,795,061!
3452
      for (int32_t k = 6; k <= 8; ++k) {
19,113,534✔
3453
        pColInfoData = taosArrayGet(p->pDataBlock, k);
14,326,468✔
3454
        colDataSetNULL(pColInfoData, numOfRows);
14,324,331!
3455
      }
3456

3457
      numOfRows += 1;
4,787,066✔
3458
    }
3459
  }
3460
_OVER:
×
3461
  mError("failed at %s:%d since %s", __FUNCTION__, lino, tstrerror(code));
×
3462
  return numOfRows;
26,608✔
3463
}
3464
#define BUILD_COL_FOR_INFO_DB 1
3465
#define BUILD_COL_FOR_PERF_DB 1 << 1
3466
#define BUILD_COL_FOR_USER_DB 1 << 2
3467
#define BUILD_COL_FOR_ALL_DB  (BUILD_COL_FOR_INFO_DB | BUILD_COL_FOR_PERF_DB | BUILD_COL_FOR_USER_DB)
3468

3469
static int32_t buildSysDbColsInfo(SSDataBlock *p, int8_t buildWhichDBs, char *tb) {
13,303✔
3470
  size_t               size = 0;
13,303✔
3471
  const SSysTableMeta *pSysDbTableMeta = NULL;
13,303✔
3472

3473
  if (buildWhichDBs & BUILD_COL_FOR_INFO_DB) {
13,303!
3474
    getInfosDbMeta(&pSysDbTableMeta, &size);
13,303✔
3475
    p->info.rows = buildDbColsInfoBlock(p, pSysDbTableMeta, size, TSDB_INFORMATION_SCHEMA_DB, tb);
13,303✔
3476
  }
3477

3478
  if (buildWhichDBs & BUILD_COL_FOR_PERF_DB) {
13,304!
3479
    getPerfDbMeta(&pSysDbTableMeta, &size);
13,304✔
3480
    p->info.rows = buildDbColsInfoBlock(p, pSysDbTableMeta, size, TSDB_PERFORMANCE_SCHEMA_DB, tb);
13,304✔
3481
  }
3482

3483
  return p->info.rows;
13,304✔
3484
}
3485

3486
static int8_t determineBuildColForWhichDBs(const char *db) {
13,303✔
3487
  int8_t buildWhichDBs;
3488
  if (!db[0])
13,303!
3489
    buildWhichDBs = BUILD_COL_FOR_ALL_DB;
13,303✔
3490
  else {
3491
    char *p = strchr(db, '.');
×
3492
    if (p && strcmp(p + 1, TSDB_INFORMATION_SCHEMA_DB) == 0) {
×
3493
      buildWhichDBs = BUILD_COL_FOR_INFO_DB;
×
3494
    } else if (p && strcmp(p + 1, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {
×
3495
      buildWhichDBs = BUILD_COL_FOR_PERF_DB;
×
3496
    } else {
3497
      buildWhichDBs = BUILD_COL_FOR_USER_DB;
×
3498
    }
3499
  }
3500
  return buildWhichDBs;
13,303✔
3501
}
3502

3503
static int32_t mndRetrieveStbCol(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
13,303✔
3504
  uint8_t  buildWhichDBs;
3505
  SMnode  *pMnode = pReq->info.node;
13,303✔
3506
  SSdb    *pSdb = pMnode->pSdb;
13,303✔
3507
  SStbObj *pStb = NULL;
13,303✔
3508
  int32_t  numOfRows = 0;
13,303✔
3509
  int32_t  lino = 0;
13,303✔
3510
  int32_t  code = 0;
13,303✔
3511

3512
  buildWhichDBs = determineBuildColForWhichDBs(pShow->db);
13,303✔
3513

3514
  if (!pShow->sysDbRsp) {
13,304!
3515
    numOfRows = buildSysDbColsInfo(pBlock, buildWhichDBs, pShow->filterTb);
13,304✔
3516
    mDebug("mndRetrieveStbCol get system table cols, rows:%d, db:%s", numOfRows, pShow->db);
13,304✔
3517
    pShow->sysDbRsp = true;
13,304✔
3518
  }
3519

3520
  if (buildWhichDBs & BUILD_COL_FOR_USER_DB) {
13,304!
3521
    SDbObj *pDb = NULL;
13,304✔
3522
    if (strlen(pShow->db) > 0) {
13,304!
3523
      pDb = mndAcquireDb(pMnode, pShow->db);
×
3524
      if (pDb == NULL && TSDB_CODE_MND_DB_NOT_EXIST != terrno && pBlock->info.rows == 0) return terrno;
×
3525
    }
3526

3527
    char typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
13,304✔
3528
    STR_TO_VARSTR(typeName, "SUPER_TABLE");
13,304✔
3529
    bool fetch = pShow->restore ? false : true;
13,304✔
3530
    pShow->restore = false;
13,304✔
3531
    while (numOfRows < rows) {
713,642✔
3532
      if (fetch) {
713,629!
3533
        pShow->pIter = sdbFetch(pSdb, SDB_STB, pShow->pIter, (void **)&pStb);
713,629✔
3534
        if (pShow->pIter == NULL) break;
713,627✔
3535
      } else {
3536
        fetch = true;
×
3537
        void *pKey = taosHashGetKey(pShow->pIter, NULL);
×
3538
        pStb = sdbAcquire(pSdb, SDB_STB, pKey);
×
3539
        if (!pStb) continue;
×
3540
      }
3541

3542
      if (pDb != NULL && pStb->dbUid != pDb->uid) {
700,323!
3543
        sdbRelease(pSdb, pStb);
×
3544
        continue;
×
3545
      }
3546

3547
      SName name = {0};
700,323✔
3548
      char  stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
700,323✔
3549
      mndExtractTbNameFromStbFullName(pStb->name, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN);
700,323✔
3550
      if (pShow->filterTb[0] && strncmp(pShow->filterTb, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN) != 0) {
700,290!
3551
        sdbRelease(pSdb, pStb);
×
3552
        continue;
×
3553
      }
3554

3555
      if ((numOfRows + pStb->numOfColumns) > rows) {
700,290!
3556
        pShow->restore = true;
×
3557
        if (numOfRows == 0) {
×
3558
          mError("mndRetrieveStbCol failed to get stable cols since buf:%d less than result:%d, stable name:%s, db:%s",
×
3559
                 rows, pStb->numOfColumns, pStb->name, pStb->db);
3560
        }
3561
        sdbRelease(pSdb, pStb);
×
3562
        break;
×
3563
      }
3564

3565
      varDataSetLen(stbName, strlen(&stbName[VARSTR_HEADER_SIZE]));
700,290✔
3566

3567
      mDebug("mndRetrieveStbCol get stable cols, stable name:%s, db:%s", pStb->name, pStb->db);
700,290✔
3568

3569
      char db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
700,290✔
3570
      RETRIEVE_CHECK_GOTO(tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB), pStb, &lino, _OVER);
700,290!
3571
      RETRIEVE_CHECK_GOTO(tNameGetDbName(&name, varDataVal(db)), pStb, &lino, _OVER);
700,218!
3572
      varDataSetLen(db, strlen(varDataVal(db)));
700,258✔
3573

3574
      for (int i = 0; i < pStb->numOfColumns; i++) {
16,226,060✔
3575
        int32_t          cols = 0;
16,042,497✔
3576
        SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
16,042,497✔
3577
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbName, false), pStb, &lino, _OVER);
16,064,441!
3578

3579
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
15,838,157✔
3580
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)db, false), pStb, &lino, _OVER);
15,709,502!
3581

3582
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
15,711,895✔
3583
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, typeName, false), pStb, &lino, _OVER);
15,603,089!
3584

3585
        // col name
3586
        char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
15,691,835✔
3587
        STR_TO_VARSTR(colName, pStb->pColumns[i].name);
15,691,835✔
3588
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
15,691,835✔
3589
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, colName, false), pStb, &lino, _OVER);
16,021,471!
3590

3591
        // col type
3592
        int8_t colType = pStb->pColumns[i].type;
15,847,425✔
3593
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
15,847,425✔
3594
        char colTypeStr[VARSTR_HEADER_SIZE + 32];
3595
        int  colTypeLen =
16,428,200✔
3596
            tsnprintf(varDataVal(colTypeStr), sizeof(colTypeStr) - VARSTR_HEADER_SIZE, "%s", tDataTypes[colType].name);
15,747,259✔
3597
        if (colType == TSDB_DATA_TYPE_VARCHAR) {
16,428,200✔
3598
          colTypeLen +=
1,302,092✔
3599
              tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE,
1,302,092✔
3600
                        "(%d)", (int32_t)(pStb->pColumns[i].bytes - VARSTR_HEADER_SIZE));
1,302,092✔
3601
        } else if (colType == TSDB_DATA_TYPE_NCHAR) {
15,126,108✔
3602
          colTypeLen +=
1,315,026✔
3603
              tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE,
1,315,107✔
3604
                        "(%d)", (int32_t)((pStb->pColumns[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
1,315,107✔
3605
        }
3606
        varDataSetLen(colTypeStr, colTypeLen);
16,428,119✔
3607
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (char *)colTypeStr, false), pStb, &lino, _OVER);
16,428,119!
3608

3609
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
15,941,778✔
3610
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->pColumns[i].bytes, false), pStb,
15,815,937!
3611
                            &lino, _OVER);
3612
        while (cols < pShow->numOfColumns) {
61,572,153✔
3613
          pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
46,046,351✔
3614
          colDataSetNULL(pColInfo, numOfRows);
45,847,825!
3615
        }
3616
        numOfRows++;
15,525,802✔
3617
      }
3618

3619
      sdbRelease(pSdb, pStb);
183,563✔
3620
    }
3621

3622
    if (pDb != NULL) {
13,317!
3623
      mndReleaseDb(pMnode, pDb);
×
3624
    }
3625
  }
3626

3627
  mDebug("mndRetrieveStbCol success, rows:%d, pShow->numOfRows:%d", numOfRows, pShow->numOfRows);
13,304✔
3628
  goto _OVER;
13,304✔
3629

3630
_ERROR:
3631
  mError("failed to mndRetrieveStbCol, rows:%d, pShow->numOfRows:%d, at %s:%d since %s", numOfRows, pShow->numOfRows,
3632
         __FUNCTION__, lino, tstrerror(code));
3633

3634
_OVER:
13,304✔
3635
  pShow->numOfRows += numOfRows;
13,304✔
3636
  return numOfRows;
13,304✔
3637
}
3638

3639
static void mndCancelGetNextStb(SMnode *pMnode, void *pIter) {
×
3640
  SSdb *pSdb = pMnode->pSdb;
×
3641
  sdbCancelFetchByType(pSdb, pIter, SDB_STB);
×
3642
}
×
3643

3644
const char *mndGetStbStr(const char *src) {
33,143✔
3645
  char *posDb = strstr(src, TS_PATH_DELIMITER);
33,143✔
3646
  if (posDb != NULL) ++posDb;
33,143!
3647
  if (posDb == NULL) return src;
33,143!
3648

3649
  char *posStb = strstr(posDb, TS_PATH_DELIMITER);
33,143✔
3650
  if (posStb != NULL) ++posStb;
33,143!
3651
  if (posStb == NULL) return posDb;
33,143!
3652
  return posStb;
33,143✔
3653
}
3654

3655
static int32_t mndCheckIndexReq(SCreateTagIndexReq *pReq) {
×
3656
  // impl
3657
  return TSDB_CODE_SUCCESS;
×
3658
}
3659

3660
/*int32_t mndAddIndexImpl(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp, void *sql,
3661
                        int32_t len) {
3662
  // impl later
3663
  int32_t code = 0;
3664
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "create-stb-index");
3665
  if (pTrans == NULL) goto _OVER;
3666

3667
  mInfo("trans:%d, used to add index to stb:%s", pTrans->id, pStb->name);
3668
  mndTransSetDbName(pTrans, pDb->name, pStb->name);
3669
  if (mndTransCheckConflict(pMnode, pTrans) != 0) goto _OVER;
3670

3671
  if (mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb) != 0) goto _OVER;
3672
  if (mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb) != 0) goto _OVER;
3673
  if (mndSetAlterStbRedoActions2(pMnode, pTrans, pDb, pStb, sql, len) != 0) goto _OVER;
3674
  if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
3675

3676
  return code;
3677

3678
_OVER:
3679
  mndTransDrop(pTrans);
3680
  return code;
3681
}
3682
static int32_t mndAddIndex(SMnode *pMnode, SRpcMsg *pReq, SCreateTagIndexReq *tagIdxReq, SDbObj *pDb, SStbObj *pOld) {
3683
  bool    needRsp = true;
3684
  int32_t code = -1;
3685
  SField *pField0 = NULL;
3686

3687
  SStbObj  stbObj = {0};
3688
  SStbObj *pNew = &stbObj;
3689

3690
  taosRLockLatch(&pOld->lock);
3691
  memcpy(&stbObj, pOld, sizeof(SStbObj));
3692
  taosRUnLockLatch(&pOld->lock);
3693

3694
  stbObj.pColumns = NULL;
3695
  stbObj.pTags = NULL;
3696
  stbObj.updateTime = taosGetTimestampMs();
3697
  stbObj.lock = 0;
3698

3699
  int32_t tag = mndFindSuperTableTagIndex(pOld, tagIdxReq->colName);
3700
  if (tag < 0) {
3701
    terrno = TSDB_CODE_MND_TAG_NOT_EXIST;
3702
    return -1;
3703
  }
3704
  if (mndAllocStbSchemas(pOld, pNew) != 0) {
3705
    return -1;
3706
  }
3707

3708
  SSchema *pTag = pNew->pTags + tag;
3709
  if (IS_IDX_ON(pTag)) {
3710
    terrno = TSDB_CODE_MND_TAG_INDEX_ALREADY_EXIST;
3711
    return -1;
3712
  } else {
3713
    pTag->flags |= COL_IDX_ON;
3714
  }
3715
  pNew->tagVer++;
3716

3717
  code = mndAddIndexImpl(pMnode, pReq, pDb, pNew, needRsp, pReq->pCont, pReq->contLen);
3718

3719
  return code;
3720
}
3721
static int32_t mndProcessCreateIndexReq(SRpcMsg *pReq) {
3722
  SMnode            *pMnode = pReq->info.node;
3723
  int32_t            code = -1;
3724
  SDbObj            *pDb = NULL;
3725
  SStbObj           *pStb = NULL;
3726
  SCreateTagIndexReq tagIdxReq = {0};
3727

3728
  if (tDeserializeSCreateTagIdxReq(pReq->pCont, pReq->contLen, &tagIdxReq) != 0) {
3729
    terrno = TSDB_CODE_INVALID_MSG;
3730
    goto _OVER;
3731
  }
3732

3733
  mInfo("stb:%s, start to alter", tagIdxReq.stbName);
3734

3735
  if (mndCheckIndexReq(&tagIdxReq) != TSDB_CODE_SUCCESS) {
3736
    goto _OVER;
3737
  }
3738

3739
  pDb = mndAcquireDbByStb(pMnode, tagIdxReq.dbFName);
3740
  if (pDb == NULL) {
3741
    terrno = TSDB_CODE_MND_DB_NOT_EXIST;
3742
    goto _OVER;
3743
  }
3744

3745
  pStb = mndAcquireStb(pMnode, tagIdxReq.stbName);
3746
  if (pStb == NULL) {
3747
    terrno = TSDB_CODE_MND_STB_NOT_EXIST;
3748
    goto _OVER;
3749
  }
3750
  if (mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pDb) != 0) {
3751
    goto _OVER;
3752
  }
3753

3754
  code = mndAddIndex(pMnode, pReq, &tagIdxReq, pDb, pStb);
3755
  if (terrno == TSDB_CODE_MND_TAG_INDEX_ALREADY_EXIST || terrno == TSDB_CODE_MND_TAG_NOT_EXIST) {
3756
    return terrno;
3757
  } else {
3758
    if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
3759
  }
3760
_OVER:
3761
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
3762
    mError("stb:%s, failed to create index since %s", tagIdxReq.stbName, terrstr());
3763
  }
3764
  mndReleaseStb(pMnode, pStb);
3765
  mndReleaseDb(pMnode, pDb);
3766
  return code;
3767
}
3768
static int32_t mndProcessDropIndexReq(SRpcMsg *pReq) {
3769
  SMnode          *pMnode = pReq->info.node;
3770
  int32_t          code = -1;
3771
  SDbObj          *pDb = NULL;
3772
  SStbObj         *pStb = NULL;
3773
  SDropTagIndexReq dropReq = {0};
3774
  if (tDeserializeSDropTagIdxReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
3775
    terrno = TSDB_CODE_INVALID_MSG;
3776
    goto _OVER;
3777
  }
3778
  //
3779
  return TSDB_CODE_SUCCESS;
3780
_OVER:
3781
  return code;
3782
}*/
3783

3784
static int32_t mndProcessDropStbReqFromMNode(SRpcMsg *pReq) {
214✔
3785
  int32_t code = mndProcessDropStbReq(pReq);
214✔
3786
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
214!
3787
    pReq->info.rsp = rpcMallocCont(1);
×
3788
    pReq->info.rspLen = 1;
×
3789
    pReq->info.noResp = false;
×
3790
    pReq->code = code;
×
3791
  }
3792
  return code;
214✔
3793
}
3794

3795
typedef struct SVDropTbVgReqs {
3796
  SArray     *pBatchReqs;
3797
  SVgroupInfo info;
3798
} SVDropTbVgReqs;
3799

3800
typedef struct SMDropTbDbInfo {
3801
  SArray *dbVgInfos;
3802
  int32_t hashPrefix;
3803
  int32_t hashSuffix;
3804
  int32_t hashMethod;
3805
} SMDropTbDbInfo;
3806

3807
typedef struct SMDropTbTsmaInfo {
3808
  char           tsmaResTbDbFName[TSDB_DB_FNAME_LEN];
3809
  char           tsmaResTbNamePrefix[TSDB_TABLE_FNAME_LEN];
3810
  int32_t        suid;
3811
  SMDropTbDbInfo dbInfo;  // reference to DbInfo in pDbMap
3812
} SMDropTbTsmaInfo;
3813

3814
typedef struct SMDropTbTsmaInfos {
3815
  SArray *pTsmaInfos;  // SMDropTbTsmaInfo
3816
} SMDropTbTsmaInfos;
3817

3818
typedef struct SMndDropTbsWithTsmaCtx {
3819
  SHashObj *pVgMap;  // <vgId, SVDropTbVgReqs>
3820
} SMndDropTbsWithTsmaCtx;
3821

3822
static int32_t mndDropTbForSingleVg(SMnode *pMnode, SMndDropTbsWithTsmaCtx *pCtx, SArray *pTbs, int32_t vgId);
3823

3824
static void destroySVDropTbBatchReqs(void *p);
3825
static void mndDestroyDropTbsWithTsmaCtx(SMndDropTbsWithTsmaCtx *p) {
30✔
3826
  if (!p) return;
30!
3827

3828
  if (p->pVgMap) {
30!
3829
    void *pIter = taosHashIterate(p->pVgMap, NULL);
30✔
3830
    while (pIter) {
65✔
3831
      SVDropTbVgReqs *pReqs = pIter;
35✔
3832
      taosArrayDestroyEx(pReqs->pBatchReqs, destroySVDropTbBatchReqs);
35✔
3833
      pIter = taosHashIterate(p->pVgMap, pIter);
35✔
3834
    }
3835
    taosHashCleanup(p->pVgMap);
30✔
3836
  }
3837
  taosMemoryFree(p);
30!
3838
}
3839

3840
static int32_t mndInitDropTbsWithTsmaCtx(SMndDropTbsWithTsmaCtx **ppCtx) {
30✔
3841
  int32_t                 code = 0;
30✔
3842
  SMndDropTbsWithTsmaCtx *pCtx = taosMemoryCalloc(1, sizeof(SMndDropTbsWithTsmaCtx));
30!
3843
  if (!pCtx) return terrno;
30!
3844

3845
  pCtx->pVgMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
30✔
3846
  if (!pCtx->pVgMap) {
30!
3847
    code = terrno;
×
3848
    goto _end;
×
3849
  }
3850

3851
  *ppCtx = pCtx;
30✔
3852
_end:
30✔
3853
  if (code) mndDestroyDropTbsWithTsmaCtx(pCtx);
30!
3854
  return code;
30✔
3855
}
3856

3857
static void *mndBuildVDropTbsReq(SMnode *pMnode, const SVgroupInfo *pVgInfo, const SVDropTbBatchReq *pReq,
35✔
3858
                                 int32_t *len) {
3859
  int32_t   contLen = 0;
35✔
3860
  int32_t   ret = 0;
35✔
3861
  SMsgHead *pHead = NULL;
35✔
3862
  SEncoder  encoder = {0};
35✔
3863

3864
  tEncodeSize(tEncodeSVDropTbBatchReq, pReq, contLen, ret);
35!
3865
  if (ret < 0) return NULL;
35!
3866

3867
  contLen += sizeof(SMsgHead);
35✔
3868
  pHead = taosMemoryMalloc(contLen);
35!
3869
  if (pHead == NULL) {
35!
3870
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
3871
    return NULL;
×
3872
  }
3873

3874
  pHead->contLen = htonl(contLen);
35✔
3875
  pHead->vgId = htonl(pVgInfo->vgId);
35✔
3876

3877
  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
35✔
3878

3879
  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
35✔
3880
  int32_t code = tEncodeSVDropTbBatchReq(&encoder, pReq);
35✔
3881
  tEncoderClear(&encoder);
35✔
3882
  if (code != 0) return NULL;
35!
3883

3884
  *len = contLen;
35✔
3885
  return pHead;
35✔
3886
}
3887

3888
static int32_t mndSetDropTbsRedoActions(SMnode *pMnode, STrans *pTrans, const SVDropTbVgReqs *pVgReqs, void *pCont,
35✔
3889
                                        int32_t contLen, tmsg_t msgType) {
3890
  STransAction action = {0};
35✔
3891
  action.epSet = pVgReqs->info.epSet;
35✔
3892
  action.pCont = pCont;
35✔
3893
  action.contLen = contLen;
35✔
3894
  action.msgType = msgType;
35✔
3895
  action.acceptableCode = TSDB_CODE_TDB_TABLE_NOT_EXIST;
35✔
3896
  return mndTransAppendRedoAction(pTrans, &action);
35✔
3897
}
3898

3899
static int32_t mndBuildDropTbRedoActions(SMnode *pMnode, STrans *pTrans, SHashObj *pVgMap, tmsg_t msgType) {
30✔
3900
  int32_t code = 0;
30✔
3901
  void   *pIter = taosHashIterate(pVgMap, NULL);
30✔
3902
  while (pIter) {
65✔
3903
    const SVDropTbVgReqs *pVgReqs = pIter;
35✔
3904
    int32_t               len = 0;
35✔
3905
    for (int32_t i = 0; i < taosArrayGetSize(pVgReqs->pBatchReqs) && code == TSDB_CODE_SUCCESS; ++i) {
70!
3906
      SVDropTbBatchReq *pBatchReq = taosArrayGet(pVgReqs->pBatchReqs, i);
35✔
3907
      void             *p = mndBuildVDropTbsReq(pMnode, &pVgReqs->info, pBatchReq, &len);
35✔
3908
      if (!p) {
35!
3909
        code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
3910
        if (terrno != 0) code = terrno;
×
3911
        break;
×
3912
      }
3913
      if ((code = mndSetDropTbsRedoActions(pMnode, pTrans, pVgReqs, p, len, msgType)) != 0) {
35!
3914
        break;
×
3915
      }
3916
    }
3917
    if (TSDB_CODE_SUCCESS != code) {
35!
3918
      taosHashCancelIterate(pVgMap, pIter);
×
3919
      break;
×
3920
    }
3921
    pIter = taosHashIterate(pVgMap, pIter);
35✔
3922
  }
3923
  return code;
30✔
3924
}
3925

3926
static int32_t mndCreateDropTbsTxnPrepare(SRpcMsg *pRsp, SMndDropTbsWithTsmaCtx *pCtx) {
30✔
3927
  int32_t code = 0;
30✔
3928
  SMnode *pMnode = pRsp->info.node;
30✔
3929
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, pRsp, "drop-tbs");
30✔
3930
  mndTransSetChangeless(pTrans);
30✔
3931
  mndTransSetSerial(pTrans);
30✔
3932
  if (pTrans == NULL) {
30!
3933
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
3934
    if (terrno != 0) code = terrno;
×
3935
    goto _OVER;
×
3936
  }
3937

3938
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
30!
3939

3940
  if ((code = mndBuildDropTbRedoActions(pMnode, pTrans, pCtx->pVgMap, TDMT_VND_DROP_TABLE)) != 0) goto _OVER;
30!
3941
  if ((code = mndTransPrepare(pMnode, pTrans)) != 0) goto _OVER;
30!
3942

3943
_OVER:
30✔
3944
  mndTransDrop(pTrans);
30✔
3945
  TAOS_RETURN(code);
30✔
3946
}
3947

3948
static int32_t mndProcessDropTbWithTsma(SRpcMsg *pReq) {
25✔
3949
  int32_t      code = -1;
25✔
3950
  SMnode      *pMnode = pReq->info.node;
25✔
3951
  SDbObj      *pDb = NULL;
25✔
3952
  SStbObj     *pStb = NULL;
25✔
3953
  SMDropTbsReq dropReq = {0};
25✔
3954
  bool         locked = false;
25✔
3955
  if (tDeserializeSMDropTbsReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
25!
3956
    code = TSDB_CODE_INVALID_MSG;
×
3957
    goto _OVER;
×
3958
  }
3959

3960
  SMndDropTbsWithTsmaCtx *pCtx = NULL;
25✔
3961
  code = mndInitDropTbsWithTsmaCtx(&pCtx);
25✔
3962
  if (code) goto _OVER;
25!
3963
  for (int32_t i = 0; i < dropReq.pVgReqs->size; ++i) {
55✔
3964
    SMDropTbReqsOnSingleVg *pReq = taosArrayGet(dropReq.pVgReqs, i);
30✔
3965
    code = mndDropTbForSingleVg(pMnode, pCtx, pReq->pTbs, pReq->vgInfo.vgId);
30✔
3966
    if (code) goto _OVER;
30!
3967
  }
3968
  code = mndCreateDropTbsTxnPrepare(pReq, pCtx);
25✔
3969
  if (code == 0) {
25!
3970
    code = TSDB_CODE_ACTION_IN_PROGRESS;
25✔
3971
  }
3972
_OVER:
×
3973
  tFreeSMDropTbsReq(&dropReq);
25✔
3974
  if (pCtx) mndDestroyDropTbsWithTsmaCtx(pCtx);
25!
3975
  TAOS_RETURN(code);
25✔
3976
}
3977

3978
static int32_t createDropTbBatchReq(const SVDropTbReq *pReq, SVDropTbBatchReq *pBatchReq) {
35✔
3979
  pBatchReq->nReqs = 1;
35✔
3980
  pBatchReq->pArray = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SVDropTbReq));
35✔
3981
  if (!pBatchReq->pArray) return terrno;
35!
3982
  if (taosArrayPush(pBatchReq->pArray, pReq) == NULL) {
70!
3983
    taosArrayDestroy(pBatchReq->pArray);
×
3984
    pBatchReq->pArray = NULL;
×
3985
    return terrno;
×
3986
  }
3987
  return TSDB_CODE_SUCCESS;
35✔
3988
}
3989

3990
static void destroySVDropTbBatchReqs(void *p) {
35✔
3991
  SVDropTbBatchReq *pReq = p;
35✔
3992
  taosArrayDestroy(pReq->pArray);
35✔
3993
  pReq->pArray = NULL;
35✔
3994
}
35✔
3995

3996
static int32_t mndDropTbAdd(SMnode *pMnode, SHashObj *pVgHashMap, const SVgroupInfo *pVgInfo, char *name, tb_uid_t suid,
35✔
3997
                            bool ignoreNotExists) {
3998
  SVDropTbReq req = {.name = name, .suid = suid, .igNotExists = ignoreNotExists, .uid = 0};
35✔
3999

4000
  SVDropTbVgReqs *pVgReqs = taosHashGet(pVgHashMap, &pVgInfo->vgId, sizeof(pVgInfo->vgId));
35✔
4001
  SVDropTbVgReqs  vgReqs = {0};
35✔
4002
  if (pVgReqs == NULL) {
35!
4003
    vgReqs.info = *pVgInfo;
35✔
4004
    vgReqs.pBatchReqs = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SVDropTbBatchReq));
35✔
4005
    if (!vgReqs.pBatchReqs) return terrno;
35!
4006
    SVDropTbBatchReq batchReq = {0};
35✔
4007
    int32_t          code = createDropTbBatchReq(&req, &batchReq);
35✔
4008
    if (TSDB_CODE_SUCCESS != code) return code;
35!
4009
    if (taosArrayPush(vgReqs.pBatchReqs, &batchReq) == NULL) {
70!
4010
      taosArrayDestroy(batchReq.pArray);
×
4011
      return terrno;
×
4012
    }
4013
    if (taosHashPut(pVgHashMap, &pVgInfo->vgId, sizeof(pVgInfo->vgId), &vgReqs, sizeof(vgReqs)) != 0) {
35!
4014
      taosArrayDestroyEx(vgReqs.pBatchReqs, destroySVDropTbBatchReqs);
×
4015
      return terrno;
×
4016
    }
4017
  } else {
4018
    SVDropTbBatchReq batchReq = {0};
×
4019
    int32_t          code = createDropTbBatchReq(&req, &batchReq);
×
4020
    if (TSDB_CODE_SUCCESS != code) return code;
×
4021
    if (taosArrayPush(pVgReqs->pBatchReqs, &batchReq) == NULL) {
×
4022
      taosArrayDestroy(batchReq.pArray);
×
4023
      return terrno;
×
4024
    }
4025
  }
4026
  return 0;
35✔
4027
}
4028

4029
static int32_t mndDropTbForSingleVg(SMnode *pMnode, SMndDropTbsWithTsmaCtx *pCtx, SArray *pTbs, int32_t vgId) {
35✔
4030
  int32_t code = 0;
35✔
4031

4032
  SVgObj *pVgObj = mndAcquireVgroup(pMnode, vgId);
35✔
4033
  if (!pVgObj) {
35!
4034
    code = 0;
×
4035
    goto _end;
×
4036
  }
4037
  SVgroupInfo vgInfo = {.hashBegin = pVgObj->hashBegin,
35✔
4038
                        .hashEnd = pVgObj->hashEnd,
35✔
4039
                        .numOfTable = pVgObj->numOfTables,
35✔
4040
                        .vgId = pVgObj->vgId};
35✔
4041
  vgInfo.epSet = mndGetVgroupEpset(pMnode, pVgObj);
35✔
4042
  mndReleaseVgroup(pMnode, pVgObj);
35✔
4043

4044
  for (int32_t i = 0; i < pTbs->size; ++i) {
70✔
4045
    SVDropTbReq *pTb = taosArrayGet(pTbs, i);
35✔
4046
    TAOS_CHECK_GOTO(mndDropTbAdd(pMnode, pCtx->pVgMap, &vgInfo, pTb->name, pTb->suid, pTb->igNotExists), NULL, _end);
35!
4047
  }
4048
_end:
35✔
4049
  return code;
35✔
4050
}
4051

4052
static int32_t mndProcessFetchTtlExpiredTbs(SRpcMsg *pRsp) {
98,485✔
4053
  int32_t                 code = -1;
98,485✔
4054
  SDecoder                decoder = {0};
98,485✔
4055
  SMnode                 *pMnode = pRsp->info.node;
98,485✔
4056
  SVFetchTtlExpiredTbsRsp rsp = {0};
98,485✔
4057
  SMndDropTbsWithTsmaCtx *pCtx = NULL;
98,485✔
4058
  if (pRsp->code != TSDB_CODE_SUCCESS) {
98,485✔
4059
    code = pRsp->code;
335✔
4060
    goto _end;
335✔
4061
  }
4062
  if (pRsp->contLen == 0) {
98,150✔
4063
    code = 0;
98,145✔
4064
    goto _end;
98,145✔
4065
  }
4066

4067
  tDecoderInit(&decoder, pRsp->pCont, pRsp->contLen);
5✔
4068
  code = tDecodeVFetchTtlExpiredTbsRsp(&decoder, &rsp);
5✔
4069
  if (code) goto _end;
5!
4070

4071
  code = mndInitDropTbsWithTsmaCtx(&pCtx);
5✔
4072
  if (code) goto _end;
5!
4073

4074
  code = mndDropTbForSingleVg(pMnode, pCtx, rsp.pExpiredTbs, rsp.vgId);
5✔
4075
  if (code) goto _end;
5!
4076
  code = mndCreateDropTbsTxnPrepare(pRsp, pCtx);
5✔
4077
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
5!
4078
_end:
×
4079
  if (pCtx) mndDestroyDropTbsWithTsmaCtx(pCtx);
98,485✔
4080
  tDecoderClear(&decoder);
98,485✔
4081
  tFreeFetchTtlExpiredTbsRsp(&rsp);
98,485✔
4082
  TAOS_RETURN(code);
98,485✔
4083
}
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