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

taosdata / TDengine / #3824

01 Apr 2025 12:26PM UTC coverage: 34.064% (-0.001%) from 34.065%
#3824

push

travis-ci

happyguoxy
test:alter gcda dir

148483 of 599532 branches covered (24.77%)

Branch coverage included in aggregate %.

222466 of 489445 relevant lines covered (45.45%)

762427.9 hits per line

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

44.77
/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   3
37
#define STB_RESERVE_SIZE 56
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) {
63✔
70
  SSdbTable table = {
63✔
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);
63✔
81
  mndSetMsgHandle(pMnode, TDMT_MND_ALTER_STB, mndProcessAlterStbReq);
63✔
82
  mndSetMsgHandle(pMnode, TDMT_MND_DROP_STB, mndProcessDropStbReq);
63✔
83
  mndSetMsgHandle(pMnode, TDMT_VND_CREATE_STB_RSP, mndTransProcessRsp);
63✔
84
  mndSetMsgHandle(pMnode, TDMT_VND_DROP_TTL_TABLE_RSP, mndProcessDropTtltbRsp);
63✔
85
  mndSetMsgHandle(pMnode, TDMT_VND_TRIM_RSP, mndProcessTrimDbRsp);
63✔
86
  mndSetMsgHandle(pMnode, TDMT_VND_ALTER_STB_RSP, mndTransProcessRsp);
63✔
87
  mndSetMsgHandle(pMnode, TDMT_VND_DROP_STB_RSP, mndTransProcessRsp);
63✔
88
  mndSetMsgHandle(pMnode, TDMT_MND_TABLE_META, mndProcessTableMetaReq);
63✔
89
  mndSetMsgHandle(pMnode, TDMT_MND_TTL_TIMER, mndProcessTtlTimer);
63✔
90
  mndSetMsgHandle(pMnode, TDMT_MND_TRIM_DB_TIMER, mndProcessTrimDbTimer);
63✔
91
  mndSetMsgHandle(pMnode, TDMT_VND_S3MIGRATE_RSP, mndProcessS3MigrateDbRsp);
63✔
92
  mndSetMsgHandle(pMnode, TDMT_MND_S3MIGRATE_DB_TIMER, mndProcessS3MigrateDbTimer);
63✔
93
  mndSetMsgHandle(pMnode, TDMT_MND_TABLE_CFG, mndProcessTableCfgReq);
63✔
94
  mndSetMsgHandle(pMnode, TDMT_MND_STB_DROP, mndProcessDropStbReqFromMNode);
63✔
95
  mndSetMsgHandle(pMnode, TDMT_MND_STB_DROP_RSP, mndTransProcessRsp);
63✔
96
  mndSetMsgHandle(pMnode, TDMT_MND_DROP_TB_WITH_TSMA, mndProcessDropTbWithTsma);
63✔
97
  mndSetMsgHandle(pMnode, TDMT_VND_FETCH_TTL_EXPIRED_TBS_RSP, mndProcessFetchTtlExpiredTbs);
63✔
98
  mndSetMsgHandle(pMnode, TDMT_VND_DROP_TABLE_RSP, mndTransProcessRsp);
63✔
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);
63✔
107
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_STB, mndCancelGetNextStb);
63✔
108

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

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

115
void mndCleanupStb(SMnode *pMnode) {}
60✔
116

117
SSdbRaw *mndStbActionEncode(SStbObj *pStb) {
313✔
118
  int32_t code = 0;
313✔
119
  int32_t lino = 0;
313✔
120
  bool    hasTypeMod = false;
313✔
121
  terrno = TSDB_CODE_OUT_OF_MEMORY;
313✔
122

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

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

152
  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
6,143✔
153
    SSchema *pSchema = &pStb->pColumns[i];
5,830✔
154
    SDB_SET_INT8(pRaw, dataPos, pSchema->type, _OVER)
5,830!
155
    SDB_SET_INT8(pRaw, dataPos, pSchema->flags, _OVER)
5,830!
156
    SDB_SET_INT16(pRaw, dataPos, pSchema->colId, _OVER)
5,830!
157
    SDB_SET_INT32(pRaw, dataPos, pSchema->bytes, _OVER)
5,830!
158
    SDB_SET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, _OVER)
5,830!
159
    hasTypeMod = hasTypeMod || HAS_TYPE_MOD(pSchema);
5,830✔
160
  }
161

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

171
  for (int32_t i = 0; i < pStb->numOfFuncs; ++i) {
313!
172
    char *func = taosArrayGet(pStb->pFuncs, i);
×
173
    SDB_SET_BINARY(pRaw, dataPos, func, TSDB_FUNC_NAME_LEN, _OVER)
×
174
  }
175

176
  if (pStb->commentLen > 0) {
313✔
177
    SDB_SET_BINARY(pRaw, dataPos, pStb->comment, pStb->commentLen + 1, _OVER)
41!
178
  }
179

180
  if (pStb->ast1Len > 0) {
313!
181
    SDB_SET_BINARY(pRaw, dataPos, pStb->pAst1, pStb->ast1Len, _OVER)
×
182
  }
183

184
  if (pStb->ast2Len > 0) {
313!
185
    SDB_SET_BINARY(pRaw, dataPos, pStb->pAst2, pStb->ast2Len, _OVER)
×
186
  }
187

188
  if (pStb->pCmpr != NULL) {
313!
189
    for (int i = 0; i < pStb->numOfColumns; i++) {
6,143✔
190
      SColCmpr *p = &pStb->pCmpr[i];
5,830✔
191
      SDB_SET_INT16(pRaw, dataPos, p->id, _OVER)
5,830!
192
      SDB_SET_INT32(pRaw, dataPos, p->alg, _OVER)
5,830!
193
    }
194
  }
195
  SDB_SET_INT64(pRaw, dataPos, pStb->keep, _OVER)
313!
196

197
  if (hasTypeMod) {
313✔
198
    for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
205✔
199
      SDB_SET_INT32(pRaw, dataPos, pStb->pExtSchemas[i].typeMod, _OVER);
166!
200
    }
201
  }
202

203
  SDB_SET_INT8(pRaw, dataPos, pStb->virtualStb, _OVER)
313!
204
  SDB_SET_RESERVE(pRaw, dataPos, STB_RESERVE_SIZE, _OVER)
313!
205
  SDB_SET_DATALEN(pRaw, dataPos, _OVER)
313!
206

207
  terrno = 0;
313✔
208

209
_OVER:
313✔
210
  if (terrno != 0) {
313!
211
    mError("stb:%s, failed to encode to raw:%p since %s", pStb->name, pRaw, terrstr());
×
212
    sdbFreeRaw(pRaw);
×
213
    return NULL;
×
214
  }
215

216
  mTrace("stb:%s, encode to raw:%p, row:%p", pStb->name, pRaw, pStb);
313✔
217
  return pRaw;
313✔
218
}
219

220
static SSdbRow *mndStbActionDecode(SSdbRaw *pRaw) {
280✔
221
  int32_t code = 0;
280✔
222
  int32_t lino = 0;
280✔
223
  terrno = TSDB_CODE_OUT_OF_MEMORY;
280✔
224
  SSdbRow *pRow = NULL;
280✔
225
  SStbObj *pStb = NULL;
280✔
226
  bool     hasExtSchemas = false;
280✔
227

228
  int8_t sver = 0;
280✔
229
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto _OVER;
280!
230

231
  if (sver > STB_VER_NUMBER) {
280!
232
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
×
233
    goto _OVER;
×
234
  }
235

236
  pRow = sdbAllocRow(sizeof(SStbObj));
280✔
237
  if (pRow == NULL) goto _OVER;
280!
238

239
  pStb = sdbGetRowObj(pRow);
280✔
240
  if (pStb == NULL) goto _OVER;
280!
241

242
  int32_t dataPos = 0;
280✔
243
  SDB_GET_BINARY(pRaw, dataPos, pStb->name, TSDB_TABLE_FNAME_LEN, _OVER)
280!
244
  SDB_GET_BINARY(pRaw, dataPos, pStb->db, TSDB_DB_FNAME_LEN, _OVER)
280!
245
  SDB_GET_INT64(pRaw, dataPos, &pStb->createdTime, _OVER)
280!
246
  SDB_GET_INT64(pRaw, dataPos, &pStb->updateTime, _OVER)
280!
247
  SDB_GET_INT64(pRaw, dataPos, &pStb->uid, _OVER)
280!
248
  SDB_GET_INT64(pRaw, dataPos, &pStb->dbUid, _OVER)
280!
249
  SDB_GET_INT32(pRaw, dataPos, &pStb->tagVer, _OVER)
280!
250
  SDB_GET_INT32(pRaw, dataPos, &pStb->colVer, _OVER)
280!
251
  SDB_GET_INT32(pRaw, dataPos, &pStb->smaVer, _OVER)
280!
252
  SDB_GET_INT32(pRaw, dataPos, &pStb->nextColId, _OVER)
280!
253
  SDB_GET_INT64(pRaw, dataPos, &pStb->maxdelay[0], _OVER)
280!
254
  SDB_GET_INT64(pRaw, dataPos, &pStb->maxdelay[1], _OVER)
280!
255
  SDB_GET_INT64(pRaw, dataPos, &pStb->watermark[0], _OVER)
280!
256
  SDB_GET_INT64(pRaw, dataPos, &pStb->watermark[1], _OVER)
280!
257
  SDB_GET_INT32(pRaw, dataPos, &pStb->ttl, _OVER)
280!
258
  SDB_GET_INT32(pRaw, dataPos, &pStb->numOfColumns, _OVER)
280!
259
  SDB_GET_INT32(pRaw, dataPos, &pStb->numOfTags, _OVER)
280!
260
  SDB_GET_INT32(pRaw, dataPos, &pStb->numOfFuncs, _OVER)
280!
261
  SDB_GET_INT32(pRaw, dataPos, &pStb->commentLen, _OVER)
280!
262
  SDB_GET_INT32(pRaw, dataPos, &pStb->ast1Len, _OVER)
280!
263
  SDB_GET_INT32(pRaw, dataPos, &pStb->ast2Len, _OVER)
280!
264

265
  pStb->pColumns = taosMemoryCalloc(pStb->numOfColumns, sizeof(SSchema));
280!
266
  pStb->pTags = taosMemoryCalloc(pStb->numOfTags, sizeof(SSchema));
280!
267
  pStb->pFuncs = taosArrayInit(pStb->numOfFuncs, TSDB_FUNC_NAME_LEN);
280✔
268
  if (pStb->pColumns == NULL || pStb->pTags == NULL || pStb->pFuncs == NULL) {
280!
269
    goto _OVER;
×
270
  }
271

272
  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
9,971✔
273
    SSchema *pSchema = &pStb->pColumns[i];
9,691✔
274
    SDB_GET_INT8(pRaw, dataPos, &pSchema->type, _OVER)
9,691!
275
    SDB_GET_INT8(pRaw, dataPos, &pSchema->flags, _OVER)
9,691!
276
    SDB_GET_INT16(pRaw, dataPos, &pSchema->colId, _OVER)
9,691!
277
    SDB_GET_INT32(pRaw, dataPos, &pSchema->bytes, _OVER)
9,691!
278
    SDB_GET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, _OVER)
9,691!
279
    hasExtSchemas = hasExtSchemas || HAS_TYPE_MOD(pSchema);
9,691✔
280
  }
281

282
  for (int32_t i = 0; i < pStb->numOfTags; ++i) {
1,423✔
283
    SSchema *pSchema = &pStb->pTags[i];
1,143✔
284
    SDB_GET_INT8(pRaw, dataPos, &pSchema->type, _OVER)
1,143!
285
    SDB_GET_INT8(pRaw, dataPos, &pSchema->flags, _OVER)
1,143!
286
    SDB_GET_INT16(pRaw, dataPos, &pSchema->colId, _OVER)
1,143!
287
    SDB_GET_INT32(pRaw, dataPos, &pSchema->bytes, _OVER)
1,143!
288
    SDB_GET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, _OVER)
1,143!
289
  }
290

291
  for (int32_t i = 0; i < pStb->numOfFuncs; ++i) {
280!
292
    char funcName[TSDB_FUNC_NAME_LEN] = {0};
×
293
    SDB_GET_BINARY(pRaw, dataPos, funcName, TSDB_FUNC_NAME_LEN, _OVER)
×
294
    if (taosArrayPush(pStb->pFuncs, funcName) == NULL) goto _OVER;
×
295
  }
296

297
  if (pStb->commentLen > 0) {
280✔
298
    pStb->comment = taosMemoryCalloc(pStb->commentLen + 1, 1);
40!
299
    if (pStb->comment == NULL) goto _OVER;
40!
300
    SDB_GET_BINARY(pRaw, dataPos, pStb->comment, pStb->commentLen + 1, _OVER)
40!
301
  }
302

303
  if (pStb->ast1Len > 0) {
280!
304
    pStb->pAst1 = taosMemoryCalloc(pStb->ast1Len, 1);
×
305
    if (pStb->pAst1 == NULL) goto _OVER;
×
306
    SDB_GET_BINARY(pRaw, dataPos, pStb->pAst1, pStb->ast1Len, _OVER)
×
307
  }
308

309
  if (pStb->ast2Len > 0) {
280!
310
    pStb->pAst2 = taosMemoryCalloc(pStb->ast2Len, 1);
×
311
    if (pStb->pAst2 == NULL) goto _OVER;
×
312
    SDB_GET_BINARY(pRaw, dataPos, pStb->pAst2, pStb->ast2Len, _OVER)
×
313
  }
314

315
  pStb->pCmpr = taosMemoryCalloc(pStb->numOfColumns, sizeof(SColCmpr));
280!
316
  if (sver < STB_VER_NUMBER - 1) {
280✔
317
    // compatible with old data, setup default compress value
318
    // impl later
319
    for (int i = 0; i < pStb->numOfColumns; i++) {
8,372✔
320
      SSchema  *pSchema = &pStb->pColumns[i];
8,334✔
321
      SColCmpr *pCmpr = &pStb->pCmpr[i];
8,334✔
322
      pCmpr->id = pSchema->colId;
8,334✔
323
      pCmpr->alg = createDefaultColCmprByType(pSchema->type);
8,334✔
324
    }
325
  } else {
326
    for (int i = 0; i < pStb->numOfColumns; i++) {
1,599✔
327
      SColCmpr *pCmpr = &pStb->pCmpr[i];
1,357✔
328
      SDB_GET_INT16(pRaw, dataPos, &pCmpr->id, _OVER)
1,357!
329
      SDB_GET_INT32(pRaw, dataPos, (int32_t *)&pCmpr->alg, _OVER)  // compatiable
1,357!
330
    }
331
  }
332
  SDB_GET_INT64(pRaw, dataPos, &pStb->keep, _OVER)
280!
333

334
  // type mod
335
  if (hasExtSchemas) {
280✔
336
    pStb->pExtSchemas = taosMemoryCalloc(pStb->numOfColumns, sizeof(SExtSchema));
38!
337
    if (!pStb->pExtSchemas) goto _OVER;
38!
338
    for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
200✔
339
      SSchema *pSchema = &pStb->pColumns[i];
162✔
340
      SDB_GET_INT32(pRaw, dataPos, &pStb->pExtSchemas[i].typeMod, _OVER)
162!
341
    }
342
  }
343

344
  if (sver < STB_VER_NUMBER) {
280✔
345
    pStb->virtualStb = 0;
38✔
346
  } else {
347
    SDB_GET_INT8(pRaw, dataPos, &pStb->virtualStb, _OVER)
242!
348
  }
349

350
  SDB_GET_RESERVE(pRaw, dataPos, STB_RESERVE_SIZE, _OVER)
280!
351

352
  terrno = 0;
280✔
353

354
_OVER:
280✔
355
  if (terrno != 0) {
280!
356
    mError("stb:%s, failed to decode from raw:%p since %s", pStb == NULL ? "null" : pStb->name, pRaw, terrstr());
×
357
    if (pStb != NULL) {
×
358
      taosMemoryFreeClear(pStb->pColumns);
×
359
      taosMemoryFreeClear(pStb->pTags);
×
360
      taosMemoryFreeClear(pStb->comment);
×
361
      taosMemoryFree(pStb->pCmpr);
×
362
      taosMemoryFreeClear(pStb->pExtSchemas);
×
363
    }
364
    taosMemoryFreeClear(pRow);
×
365
    return NULL;
×
366
  }
367

368
  mTrace("stb:%s, decode from raw:%p, row:%p", pStb->name, pRaw, pStb);
280✔
369
  return pRow;
280✔
370
}
371

372
void mndFreeStb(SStbObj *pStb) {
371✔
373
  taosArrayDestroy(pStb->pFuncs);
371✔
374
  taosMemoryFreeClear(pStb->pColumns);
371!
375
  taosMemoryFreeClear(pStb->pTags);
371!
376
  taosMemoryFreeClear(pStb->comment);
371!
377
  taosMemoryFreeClear(pStb->pAst1);
371!
378
  taosMemoryFreeClear(pStb->pAst2);
371!
379
  taosMemoryFreeClear(pStb->pCmpr);
371!
380
  taosMemoryFreeClear(pStb->pExtSchemas);
371!
381
}
371✔
382

383
static int32_t mndStbActionInsert(SSdb *pSdb, SStbObj *pStb) {
142✔
384
  mTrace("stb:%s, perform insert action, row:%p", pStb->name, pStb);
142✔
385
  return 0;
142✔
386
}
387

388
static int32_t mndStbActionDelete(SSdb *pSdb, SStbObj *pStb) {
368✔
389
  mTrace("stb:%s, perform delete action, row:%p", pStb->name, pStb);
368✔
390
  mndFreeStb(pStb);
368✔
391
  return 0;
368✔
392
}
393

394
static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew) {
104✔
395
  terrno = 0;
104✔
396
  mTrace("stb:%s, perform update action, old row:%p new row:%p", pOld->name, pOld, pNew);
104✔
397

398
  taosWLockLatch(&pOld->lock);
104✔
399
  int32_t numOfColumns = pOld->numOfColumns;
104✔
400
  if (pOld->numOfColumns < pNew->numOfColumns) {
104✔
401
    void *pColumns = taosMemoryMalloc(pNew->numOfColumns * sizeof(SSchema));
7!
402
    if (pColumns == NULL) {
7!
403
      goto END;
×
404
    }
405
    taosMemoryFree(pOld->pColumns);
7!
406
    pOld->pColumns = pColumns;
7✔
407
  }
408

409
  if (pOld->numOfTags < pNew->numOfTags) {
104✔
410
    void *pTags = taosMemoryMalloc(pNew->numOfTags * sizeof(SSchema));
5!
411
    if (pTags == NULL) {
5!
412
      goto END;
×
413
    }
414
    taosMemoryFree(pOld->pTags);
5!
415
    pOld->pTags = pTags;
5✔
416
  }
417

418
  if (pOld->commentLen < pNew->commentLen && pNew->commentLen > 0) {
104!
419
    void *comment = taosMemoryMalloc(pNew->commentLen + 1);
1!
420
    if (comment == NULL) {
1!
421
      goto END;
×
422
    }
423
    taosMemoryFree(pOld->comment);
1!
424
    pOld->comment = comment;
1✔
425
  }
426
  pOld->commentLen = pNew->commentLen;
104✔
427

428
  if (pOld->ast1Len < pNew->ast1Len) {
104!
429
    void *pAst1 = taosMemoryMalloc(pNew->ast1Len + 1);
×
430
    if (pAst1 == NULL) {
×
431
      goto END;
×
432
    }
433
    taosMemoryFree(pOld->pAst1);
×
434
    pOld->pAst1 = pAst1;
×
435
  }
436

437
  if (pOld->ast2Len < pNew->ast2Len) {
104!
438
    void *pAst2 = taosMemoryMalloc(pNew->ast2Len + 1);
×
439
    if (pAst2 == NULL) {
×
440
      goto END;
×
441
    }
442
    taosMemoryFree(pOld->pAst2);
×
443
    pOld->pAst2 = pAst2;
×
444
  }
445

446
  pOld->updateTime = pNew->updateTime;
104✔
447
  pOld->tagVer = pNew->tagVer;
104✔
448
  pOld->colVer = pNew->colVer;
104✔
449
  pOld->smaVer = pNew->smaVer;
104✔
450
  pOld->nextColId = pNew->nextColId;
104✔
451
  pOld->ttl = pNew->ttl;
104✔
452
  pOld->keep = pNew->keep;
104✔
453
  
454
  if (pNew->numOfColumns > 0) {
104!
455
    pOld->numOfColumns = pNew->numOfColumns;
104✔
456
    memcpy(pOld->pColumns, pNew->pColumns, pOld->numOfColumns * sizeof(SSchema));
104✔
457
  }
458
  if (pNew->numOfTags > 0) {
104!
459
    pOld->numOfTags = pNew->numOfTags;
104✔
460
    memcpy(pOld->pTags, pNew->pTags, pOld->numOfTags * sizeof(SSchema));
104✔
461
  }
462
  if (pNew->commentLen > 0) {
104✔
463
    memcpy(pOld->comment, pNew->comment, pNew->commentLen + 1);
37✔
464
    pOld->commentLen = pNew->commentLen;
37✔
465
  }
466
  if (pNew->ast1Len != 0) {
104!
467
    memcpy(pOld->pAst1, pNew->pAst1, pNew->ast1Len);
×
468
    pOld->ast1Len = pNew->ast1Len;
×
469
  }
470
  if (pNew->ast2Len != 0) {
104!
471
    memcpy(pOld->pAst2, pNew->pAst2, pNew->ast2Len);
×
472
    pOld->ast2Len = pNew->ast2Len;
×
473
  }
474
  if (numOfColumns < pNew->numOfColumns) {
104✔
475
    taosMemoryFree(pOld->pCmpr);
7!
476
    pOld->pCmpr = taosMemoryCalloc(pNew->numOfColumns, sizeof(SColCmpr));
7!
477
    if (pOld->pCmpr == NULL){
7!
478
      goto END;
×
479
    }
480
    memcpy(pOld->pCmpr, pNew->pCmpr, pNew->numOfColumns * sizeof(SColCmpr));
7✔
481
  } else {
482
    memcpy(pOld->pCmpr, pNew->pCmpr, pNew->numOfColumns * sizeof(SColCmpr));
97✔
483
  }
484

485
  if (pNew->pExtSchemas) {
104✔
486
    taosMemoryFreeClear(pOld->pExtSchemas);
37!
487
    pOld->pExtSchemas = taosMemoryCalloc(pNew->numOfColumns, sizeof(SExtSchema));
37!
488
    if (pOld->pExtSchemas == NULL){
37!
489
      goto END;
×
490
    }
491
    memcpy(pOld->pExtSchemas, pNew->pExtSchemas, pNew->numOfColumns * sizeof(SExtSchema));
37✔
492
  }
493

494
END:
67✔
495
  taosWUnLockLatch(&pOld->lock);
104✔
496
  return terrno;
104✔
497
}
498

499
SStbObj *mndAcquireStb(SMnode *pMnode, char *stbName) {
685✔
500
  SSdb    *pSdb = pMnode->pSdb;
685✔
501
  SStbObj *pStb = sdbAcquire(pSdb, SDB_STB, stbName);
685✔
502
  if (pStb == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
685!
503
    terrno = TSDB_CODE_MND_STB_NOT_EXIST;
107✔
504
  }
505
  return pStb;
685✔
506
}
507

508
void mndReleaseStb(SMnode *pMnode, SStbObj *pStb) {
687✔
509
  SSdb *pSdb = pMnode->pSdb;
687✔
510
  sdbRelease(pSdb, pStb);
687✔
511
}
687✔
512

513
SDbObj *mndAcquireDbByStb(SMnode *pMnode, const char *stbName) {
169✔
514
  SName name = {0};
169✔
515
  if ((terrno = tNameFromString(&name, stbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) return NULL;
169!
516

517
  char db[TSDB_TABLE_FNAME_LEN] = {0};
169✔
518
  if ((terrno = tNameGetFullDbName(&name, db)) != 0) return NULL;
169!
519

520
  return mndAcquireDb(pMnode, db);
169✔
521
}
522

523
static FORCE_INLINE int32_t schemaExColIdCompare(const void *colId, const void *pSchema) {
524
  if (*(col_id_t *)colId < ((SSchema *)pSchema)->colId) {
525
    return -1;
526
  } else if (*(col_id_t *)colId > ((SSchema *)pSchema)->colId) {
527
    return 1;
528
  }
529
  return 0;
530
}
531

532
void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen, void *alterOriData,
302✔
533
                            int32_t alterOriDataLen) {
534
  SEncoder       encoder = {0};
302✔
535
  int32_t        contLen;
536
  SName          name = {0};
302✔
537
  SVCreateStbReq req = {0};
302✔
538

539
  if ((terrno = tNameFromString(&name, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) {
302!
540
    goto _err;
×
541
  }
542
  char dbFName[TSDB_DB_FNAME_LEN] = {0};
302✔
543
  if ((terrno = tNameGetFullDbName(&name, dbFName)) != 0) {
302!
544
    goto _err;
×
545
  };
546

547
  req.name = (char *)tNameGetTableName(&name);
302✔
548
  req.suid = pStb->uid;
302✔
549
  req.rollup = pStb->ast1Len > 0 ? 1 : 0;
302✔
550
  req.alterOriData = alterOriData;
302✔
551
  req.alterOriDataLen = alterOriDataLen;
302✔
552
  req.source = pStb->source;
302✔
553
  req.virtualStb = pStb->virtualStb;
302✔
554
  // todo
555
  req.schemaRow.nCols = pStb->numOfColumns;
302✔
556
  req.schemaRow.version = pStb->colVer;
302✔
557
  req.schemaRow.pSchema = pStb->pColumns;
302✔
558
  req.schemaTag.nCols = pStb->numOfTags;
302✔
559
  req.schemaTag.version = pStb->tagVer;
302✔
560
  req.schemaTag.pSchema = pStb->pTags;
302✔
561

562
  req.colCmpred = 1;
302✔
563
  SColCmprWrapper *pCmpr = &req.colCmpr;
302✔
564
  req.keep = pStb->keep;
302✔
565
  pCmpr->version = pStb->colVer;
302✔
566
  pCmpr->nCols = pStb->numOfColumns;
302✔
567

568
  req.colCmpr.pColCmpr = taosMemoryCalloc(pCmpr->nCols, sizeof(SColCmpr));
302!
569
  for (int32_t i = 0; i < pStb->numOfColumns; i++) {
2,076✔
570
    SColCmpr *p = &pCmpr->pColCmpr[i];
1,774✔
571
    p->alg = pStb->pCmpr[i].alg;
1,774✔
572
    p->id = pStb->pCmpr[i].id;
1,774✔
573
  }
574

575
  if (req.rollup) {
302!
576
    req.rsmaParam.maxdelay[0] = pStb->maxdelay[0];
×
577
    req.rsmaParam.maxdelay[1] = pStb->maxdelay[1];
×
578
    req.rsmaParam.watermark[0] = pStb->watermark[0];
×
579
    req.rsmaParam.watermark[1] = pStb->watermark[1];
×
580
    if (pStb->ast1Len > 0) {
×
581
      if (mndConvertRsmaTask(&req.rsmaParam.qmsg[0], &req.rsmaParam.qmsgLen[0], pStb->pAst1, pStb->uid,
×
582
                             STREAM_TRIGGER_WINDOW_CLOSE, req.rsmaParam.watermark[0],
583
                             req.rsmaParam.deleteMark[0]) < 0) {
584
        goto _err;
×
585
      }
586
    }
587
    if (pStb->ast2Len > 0) {
×
588
      if (mndConvertRsmaTask(&req.rsmaParam.qmsg[1], &req.rsmaParam.qmsgLen[1], pStb->pAst2, pStb->uid,
×
589
                             STREAM_TRIGGER_WINDOW_CLOSE, req.rsmaParam.watermark[1],
590
                             req.rsmaParam.deleteMark[1]) < 0) {
591
        goto _err;
×
592
      }
593
    }
594
  }
595
  req.pExtSchemas = pStb->pExtSchemas; // only reference to it.
302✔
596
  // get length
597
  int32_t ret = 0;
302✔
598
  tEncodeSize(tEncodeSVCreateStbReq, &req, contLen, ret);
302!
599
  if (ret < 0) {
302!
600
    goto _err;
×
601
  }
602

603
  contLen += sizeof(SMsgHead);
302✔
604

605
  SMsgHead *pHead = taosMemoryCalloc(1, contLen);
302!
606
  if (pHead == NULL) {
302!
607
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
608
    goto _err;
×
609
  }
610

611
  pHead->contLen = htonl(contLen);
302✔
612
  pHead->vgId = htonl(pVgroup->vgId);
302✔
613

614
  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
302✔
615
  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
302✔
616
  if (tEncodeSVCreateStbReq(&encoder, &req) < 0) {
302!
617
    taosMemoryFreeClear(pHead);
×
618
    tEncoderClear(&encoder);
×
619
    goto _err;
×
620
  }
621
  tEncoderClear(&encoder);
302✔
622

623
  *pContLen = contLen;
302✔
624
  taosMemoryFreeClear(req.rsmaParam.qmsg[0]);
302!
625
  taosMemoryFreeClear(req.rsmaParam.qmsg[1]);
302!
626
  taosMemoryFreeClear(req.colCmpr.pColCmpr);
302!
627
  return pHead;
302✔
628
_err:
×
629
  taosMemoryFreeClear(req.rsmaParam.qmsg[0]);
×
630
  taosMemoryFreeClear(req.rsmaParam.qmsg[1]);
×
631
  taosMemoryFreeClear(req.colCmpr.pColCmpr);
×
632
  return NULL;
×
633
}
634

635
static void *mndBuildVDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen) {
232✔
636
  SName        name = {0};
232✔
637
  SVDropStbReq req = {0};
232✔
638
  int32_t      contLen = 0;
232✔
639
  int32_t      ret = 0;
232✔
640
  SMsgHead    *pHead = NULL;
232✔
641
  SEncoder     encoder = {0};
232✔
642

643
  if ((terrno = tNameFromString(&name, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) {
232!
644
    return NULL;
×
645
  }
646

647
  req.name = (char *)tNameGetTableName(&name);
232✔
648
  req.suid = pStb->uid;
232✔
649

650
  tEncodeSize(tEncodeSVDropStbReq, &req, contLen, ret);
232!
651
  if (ret < 0) return NULL;
232!
652

653
  contLen += sizeof(SMsgHead);
232✔
654
  pHead = taosMemoryMalloc(contLen);
232!
655
  if (pHead == NULL) {
232!
656
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
657
    return NULL;
×
658
  }
659

660
  pHead->contLen = htonl(contLen);
232✔
661
  pHead->vgId = htonl(pVgroup->vgId);
232✔
662

663
  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
232✔
664

665
  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
232✔
666
  int32_t code = tEncodeSVDropStbReq(&encoder, &req);
232✔
667
  tEncoderClear(&encoder);
232✔
668
  if (code != 0) {
232!
669
    terrno = code;
×
670
    return NULL;
×
671
  }
672

673
  *pContLen = contLen;
232✔
674
  return pHead;
232✔
675
}
676

677
int32_t mndCheckCreateStbReq(SMCreateStbReq *pCreate) {
96✔
678
  int32_t code = 0;
96✔
679
  if (pCreate->igExists < 0 || pCreate->igExists > 1) {
96!
680
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
681
    TAOS_RETURN(code);
×
682
  }
683

684
  if (pCreate->virtualStb != 0 && pCreate->virtualStb != 1) {
96!
685
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
686
    TAOS_RETURN(code);
×
687
  }
688

689
  if (pCreate->numOfColumns < TSDB_MIN_COLUMNS || pCreate->numOfTags + pCreate->numOfColumns > TSDB_MAX_COLUMNS) {
96!
690
    code = TSDB_CODE_PAR_INVALID_COLUMNS_NUM;
×
691
    TAOS_RETURN(code);
×
692
  }
693

694
  if (pCreate->numOfTags <= 0 || pCreate->numOfTags > TSDB_MAX_TAGS) {
96!
695
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
696
    TAOS_RETURN(code);
×
697
  }
698

699
  SField *pField = taosArrayGet(pCreate->pColumns, 0);
96✔
700
  if (pField->type != TSDB_DATA_TYPE_TIMESTAMP) {
96!
701
    code = TSDB_CODE_PAR_INVALID_FIRST_COLUMN;
×
702
    TAOS_RETURN(code);
×
703
  }
704

705
  for (int32_t i = 0; i < pCreate->numOfColumns; ++i) {
583✔
706
    SFieldWithOptions *pField1 = taosArrayGet(pCreate->pColumns, i);
487✔
707
    if (pField1->type >= TSDB_DATA_TYPE_MAX) {
487!
708
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
709
      TAOS_RETURN(code);
×
710
    }
711
    if (pField1->bytes <= 0) {
487!
712
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
713
      TAOS_RETURN(code);
×
714
    }
715
    if (pField1->name[0] == 0) {
487!
716
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
717
      TAOS_RETURN(code);
×
718
    }
719
  }
720

721
  for (int32_t i = 0; i < pCreate->numOfTags; ++i) {
344✔
722
    SField *pField1 = taosArrayGet(pCreate->pTags, i);
248✔
723
    if (pField1->type >= TSDB_DATA_TYPE_MAX) {
248!
724
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
725
      TAOS_RETURN(code);
×
726
    }
727
    if (pField1->bytes <= 0) {
248!
728
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
729
      TAOS_RETURN(code);
×
730
    }
731
    if (pField1->name[0] == 0) {
248!
732
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
733
      TAOS_RETURN(code);
×
734
    }
735
  }
736

737
  TAOS_RETURN(code);
96✔
738
}
739

740
static int32_t mndSetCreateStbPrepareLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
×
741
  int32_t  code = 0;
×
742
  SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);
×
743
  if (pRedoRaw == NULL) {
×
744
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
745
    if (terrno != 0) code = terrno;
×
746
    TAOS_RETURN(code);
×
747
  }
748
  if ((code = mndTransAppendPrepareLog(pTrans, pRedoRaw)) != 0) {
×
749
    sdbFreeRaw(pRedoRaw);
×
750
    TAOS_RETURN(code);
×
751
  }
752
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING));
×
753

754
  TAOS_RETURN(code);
×
755
}
756

757
static int32_t mndSetCreateStbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
94✔
758
  int32_t  code = 0;
94✔
759
  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);
94✔
760
  if (pCommitRaw == NULL) {
94!
761
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
762
    if (terrno != 0) code = terrno;
×
763
    TAOS_RETURN(code);
×
764
  }
765
  if ((code = mndTransAppendCommitlog(pTrans, pCommitRaw)) != 0) {
94!
766
    sdbFreeRaw(pCommitRaw);
×
767
    TAOS_RETURN(code);
×
768
  }
769
  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));
94!
770

771
  TAOS_RETURN(code);
94✔
772
}
773

774
static int32_t mndSetCreateStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
94✔
775
  int32_t code = 0;
94✔
776
  SSdb   *pSdb = pMnode->pSdb;
94✔
777
  SVgObj *pVgroup = NULL;
94✔
778
  void   *pIter = NULL;
94✔
779
  int32_t contLen;
780

781
  while (1) {
309✔
782
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
403✔
783
    if (pIter == NULL) break;
403✔
784
    if (!mndVgroupInDb(pVgroup, pDb->uid)) {
309✔
785
      sdbRelease(pSdb, pVgroup);
87✔
786
      continue;
87✔
787
    }
788

789
    void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, NULL, 0);
222✔
790
    if (pReq == NULL) {
222!
791
      sdbCancelFetch(pSdb, pIter);
×
792
      sdbRelease(pSdb, pVgroup);
×
793
      code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
794
      if (terrno != 0) code = terrno;
×
795
      TAOS_RETURN(code);
×
796
    }
797

798
    STransAction action = {0};
222✔
799
    action.mTraceId = pTrans->mTraceId;
222✔
800
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
222✔
801
    action.pCont = pReq;
222✔
802
    action.contLen = contLen;
222✔
803
    action.msgType = TDMT_VND_CREATE_STB;
222✔
804
    action.acceptableCode = TSDB_CODE_TDB_STB_ALREADY_EXIST;
222✔
805
    action.retryCode = TSDB_CODE_TDB_STB_NOT_EXIST;
222✔
806
    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
222!
807
      taosMemoryFree(pReq);
×
808
      sdbCancelFetch(pSdb, pIter);
×
809
      sdbRelease(pSdb, pVgroup);
×
810
      TAOS_RETURN(code);
×
811
    }
812
    sdbRelease(pSdb, pVgroup);
222✔
813
  }
814

815
  TAOS_RETURN(code);
94✔
816
}
817

818
int32_t mndSetForceDropCreateStbRedoActions(SMnode *pMnode, STrans *pTrans, SVgObj *pVgroup, SStbObj *pStb) {
×
819
  int32_t code = 0;
×
820
  SSdb   *pSdb = pMnode->pSdb;
×
821
  int32_t contLen;
822

823
  void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, NULL, 0);
×
824
  if (pReq == NULL) {
×
825
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
826
    if (terrno != 0) code = terrno;
×
827
    TAOS_RETURN(code);
×
828
  }
829

830
  STransAction action = {0};
×
831
  action.mTraceId = pTrans->mTraceId;
×
832
  action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
×
833
  action.pCont = pReq;
×
834
  action.contLen = contLen;
×
835
  action.msgType = TDMT_VND_CREATE_STB;
×
836
  action.acceptableCode = TSDB_CODE_TDB_STB_ALREADY_EXIST;
×
837
  action.retryCode = TSDB_CODE_TDB_STB_NOT_EXIST;
×
838
  if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
×
839
    taosMemoryFree(pReq);
×
840
    TAOS_RETURN(code);
×
841
  }
842

843
  TAOS_RETURN(code);
×
844
}
845

846
static int32_t mndSetCreateStbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
94✔
847
  int32_t code = 0;
94✔
848
  SSdb   *pSdb = pMnode->pSdb;
94✔
849
  SVgObj *pVgroup = NULL;
94✔
850
  void   *pIter = NULL;
94✔
851

852
  while (1) {
309✔
853
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
403✔
854
    if (pIter == NULL) break;
403✔
855
    if (!mndVgroupInDb(pVgroup, pDb->uid)) {
309✔
856
      sdbRelease(pSdb, pVgroup);
87✔
857
      continue;
87✔
858
    }
859

860
    int32_t contLen = 0;
222✔
861
    void   *pReq = mndBuildVDropStbReq(pMnode, pVgroup, pStb, &contLen);
222✔
862
    if (pReq == NULL) {
222!
863
      sdbCancelFetch(pSdb, pIter);
×
864
      sdbRelease(pSdb, pVgroup);
×
865
      code = TSDB_CODE_OUT_OF_MEMORY;
×
866
      TAOS_RETURN(code);
×
867
    }
868

869
    STransAction action = {0};
222✔
870
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
222✔
871
    action.pCont = pReq;
222✔
872
    action.contLen = contLen;
222✔
873
    action.msgType = TDMT_VND_DROP_STB;
222✔
874
    action.acceptableCode = TSDB_CODE_TDB_STB_NOT_EXIST;
222✔
875
    if ((code = mndTransAppendUndoAction(pTrans, &action)) != 0) {
222!
876
      taosMemoryFree(pReq);
×
877
      sdbCancelFetch(pSdb, pIter);
×
878
      sdbRelease(pSdb, pVgroup);
×
879
      TAOS_RETURN(code);
×
880
    }
881
    sdbRelease(pSdb, pVgroup);
222✔
882
  }
883

884
  TAOS_RETURN(code);
94✔
885
}
886

887
static SSchema *mndFindStbColumns(const SStbObj *pStb, const char *colName) {
×
888
  for (int32_t col = 0; col < pStb->numOfColumns; ++col) {
×
889
    SSchema *pSchema = &pStb->pColumns[col];
×
890
    if (taosStrncasecmp(pSchema->name, colName, TSDB_COL_NAME_LEN) == 0) {
×
891
      return pSchema;
×
892
    }
893
  }
894
  return NULL;
×
895
}
896

897
int32_t mndBuildStbFromReq(SMnode *pMnode, SStbObj *pDst, SMCreateStbReq *pCreate, SDbObj *pDb) {
94✔
898
  int32_t code = 0;
94✔
899
  bool    hasTypeMods = false;
94✔
900
  memcpy(pDst->name, pCreate->name, TSDB_TABLE_FNAME_LEN);
94✔
901
  memcpy(pDst->db, pDb->name, TSDB_DB_FNAME_LEN);
94✔
902
  pDst->createdTime = taosGetTimestampMs();
94✔
903
  pDst->updateTime = pDst->createdTime;
94✔
904
  pDst->uid = (pCreate->source == TD_REQ_FROM_TAOX_OLD || pCreate->source == TD_REQ_FROM_TAOX)
94!
905
                  ? pCreate->suid
906
                  : mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN);
188!
907
  pDst->dbUid = pDb->uid;
94✔
908
  pDst->tagVer = 1;
94✔
909
  pDst->colVer = 1;
94✔
910
  pDst->smaVer = 1;
94✔
911
  pDst->nextColId = 1;
94✔
912
  pDst->maxdelay[0] = pCreate->delay1;
94✔
913
  pDst->maxdelay[1] = pCreate->delay2;
94✔
914
  pDst->watermark[0] = pCreate->watermark1;
94✔
915
  pDst->watermark[1] = pCreate->watermark2;
94✔
916
  pDst->ttl = pCreate->ttl;
94✔
917
  pDst->numOfColumns = pCreate->numOfColumns;
94✔
918
  pDst->numOfTags = pCreate->numOfTags;
94✔
919
  pDst->numOfFuncs = pCreate->numOfFuncs;
94✔
920
  pDst->commentLen = pCreate->commentLen;
94✔
921
  pDst->pFuncs = pCreate->pFuncs;
94✔
922
  pDst->source = pCreate->source;
94✔
923
  pDst->keep = pCreate->keep;
94✔
924
  pDst->virtualStb = pCreate->virtualStb;
94✔
925
  pCreate->pFuncs = NULL;
94✔
926

927
  if (pDst->commentLen > 0) {
94✔
928
    pDst->comment = taosMemoryCalloc(pDst->commentLen + 1, 1);
2!
929
    if (pDst->comment == NULL) {
2!
930
      code = terrno;
×
931
      TAOS_RETURN(code);
×
932
    }
933
    memcpy(pDst->comment, pCreate->pComment, pDst->commentLen + 1);
2✔
934
  }
935

936
  pDst->ast1Len = pCreate->ast1Len;
94✔
937
  if (pDst->ast1Len > 0) {
94!
938
    pDst->pAst1 = taosMemoryCalloc(pDst->ast1Len, 1);
×
939
    if (pDst->pAst1 == NULL) {
×
940
      code = terrno;
×
941
      TAOS_RETURN(code);
×
942
    }
943
    memcpy(pDst->pAst1, pCreate->pAst1, pDst->ast1Len);
×
944
  }
945

946
  pDst->ast2Len = pCreate->ast2Len;
94✔
947
  if (pDst->ast2Len > 0) {
94!
948
    pDst->pAst2 = taosMemoryCalloc(pDst->ast2Len, 1);
×
949
    if (pDst->pAst2 == NULL) {
×
950
      code = terrno;
×
951
      TAOS_RETURN(code);
×
952
    }
953
    memcpy(pDst->pAst2, pCreate->pAst2, pDst->ast2Len);
×
954
  }
955

956
  pDst->pColumns = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SSchema));
94!
957
  pDst->pTags = taosMemoryCalloc(1, pDst->numOfTags * sizeof(SSchema));
94!
958
  if (pDst->pColumns == NULL || pDst->pTags == NULL) {
94!
959
    code = terrno;
×
960
    TAOS_RETURN(code);
×
961
  }
962

963
  if (pDst->nextColId < 0 || pDst->nextColId >= 0x7fff - pDst->numOfColumns - pDst->numOfTags) {
94!
964
    code = TSDB_CODE_OUT_OF_RANGE;
×
965
    TAOS_RETURN(code);
×
966
  }
967

968
  for (int32_t i = 0; i < pDst->numOfColumns; ++i) {
576✔
969
    SFieldWithOptions *pField = taosArrayGet(pCreate->pColumns, i);
482✔
970
    SSchema           *pSchema = &pDst->pColumns[i];
482✔
971
    pSchema->type = pField->type;
482✔
972
    pSchema->bytes = pField->bytes;
482✔
973
    pSchema->flags = pField->flags;
482✔
974
    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
482✔
975
    pSchema->colId = pDst->nextColId;
482✔
976
    pDst->nextColId++;
482✔
977
    hasTypeMods = hasTypeMods || HAS_TYPE_MOD(pSchema);
482!
978
  }
979

980
  for (int32_t i = 0; i < pDst->numOfTags; ++i) {
337✔
981
    SField  *pField = taosArrayGet(pCreate->pTags, i);
243✔
982
    SSchema *pSchema = &pDst->pTags[i];
243✔
983
    pSchema->type = pField->type;
243✔
984
    pSchema->bytes = pField->bytes;
243✔
985
    if (i == 0) {
243✔
986
      SSCHMEA_SET_IDX_ON(pSchema);
94✔
987
    }
988
    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
243✔
989
    pSchema->colId = pDst->nextColId;
243✔
990
    pDst->nextColId++;
243✔
991
  }
992
  // set col compress
993
  pDst->pCmpr = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SCmprObj));
94!
994
  for (int32_t i = 0; i < pDst->numOfColumns; i++) {
576✔
995
    SFieldWithOptions *pField = taosArrayGet(pCreate->pColumns, i);
482✔
996
    SSchema           *pSchema = &pDst->pColumns[i];
482✔
997

998
    SColCmpr *pColCmpr = &pDst->pCmpr[i];
482✔
999
    pColCmpr->id = pSchema->colId;
482✔
1000
    pColCmpr->alg = pField->compress;
482✔
1001
  }
1002

1003
  if (hasTypeMods) {
94!
1004
    pDst->pExtSchemas = taosMemoryCalloc(pDst->numOfColumns, sizeof(SExtSchema));
×
1005
    if (!pDst->pExtSchemas) {
×
1006
      code = terrno;
×
1007
      TAOS_RETURN(code);
×
1008
    }
1009
    for (int32_t i = 0; i < pDst->numOfColumns; ++i) {
×
1010
      SFieldWithOptions * pField = taosArrayGet(pCreate->pColumns, i);
×
1011
      pDst->pExtSchemas[i].typeMod = pField->typeMod;
×
1012
    }
1013
  }
1014
  TAOS_RETURN(code);
94✔
1015
}
1016
static int32_t mndGenIdxNameForFirstTag(char *fullname, char *dbname, char *stbname, char *tagname) {
91✔
1017
  SName name = {0};
91✔
1018
  if ((terrno = tNameFromString(&name, stbname, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) {
91!
1019
    return -1;
×
1020
  }
1021
  return snprintf(fullname, TSDB_INDEX_FNAME_LEN, "%s.%s_%s", dbname, tagname, tNameGetTableName(&name));
91✔
1022
}
1023

1024
static int32_t mndCreateStb(SMnode *pMnode, SRpcMsg *pReq, SMCreateStbReq *pCreate, SDbObj *pDb) {
91✔
1025
  SStbObj stbObj = {0};
91✔
1026
  int32_t code = -1;
91✔
1027

1028
  char fullIdxName[TSDB_INDEX_FNAME_LEN * 2] = {0};
91✔
1029

1030
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pReq, "create-stb");
91✔
1031
  if (pTrans == NULL) {
91!
1032
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
1033
    if (terrno != 0) code = terrno;
×
1034
    goto _OVER;
×
1035
  }
1036

1037
  mInfo("trans:%d, used to create stb:%s", pTrans->id, pCreate->name);
91!
1038
  TAOS_CHECK_GOTO(mndBuildStbFromReq(pMnode, &stbObj, pCreate, pDb), NULL, _OVER);
91!
1039

1040
  SSchema *pSchema = &(stbObj.pTags[0]);
91✔
1041
  if (mndGenIdxNameForFirstTag(fullIdxName, pDb->name, stbObj.name, pSchema->name) < 0) {
91!
1042
    goto _OVER;
×
1043
  }
1044
  SSIdx idx = {0};
91✔
1045
  if (mndAcquireGlobalIdx(pMnode, fullIdxName, SDB_IDX, &idx) == 0 && idx.pIdx != NULL) {
91!
1046
    code = TSDB_CODE_MND_TAG_INDEX_ALREADY_EXIST;
×
1047
    mndReleaseIdx(pMnode, idx.pIdx);
×
1048
    goto _OVER;
×
1049
  }
1050

1051
  SIdxObj idxObj = {0};
91✔
1052
  memcpy(idxObj.name, fullIdxName, TSDB_INDEX_FNAME_LEN);
91✔
1053
  memcpy(idxObj.stb, stbObj.name, TSDB_TABLE_FNAME_LEN);
91✔
1054
  memcpy(idxObj.db, stbObj.db, TSDB_DB_FNAME_LEN);
91✔
1055
  memcpy(idxObj.colName, pSchema->name, TSDB_COL_NAME_LEN);
91✔
1056
  idxObj.createdTime = taosGetTimestampMs();
91✔
1057
  idxObj.uid = mndGenerateUid(fullIdxName, strlen(fullIdxName));
91✔
1058
  idxObj.stbUid = stbObj.uid;
91✔
1059
  idxObj.dbUid = stbObj.dbUid;
91✔
1060

1061
  TAOS_CHECK_GOTO(mndSetCreateIdxCommitLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
91!
1062

1063
  TAOS_CHECK_GOTO(mndAddStbToTrans(pMnode, pTrans, pDb, &stbObj), NULL, _OVER);
91!
1064
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
91!
1065
  code = 0;
91✔
1066

1067
_OVER:
91✔
1068
  mndTransDrop(pTrans);
91✔
1069
  if (mndStbActionDelete(pMnode->pSdb, &stbObj) != 0) mError("failed to mndStbActionDelete");
91!
1070
  TAOS_RETURN(code);
91✔
1071
}
1072

1073
int32_t mndAddStbToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
94✔
1074
  mndTransSetDbName(pTrans, pDb->name, pStb->name);
94✔
1075
  TAOS_CHECK_RETURN(mndTransCheckConflict(pMnode, pTrans));
94!
1076
  TAOS_CHECK_RETURN(mndSetCreateStbCommitLogs(pMnode, pTrans, pDb, pStb));
94!
1077
  TAOS_CHECK_RETURN(mndSetCreateStbRedoActions(pMnode, pTrans, pDb, pStb));
94!
1078
  TAOS_CHECK_RETURN(mndSetCreateStbUndoActions(pMnode, pTrans, pDb, pStb));
94!
1079
  return 0;
94✔
1080
}
1081

1082
static int32_t mndProcessTtlTimer(SRpcMsg *pReq) {
148✔
1083
  SMnode           *pMnode = pReq->info.node;
148✔
1084
  SSdb             *pSdb = pMnode->pSdb;
148✔
1085
  SVgObj           *pVgroup = NULL;
148✔
1086
  void             *pIter = NULL;
148✔
1087
  SVDropTtlTableReq ttlReq = {
148✔
1088
      .timestampSec = taosGetTimestampSec(), .ttlDropMaxCount = tsTtlBatchDropNum, .nUids = 0, .pTbUids = NULL};
148✔
1089
  int32_t reqLen = tSerializeSVDropTtlTableReq(NULL, 0, &ttlReq);
148✔
1090
  int32_t contLen = reqLen + sizeof(SMsgHead);
148✔
1091

1092
  mDebug("start to process ttl timer");
148✔
1093

1094
  while (1) {
511✔
1095
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
659✔
1096
    if (pIter == NULL) break;
659✔
1097

1098
    int32_t   code = 0;
511✔
1099
    SMsgHead *pHead = rpcMallocCont(contLen);
511✔
1100
    if (pHead == NULL) {
511!
1101
      sdbRelease(pSdb, pVgroup);
×
1102
      continue;
×
1103
    }
1104
    pHead->contLen = htonl(contLen);
511✔
1105
    pHead->vgId = htonl(pVgroup->vgId);
511✔
1106
    if ((code = tSerializeSVDropTtlTableReq((char *)pHead + sizeof(SMsgHead), reqLen, &ttlReq)) < 0) {
511!
1107
      mError("vgId:%d, failed to serialize drop ttl table request since %s", pVgroup->vgId, tstrerror(code));
×
1108
      sdbRelease(pSdb, pVgroup);
×
1109
      continue;
×
1110
    }
1111

1112
    SRpcMsg rpcMsg = {
511✔
1113
        .msgType = TDMT_VND_FETCH_TTL_EXPIRED_TBS, .pCont = pHead, .contLen = contLen, .info = pReq->info};
1114
    SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup);
511✔
1115
    code = tmsgSendReq(&epSet, &rpcMsg);
511✔
1116
    if (code != 0) {
511!
1117
      mError("vgId:%d, failed to send drop ttl table request to vnode since 0x%x", pVgroup->vgId, code);
×
1118
    } else {
1119
      mDebug("vgId:%d, send drop ttl table request to vnode, time:%" PRId32, pVgroup->vgId, ttlReq.timestampSec);
511✔
1120
    }
1121
    sdbRelease(pSdb, pVgroup);
511✔
1122
  }
1123

1124
  return 0;
148✔
1125
}
1126

1127
static int32_t mndProcessTrimDbTimer(SRpcMsg *pReq) {
×
1128
  SMnode     *pMnode = pReq->info.node;
×
1129
  SSdb       *pSdb = pMnode->pSdb;
×
1130
  SVgObj     *pVgroup = NULL;
×
1131
  void       *pIter = NULL;
×
1132
  SVTrimDbReq trimReq = {.timestamp = taosGetTimestampSec()};
×
1133
  int32_t     reqLen = tSerializeSVTrimDbReq(NULL, 0, &trimReq);
×
1134
  int32_t     contLen = reqLen + sizeof(SMsgHead);
×
1135

1136
  while (1) {
×
1137
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
×
1138
    if (pIter == NULL) break;
×
1139

1140
    int32_t code = 0;
×
1141

1142
    SMsgHead *pHead = rpcMallocCont(contLen);
×
1143
    if (pHead == NULL) {
×
1144
      sdbCancelFetch(pSdb, pVgroup);
×
1145
      sdbRelease(pSdb, pVgroup);
×
1146
      continue;
×
1147
    }
1148
    pHead->contLen = htonl(contLen);
×
1149
    pHead->vgId = htonl(pVgroup->vgId);
×
1150
    if ((code = tSerializeSVTrimDbReq((char *)pHead + sizeof(SMsgHead), reqLen, &trimReq)) < 0) {
×
1151
      mError("vgId:%d, failed to serialize trim db request since %s", pVgroup->vgId, tstrerror(code));
×
1152
    }
1153

1154
    SRpcMsg rpcMsg = {.msgType = TDMT_VND_TRIM, .pCont = pHead, .contLen = contLen};
×
1155
    SEpSet  epSet = mndGetVgroupEpset(pMnode, pVgroup);
×
1156
    code = tmsgSendReq(&epSet, &rpcMsg);
×
1157
    if (code != 0) {
×
1158
      mError("vgId:%d, timer failed to send vnode-trim request to vnode since 0x%x", pVgroup->vgId, code);
×
1159
    } else {
1160
      mInfo("vgId:%d, timer send vnode-trim request to vnode, time:%d", pVgroup->vgId, trimReq.timestamp);
×
1161
    }
1162
    sdbRelease(pSdb, pVgroup);
×
1163
  }
1164

1165
  return 0;
×
1166
}
1167

1168
static int32_t mndProcessS3MigrateDbTimer(SRpcMsg *pReq) {
×
1169
  SMnode          *pMnode = pReq->info.node;
×
1170
  SSdb            *pSdb = pMnode->pSdb;
×
1171
  SVgObj          *pVgroup = NULL;
×
1172
  void            *pIter = NULL;
×
1173
  SVS3MigrateDbReq s3migrateReq = {.timestamp = taosGetTimestampSec()};
×
1174
  int32_t          reqLen = tSerializeSVS3MigrateDbReq(NULL, 0, &s3migrateReq);
×
1175
  int32_t          contLen = reqLen + sizeof(SMsgHead);
×
1176

1177
  while (1) {
×
1178
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
×
1179
    if (pIter == NULL) break;
×
1180

1181
    int32_t code = 0;
×
1182

1183
    SMsgHead *pHead = rpcMallocCont(contLen);
×
1184
    if (pHead == NULL) {
×
1185
      sdbRelease(pSdb, pVgroup);
×
1186
      continue;
×
1187
    }
1188
    pHead->contLen = htonl(contLen);
×
1189
    pHead->vgId = htonl(pVgroup->vgId);
×
1190
    if ((code = tSerializeSVS3MigrateDbReq((char *)pHead + sizeof(SMsgHead), reqLen, &s3migrateReq)) < 0) {
×
1191
      mError("vgId:%d, failed to serialize s3migrate db request since %s", pVgroup->vgId, tstrerror(code));
×
1192
      sdbRelease(pSdb, pVgroup);
×
1193
      continue;
×
1194
    }
1195

1196
    SRpcMsg rpcMsg = {.msgType = TDMT_VND_S3MIGRATE, .pCont = pHead, .contLen = contLen};
×
1197
    SEpSet  epSet = mndGetVgroupEpset(pMnode, pVgroup);
×
1198
    code = tmsgSendReq(&epSet, &rpcMsg);
×
1199
    if (code != 0) {
×
1200
      mError("vgId:%d, timer failed to send vnode-s3migrate request to vnode since 0x%x", pVgroup->vgId, code);
×
1201
    } else {
1202
      mInfo("vgId:%d, timer send vnode-s3migrate request to vnode, time:%d", pVgroup->vgId, s3migrateReq.timestamp);
×
1203
    }
1204
    sdbRelease(pSdb, pVgroup);
×
1205
  }
1206

1207
  return 0;
×
1208
}
1209

1210
static int32_t mndFindSuperTableTagIndex(const SStbObj *pStb, const char *tagName) {
35✔
1211
  for (int32_t tag = 0; tag < pStb->numOfTags; tag++) {
126✔
1212
    if (strcmp(pStb->pTags[tag].name, tagName) == 0) {
107✔
1213
      return tag;
16✔
1214
    }
1215
  }
1216

1217
  return -1;
19✔
1218
}
1219

1220
static int32_t mndFindSuperTableColumnIndex(const SStbObj *pStb, const char *colName) {
35✔
1221
  for (int32_t col = 0; col < pStb->numOfColumns; col++) {
106✔
1222
    if (strcmp(pStb->pColumns[col].name, colName) == 0) {
87✔
1223
      return col;
16✔
1224
    }
1225
  }
1226

1227
  return -1;
19✔
1228
}
1229

1230
static bool mndValidateSchema(SSchema *pSchemas, int32_t nSchema, SArray *pFields, int32_t maxLen) {
18✔
1231
  int32_t rowLen = 0;
18✔
1232
  for (int32_t i = 0; i < nSchema; ++i) {
75✔
1233
    rowLen += (pSchemas + i)->bytes;
57✔
1234
  }
1235

1236
  int32_t nField = taosArrayGetSize(pFields);
18✔
1237
  for (int32_t i = 0; i < nField; ++i) {
36✔
1238
    rowLen += ((SField *)TARRAY_GET_ELEM(pFields, i))->bytes;
18✔
1239
  }
1240

1241
  return rowLen <= maxLen;
18✔
1242
}
1243

1244
static int32_t mndBuildStbFromAlter(SStbObj *pStb, SStbObj *pDst, SMCreateStbReq *createReq) {
×
1245
  int32_t code = 0;
×
1246
  taosRLockLatch(&pStb->lock);
×
1247
  memcpy(pDst, pStb, sizeof(SStbObj));
×
1248
  taosRUnLockLatch(&pStb->lock);
×
1249

1250
  pDst->source = createReq->source;
×
1251
  pDst->updateTime = taosGetTimestampMs();
×
1252
  pDst->numOfColumns = createReq->numOfColumns;
×
1253
  pDst->numOfTags = createReq->numOfTags;
×
1254
  pDst->pColumns = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SSchema));
×
1255
  pDst->pTags = taosMemoryCalloc(1, pDst->numOfTags * sizeof(SSchema));
×
1256
  pDst->pCmpr = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SColCmpr));
×
1257
  pDst->pExtSchemas = taosMemoryCalloc(pDst->numOfColumns, sizeof(SExtSchema));
×
1258

1259
  if (pDst->pColumns == NULL || pDst->pTags == NULL || pDst->pCmpr == NULL || pDst->pExtSchemas == NULL) {
×
1260
    code = terrno;
×
1261
    TAOS_RETURN(code);
×
1262
  }
1263

1264
  if (pDst->nextColId < 0 || pDst->nextColId >= 0x7fff - pDst->numOfColumns - pDst->numOfTags) {
×
1265
    code = TSDB_CODE_OUT_OF_RANGE;
×
1266
    TAOS_RETURN(code);
×
1267
  }
1268

1269
  for (int32_t i = 0; i < pDst->numOfColumns; ++i) {
×
1270
    SFieldWithOptions *pField = taosArrayGet(createReq->pColumns, i);
×
1271
    SSchema           *pSchema = &pDst->pColumns[i];
×
1272
    pSchema->type = pField->type;
×
1273
    pSchema->bytes = pField->bytes;
×
1274
    pSchema->flags = pField->flags;
×
1275
    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
×
1276
    int32_t cIndex = mndFindSuperTableColumnIndex(pStb, pField->name);
×
1277
    if (cIndex >= 0) {
×
1278
      pSchema->colId = pStb->pColumns[cIndex].colId;
×
1279
    } else {
1280
      pSchema->colId = pDst->nextColId++;
×
1281
    }
1282
  }
1283

1284
  for (int32_t i = 0; i < pDst->numOfTags; ++i) {
×
1285
    SField  *pField = taosArrayGet(createReq->pTags, i);
×
1286
    SSchema *pSchema = &pDst->pTags[i];
×
1287
    pSchema->type = pField->type;
×
1288
    pSchema->bytes = pField->bytes;
×
1289
    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
×
1290
    int32_t cIndex = mndFindSuperTableTagIndex(pStb, pField->name);
×
1291
    if (cIndex >= 0) {
×
1292
      pSchema->colId = pStb->pTags[cIndex].colId;
×
1293
    } else {
1294
      pSchema->colId = pDst->nextColId++;
×
1295
    }
1296
  }
1297
  for (int32_t i = 0; i < pDst->numOfColumns; i++) {
×
1298
    SColCmpr          *p = pDst->pCmpr + i;
×
1299
    SFieldWithOptions *pField = taosArrayGet(createReq->pColumns, i);
×
1300
    SSchema           *pSchema = &pDst->pColumns[i];
×
1301
    p->id = pSchema->colId;
×
1302
    if (pField->compress == 0) {
×
1303
      p->alg = createDefaultColCmprByType(pSchema->type);
×
1304
    } else {
1305
      p->alg = pField->compress;
×
1306
    }
1307
    if (pField->flags & COL_HAS_TYPE_MOD) {
×
1308
      pDst->pExtSchemas[i].typeMod = pField->typeMod;
×
1309
    }
1310
  }
1311
  pDst->tagVer = createReq->tagVer;
×
1312
  pDst->colVer = createReq->colVer;
×
1313
  return TSDB_CODE_SUCCESS;
×
1314
}
1315

1316
static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) {
93✔
1317
  SMnode        *pMnode = pReq->info.node;
93✔
1318
  int32_t        code = -1;
93✔
1319
  SStbObj       *pStb = NULL;
93✔
1320
  SDbObj        *pDb = NULL;
93✔
1321
  SMCreateStbReq createReq = {0};
93✔
1322
  bool           isAlter = false;
93✔
1323
  SHashObj      *pHash = NULL;
93✔
1324

1325
  if (tDeserializeSMCreateStbReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
93!
1326
    code = TSDB_CODE_INVALID_MSG;
×
1327
    goto _OVER;
×
1328
  }
1329

1330
  mInfo("stb:%s, start to create", createReq.name);
93!
1331
  if (mndCheckCreateStbReq(&createReq) != 0) {
93!
1332
    code = TSDB_CODE_INVALID_MSG;
×
1333
    goto _OVER;
×
1334
  }
1335

1336
  pStb = mndAcquireStb(pMnode, createReq.name);
93✔
1337
  if (pStb != NULL) {
93✔
1338
    if (createReq.igExists) {
1!
1339
      if (createReq.source == TD_REQ_FROM_APP) {
×
1340
        mInfo("stb:%s, already exist, ignore exist is set", createReq.name);
×
1341
        code = 0;
×
1342
        goto _OVER;
×
1343
      } else if (pStb->uid != createReq.suid) {
×
1344
        mInfo("stb:%s, alter table does not need to be done, because table is deleted", createReq.name);
×
1345
        code = 0;
×
1346
        goto _OVER;
×
1347
      } else if (createReq.tagVer > 0 || createReq.colVer > 0) {
×
1348
        int32_t tagDelta = createReq.tagVer - pStb->tagVer;
×
1349
        int32_t colDelta = createReq.colVer - pStb->colVer;
×
1350
        mInfo("stb:%s, already exist while create, input tagVer:%d colVer:%d, exist tagVer:%d colVer:%d",
×
1351
              createReq.name, createReq.tagVer, createReq.colVer, pStb->tagVer, pStb->colVer);
1352
        if (tagDelta <= 0 && colDelta <= 0) {
×
1353
          mInfo("stb:%s, schema version is not incremented and nothing needs to be done", createReq.name);
×
1354
          code = 0;
×
1355
          goto _OVER;
×
1356
        } else if ((tagDelta == 1 && colDelta == 0) || (tagDelta == 0 && colDelta == 1) ||
×
1357
                   (pStb->colVer == 1 && createReq.colVer > 1) || (pStb->tagVer == 1 && createReq.tagVer > 1)) {
×
1358
          isAlter = true;
×
1359
          mInfo("stb:%s, schema version is only increased by 1 number, do alter operation", createReq.name);
×
1360
        } else {
1361
          mError("stb:%s, schema version increase more than 1 number, error is returned", createReq.name);
×
1362
          code = TSDB_CODE_MND_INVALID_SCHEMA_VER;
×
1363
          goto _OVER;
×
1364
        }
1365
      } else {
1366
        mError("stb:%s, already exist while create, input tagVer:%d colVer:%d is invalid, origin tagVer:%d colVer:%d",
×
1367
               createReq.name, createReq.tagVer, createReq.colVer, pStb->tagVer, pStb->colVer);
1368
        code = TSDB_CODE_MND_INVALID_SCHEMA_VER;
×
1369
        goto _OVER;
×
1370
      }
1371
    } else {
1372
      code = TSDB_CODE_MND_STB_ALREADY_EXIST;
1✔
1373
      goto _OVER;
1✔
1374
    }
1375
  } else if (terrno != TSDB_CODE_MND_STB_NOT_EXIST) {
92!
1376
    goto _OVER;
×
1377
  } else if ((createReq.source == TD_REQ_FROM_TAOX_OLD || createReq.source == TD_REQ_FROM_TAOX) &&
92!
1378
             (createReq.tagVer != 1 || createReq.colVer != 1)) {
×
1379
    mInfo("stb:%s, alter table does not need to be done, because table is deleted", createReq.name);
×
1380
    code = 0;
×
1381
    goto _OVER;
×
1382
  }
1383

1384
  pHash = taosHashInit(createReq.numOfColumns + createReq.numOfTags, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY),
92✔
1385
                       false, HASH_NO_LOCK);
1386
  if (pHash == NULL) {
92!
1387
    code = TSDB_CODE_OUT_OF_MEMORY;
×
1388
    goto _OVER;
×
1389
  }
1390

1391
  for (int32_t i = 0; i < createReq.numOfColumns; ++i) {
566✔
1392
    SFieldWithOptions *pField = taosArrayGet(createReq.pColumns, i);
474✔
1393
    if ((code = taosHashPut(pHash, pField->name, strlen(pField->name), NULL, 0)) != 0) {
474!
1394
      if (code == TSDB_CODE_DUP_KEY) {
×
1395
        code = TSDB_CODE_TSC_DUP_COL_NAMES;
×
1396
      }
1397
      goto _OVER;
×
1398
    }
1399
  }
1400

1401
  for (int32_t i = 0; i < createReq.numOfTags; ++i) {
335✔
1402
    SField *pField = taosArrayGet(createReq.pTags, i);
244✔
1403
    if ((code = taosHashPut(pHash, pField->name, strlen(pField->name), NULL, 0)) != 0) {
244✔
1404
      if (code == TSDB_CODE_DUP_KEY) {
1!
1405
        code = TSDB_CODE_TSC_DUP_COL_NAMES;
1✔
1406
      }
1407
      goto _OVER;
1✔
1408
    }
1409
  }
1410

1411
  pDb = mndAcquireDbByStb(pMnode, createReq.name);
91✔
1412
  if (pDb == NULL) {
91!
1413
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
1414
    goto _OVER;
×
1415
  }
1416

1417
  if ((code = mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pDb)) != 0) {
91!
1418
    goto _OVER;
×
1419
  }
1420

1421
  int32_t numOfStbs = -1;
91✔
1422
  if ((code = mndGetNumOfStbs(pMnode, pDb->name, &numOfStbs)) != 0) {
91!
1423
    goto _OVER;
×
1424
  }
1425

1426
  if (pDb->cfg.numOfStables == 1 && numOfStbs != 0) {
91!
1427
    code = TSDB_CODE_MND_SINGLE_STB_MODE_DB;
×
1428
    goto _OVER;
×
1429
  }
1430

1431
  if ((code = grantCheck(TSDB_GRANT_STABLE)) < 0) {
91!
1432
    goto _OVER;
×
1433
  }
1434

1435
  if (isAlter) {
91!
1436
    bool    needRsp = false;
×
1437
    SStbObj pDst = {0};
×
1438
    if ((code = mndBuildStbFromAlter(pStb, &pDst, &createReq)) != 0) {
×
1439
      taosMemoryFreeClear(pDst.pTags);
×
1440
      taosMemoryFreeClear(pDst.pColumns);
×
1441
      taosMemoryFreeClear(pDst.pCmpr);
×
1442
      taosMemoryFreeClear(pDst.pExtSchemas);
×
1443
      goto _OVER;
×
1444
    }
1445

1446
    code = mndAlterStbImp(pMnode, pReq, pDb, &pDst, needRsp, NULL, 0);
×
1447
    taosMemoryFreeClear(pDst.pTags);
×
1448
    taosMemoryFreeClear(pDst.pColumns);
×
1449
    taosMemoryFreeClear(pDst.pCmpr);
×
1450
    taosMemoryFreeClear(pDst.pExtSchemas);
×
1451
  } else {
1452
    code = mndCreateStb(pMnode, pReq, &createReq, pDb);
91✔
1453
  }
1454
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
91!
1455

1456
  SName name = {0};
91✔
1457
  TAOS_CHECK_RETURN(tNameFromString(&name, createReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
91!
1458

1459
  if (createReq.sql == NULL && createReq.sqlLen == 0) {
100!
1460
    char detail[1000] = {0};
9✔
1461

1462
    (void)tsnprintf(detail, sizeof(detail), "dbname:%s, stable name:%s", name.dbname, name.tname);
9✔
1463

1464
    auditRecord(pReq, pMnode->clusterId, "createStb", name.dbname, name.tname, detail, strlen(detail));
9✔
1465
  } else {
1466
    auditRecord(pReq, pMnode->clusterId, "createStb", name.dbname, name.tname, createReq.sql, createReq.sqlLen);
82✔
1467
  }
1468
_OVER:
93✔
1469
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
93!
1470
    mError("stb:%s, failed to create since %s", createReq.name, tstrerror(code));
2!
1471
  }
1472

1473
  mndReleaseStb(pMnode, pStb);
93✔
1474
  mndReleaseDb(pMnode, pDb);
93✔
1475
  tFreeSMCreateStbReq(&createReq);
93✔
1476

1477
  if (pHash != NULL) {
93✔
1478
    taosHashCleanup(pHash);
92✔
1479
  }
1480

1481
  TAOS_RETURN(code);
93✔
1482
}
1483

1484
static int32_t mndCheckAlterStbReq(SMAlterStbReq *pAlter) {
53✔
1485
  int32_t code = 0;
53✔
1486
  if (pAlter->commentLen >= 0) return 0;
53!
1487
  if (pAlter->ttl != 0) return 0;
×
1488
  if (pAlter->keep != -1) return 0;
×
1489

1490
  if (pAlter->numOfFields < 1 || pAlter->numOfFields != (int32_t)taosArrayGetSize(pAlter->pFields)) {
×
1491
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
1492
    TAOS_RETURN(code);
×
1493
  }
1494

1495
  for (int32_t i = 0; i < pAlter->numOfFields; ++i) {
×
1496
    SField *pField = taosArrayGet(pAlter->pFields, i);
×
1497
    if (pField->name[0] == 0) {
×
1498
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
1499
      TAOS_RETURN(code);
×
1500
    }
1501
  }
1502

1503
  TAOS_RETURN(code);
×
1504
}
1505

1506
int32_t mndAllocStbSchemas(const SStbObj *pOld, SStbObj *pNew) {
64✔
1507
  pNew->pTags = taosMemoryCalloc(pNew->numOfTags, sizeof(SSchema));
64!
1508
  pNew->pColumns = taosMemoryCalloc(pNew->numOfColumns, sizeof(SSchema));
64!
1509
  pNew->pCmpr = taosMemoryCalloc(pNew->numOfColumns, sizeof(SColCmpr));
64!
1510
  if (pNew->pTags == NULL || pNew->pColumns == NULL || pNew->pCmpr == NULL) {
64!
1511
    TAOS_RETURN(terrno);
×
1512
  }
1513

1514
  memcpy(pNew->pColumns, pOld->pColumns, sizeof(SSchema) * pOld->numOfColumns);
64✔
1515
  memcpy(pNew->pTags, pOld->pTags, sizeof(SSchema) * pOld->numOfTags);
64✔
1516
  memcpy(pNew->pCmpr, pOld->pCmpr, sizeof(SColCmpr) * pOld->numOfColumns);
64✔
1517
  if (pOld->pExtSchemas) {
64✔
1518
    pNew->pExtSchemas = taosMemoryCalloc(pNew->numOfColumns, sizeof(SExtSchema));
16!
1519
    if (pNew->pExtSchemas == NULL) {
16!
1520
      TAOS_RETURN(terrno);
×
1521
    }
1522
    memcpy(pNew->pExtSchemas, pOld->pExtSchemas, sizeof(SExtSchema) * pOld->numOfColumns);
16✔
1523
  }
1524

1525
  TAOS_RETURN(0);
64✔
1526
}
1527

1528
static int32_t mndUpdateTableOptions(const SStbObj *pOld, SStbObj *pNew, char *pComment, int32_t commentLen,
1✔
1529
                                     int32_t ttl, int64_t keep) {
1530
  int32_t code = 0;
1✔
1531
  if (commentLen > 0) {
1!
1532
    pNew->commentLen = commentLen;
1✔
1533
    pNew->comment = taosMemoryCalloc(1, commentLen + 1);
1!
1534
    if (pNew->comment == NULL) {
1!
1535
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
1536
      return -1;
×
1537
    }
1538
    memcpy(pNew->comment, pComment, commentLen + 1);
1✔
1539
  } else if (commentLen == 0) {
×
1540
    pNew->commentLen = 0;
×
1541
  } else {
1542
  }
1543

1544
  if (ttl >= 0) {
1!
1545
    pNew->ttl = ttl;
1✔
1546
  }
1547

1548
  if (keep > 0) {
1!
1549
    pNew->keep = keep;
×
1550
  }
1551

1552
  if ((code = mndAllocStbSchemas(pOld, pNew)) != 0) {
1!
1553
    TAOS_RETURN(code);
×
1554
  }
1555
  TAOS_RETURN(code);
1✔
1556
}
1557

1558
static int32_t mndAddSuperTableTag(const SStbObj *pOld, SStbObj *pNew, SArray *pFields, int32_t ntags) {
7✔
1559
  int32_t code = 0;
7✔
1560
  if (pOld->numOfTags + ntags > TSDB_MAX_TAGS) {
7!
1561
    code = TSDB_CODE_MND_TOO_MANY_TAGS;
×
1562
    TAOS_RETURN(code);
×
1563
  }
1564

1565
  if (pOld->numOfColumns + ntags + pOld->numOfTags > TSDB_MAX_COLUMNS) {
7!
1566
    code = TSDB_CODE_MND_TOO_MANY_COLUMNS;
×
1567
    TAOS_RETURN(code);
×
1568
  }
1569

1570
  if (!mndValidateSchema(pOld->pTags, pOld->numOfTags, pFields, TSDB_MAX_TAGS_LEN)) {
7!
1571
    code = TSDB_CODE_PAR_INVALID_TAGS_LENGTH;
×
1572
    TAOS_RETURN(code);
×
1573
  }
1574

1575
  pNew->numOfTags = pNew->numOfTags + ntags;
7✔
1576
  if ((code = mndAllocStbSchemas(pOld, pNew)) != 0) {
7!
1577
    TAOS_RETURN(code);
×
1578
  }
1579

1580
  if (pNew->nextColId < 0 || pNew->nextColId >= 0x7fff - ntags) {
7!
1581
    code = TSDB_CODE_OUT_OF_RANGE;
×
1582
    TAOS_RETURN(code);
×
1583
  }
1584

1585
  for (int32_t i = 0; i < ntags; i++) {
12✔
1586
    SField *pField = taosArrayGet(pFields, i);
7✔
1587
    if (mndFindSuperTableColumnIndex(pOld, pField->name) >= 0) {
7✔
1588
      code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
1✔
1589
      TAOS_RETURN(code);
1✔
1590
    }
1591

1592
    if (mndFindSuperTableTagIndex(pOld, pField->name) >= 0) {
6✔
1593
      code = TSDB_CODE_MND_TAG_ALREADY_EXIST;
1✔
1594
      TAOS_RETURN(code);
1✔
1595
    }
1596

1597
    SSchema *pSchema = &pNew->pTags[pOld->numOfTags + i];
5✔
1598
    pSchema->bytes = pField->bytes;
5✔
1599
    pSchema->type = pField->type;
5✔
1600
    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
5✔
1601
    pSchema->colId = pNew->nextColId;
5✔
1602
    pNew->nextColId++;
5✔
1603

1604
    mInfo("stb:%s, start to add tag %s", pNew->name, pSchema->name);
5!
1605
  }
1606

1607
  pNew->tagVer++;
5✔
1608
  TAOS_RETURN(code);
5✔
1609
}
1610

1611
static int32_t mndCheckAlterColForTopic(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {
20✔
1612
  int32_t code = 0;
20✔
1613
  SSdb   *pSdb = pMnode->pSdb;
20✔
1614
  void   *pIter = NULL;
20✔
1615
  while (1) {
×
1616
    SMqTopicObj *pTopic = NULL;
20✔
1617
    pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);
20✔
1618
    if (pIter == NULL) break;
20!
1619

1620
    mInfo("topic:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d, subType:%d sql:%s",
×
1621
          pTopic->name, stbFullName, suid, colId, pTopic->subType, pTopic->sql);
1622
    if (pTopic->ast == NULL) {
×
1623
      sdbRelease(pSdb, pTopic);
×
1624
      continue;
×
1625
    }
1626

1627
    SNode *pAst = NULL;
×
1628
    if (nodesStringToNode(pTopic->ast, &pAst) != 0) {
×
1629
      code = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC;
×
1630
      mError("topic:%s, create ast error", pTopic->name);
×
1631
      sdbRelease(pSdb, pTopic);
×
1632
      sdbCancelFetch(pSdb, pIter);
×
1633
      TAOS_RETURN(code);
×
1634
    }
1635

1636
    SNodeList *pNodeList = NULL;
×
1637
    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=
×
1638
        0) {
1639
      sdbRelease(pSdb, pTopic);
×
1640
      sdbCancelFetch(pSdb, pIter);
×
1641
      TAOS_RETURN(code);
×
1642
    }
1643
    SNode *pNode = NULL;
×
1644
    FOREACH(pNode, pNodeList) {
×
1645
      SColumnNode *pCol = (SColumnNode *)pNode;
×
1646
      mInfo("topic:%s, check colId:%d tableId:%" PRId64 " ctbStbUid:%" PRId64, pTopic->name, pCol->colId, pCol->tableId,
×
1647
            pTopic->ctbStbUid);
1648

1649
      if (pCol->tableId != suid && pTopic->ctbStbUid != suid) {
×
1650
        mInfo("topic:%s, check colId:%d passed", pTopic->name, pCol->colId);
×
1651
        goto NEXT;
×
1652
      }
1653
      if (pCol->colId > 0 && pCol->colId == colId) {
×
1654
        code = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC;
×
1655
        mError("topic:%s, check colId:%d conflicted", pTopic->name, pCol->colId);
×
1656
        nodesDestroyNode(pAst);
×
1657
        nodesDestroyList(pNodeList);
×
1658
        sdbCancelFetch(pSdb, pIter);
×
1659
        sdbRelease(pSdb, pTopic);
×
1660
        TAOS_RETURN(code);
×
1661
      }
1662
      mInfo("topic:%s, check colId:%d passed", pTopic->name, pCol->colId);
×
1663
    }
1664

1665
  NEXT:
×
1666
    sdbRelease(pSdb, pTopic);
×
1667
    nodesDestroyNode(pAst);
×
1668
    nodesDestroyList(pNodeList);
×
1669
  }
1670
  TAOS_RETURN(code);
20✔
1671
}
1672

1673
static int32_t mndCheckAlterColForStream(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {
20✔
1674
  int32_t code = 0;
20✔
1675
  SSdb   *pSdb = pMnode->pSdb;
20✔
1676
  void   *pIter = NULL;
20✔
1677
  while (1) {
×
1678
    SStreamObj *pStream = NULL;
20✔
1679
    pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);
20✔
1680
    if (pIter == NULL) break;
20!
1681

1682
    SNode *pAst = NULL;
×
1683
    if (nodesStringToNode(pStream->ast, &pAst) != 0) {
×
1684
      code = TSDB_CODE_MND_INVALID_STREAM_OPTION;
×
1685
      mError("stream:%s, create ast error", pStream->name);
×
1686
      sdbRelease(pSdb, pStream);
×
1687
      sdbCancelFetch(pSdb, pIter);
×
1688
      TAOS_RETURN(code);
×
1689
    }
1690

1691
    SNodeList *pNodeList = NULL;
×
1692
    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=
×
1693
        0) {
1694
      sdbRelease(pSdb, pStream);
×
1695
      sdbCancelFetch(pSdb, pIter);
×
1696
      TAOS_RETURN(code);
×
1697
    }
1698

1699
    SNode *pNode = NULL;
×
1700
    FOREACH(pNode, pNodeList) {
×
1701
      SColumnNode *pCol = (SColumnNode *)pNode;
×
1702

1703
      if (pCol->tableId != suid) {
×
1704
        mInfo("stream:%s, check colId:%d passed", pStream->name, pCol->colId);
×
1705
        goto NEXT;
×
1706
      }
1707
      if (pCol->colId > 0 && pCol->colId == colId) {
×
1708
        code = TSDB_CODE_MND_STREAM_MUST_BE_DELETED;
×
1709
        mError("stream:%s, check colId:%d conflicted", pStream->name, pCol->colId);
×
1710
        nodesDestroyNode(pAst);
×
1711
        nodesDestroyList(pNodeList);
×
1712
        sdbRelease(pSdb, pStream);
×
1713
        sdbCancelFetch(pSdb, pIter);
×
1714
        TAOS_RETURN(code);
×
1715
      }
1716
      mInfo("stream:%s, check colId:%d passed", pStream->name, pCol->colId);
×
1717
    }
1718

1719
  NEXT:
×
1720
    sdbRelease(pSdb, pStream);
×
1721
    nodesDestroyNode(pAst);
×
1722
    nodesDestroyList(pNodeList);
×
1723
  }
1724
  TAOS_RETURN(code);
20✔
1725
}
1726

1727
static int32_t mndCheckAlterColForTSma(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {
20✔
1728
  int32_t code = 0;
20✔
1729
  SSdb   *pSdb = pMnode->pSdb;
20✔
1730
  void   *pIter = NULL;
20✔
1731
  while (1) {
×
1732
    SSmaObj *pSma = NULL;
20✔
1733
    pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);
20✔
1734
    if (pIter == NULL) break;
20!
1735

1736
    mInfo("tsma:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d, sql:%s", pSma->name, stbFullName,
×
1737
          suid, colId, pSma->sql);
1738

1739
    SNode *pAst = NULL;
×
1740
    if (nodesStringToNode(pSma->ast, &pAst) != 0) {
×
1741
      code = TSDB_CODE_SDB_INVALID_DATA_CONTENT;
×
1742
      mError("tsma:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d failed since parse AST err",
×
1743
             pSma->name, stbFullName, suid, colId);
1744
      sdbCancelFetch(pSdb, pIter);
×
1745
      TAOS_RETURN(code);
×
1746
    }
1747

1748
    SNodeList *pNodeList = NULL;
×
1749
    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=
×
1750
        0) {
1751
      sdbCancelFetch(pSdb, pIter);
×
1752
      TAOS_RETURN(code);
×
1753
    }
1754
    SNode *pNode = NULL;
×
1755
    FOREACH(pNode, pNodeList) {
×
1756
      SColumnNode *pCol = (SColumnNode *)pNode;
×
1757
      mInfo("tsma:%s, check colId:%d tableId:%" PRId64, pSma->name, pCol->colId, pCol->tableId);
×
1758

1759
      if ((pCol->tableId != suid) && (pSma->stbUid != suid)) {
×
1760
        mInfo("tsma:%s, check colId:%d passed", pSma->name, pCol->colId);
×
1761
        goto NEXT;
×
1762
      }
1763
      if ((pCol->colId) > 0 && (pCol->colId == colId)) {
×
1764
        code = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TSMA;
×
1765
        mError("tsma:%s, check colId:%d conflicted", pSma->name, pCol->colId);
×
1766
        nodesDestroyNode(pAst);
×
1767
        nodesDestroyList(pNodeList);
×
1768
        sdbRelease(pSdb, pSma);
×
1769
        sdbCancelFetch(pSdb, pIter);
×
1770
        TAOS_RETURN(code);
×
1771
      }
1772
      mInfo("tsma:%s, check colId:%d passed", pSma->name, pCol->colId);
×
1773
    }
1774

1775
  NEXT:
×
1776
    sdbRelease(pSdb, pSma);
×
1777
    nodesDestroyNode(pAst);
×
1778
    nodesDestroyList(pNodeList);
×
1779
  }
1780
  TAOS_RETURN(code);
20✔
1781
}
1782

1783
int32_t mndCheckColAndTagModifiable(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId) {
20✔
1784
  TAOS_CHECK_RETURN(mndCheckAlterColForTopic(pMnode, stbFullName, suid, colId));
20!
1785
  TAOS_CHECK_RETURN(mndCheckAlterColForStream(pMnode, stbFullName, suid, colId));
20!
1786
  TAOS_CHECK_RETURN(mndCheckAlterColForTSma(pMnode, stbFullName, suid, colId));
20!
1787
  TAOS_RETURN(0);
20✔
1788
}
1789

1790
static int32_t mndDropSuperTableTag(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const char *tagName) {
4✔
1791
  int32_t code = 0;
4✔
1792
  int32_t tag = mndFindSuperTableTagIndex(pOld, tagName);
4✔
1793
  if (tag < 0) {
4✔
1794
    code = TSDB_CODE_MND_TAG_NOT_EXIST;
1✔
1795
    TAOS_RETURN(code);
1✔
1796
  }
1797

1798
  col_id_t colId = pOld->pTags[tag].colId;
3✔
1799
  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));
3!
1800

1801
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
3!
1802

1803
  memmove(pNew->pTags + tag, pNew->pTags + tag + 1, sizeof(SSchema) * (pNew->numOfTags - tag - 1));
3✔
1804
  pNew->numOfTags--;
3✔
1805

1806
  pNew->tagVer++;
3✔
1807

1808
  // if (mndDropIndexByTag(pMnode, pOld, tagName) != 0) {
1809
  //   return -1;
1810
  // }
1811
  mInfo("stb:%s, start to drop tag %s", pNew->name, tagName);
3!
1812
  TAOS_RETURN(code);
3✔
1813
}
1814

1815
static int32_t mndAlterStbTagName(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, SArray *pFields) {
7✔
1816
  int32_t code = 0;
7✔
1817
  if ((int32_t)taosArrayGetSize(pFields) != 2) {
7!
1818
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
1819
    TAOS_RETURN(code);
×
1820
  }
1821

1822
  SField *pField0 = taosArrayGet(pFields, 0);
7✔
1823
  SField *pField1 = taosArrayGet(pFields, 1);
7✔
1824

1825
  const char *oldTagName = pField0->name;
7✔
1826
  const char *newTagName = pField1->name;
7✔
1827

1828
  int32_t tag = mndFindSuperTableTagIndex(pOld, oldTagName);
7✔
1829
  if (tag < 0) {
7✔
1830
    code = TSDB_CODE_MND_TAG_NOT_EXIST;
2✔
1831
    TAOS_RETURN(code);
2✔
1832
  }
1833

1834
  col_id_t colId = pOld->pTags[tag].colId;
5✔
1835
  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));
5!
1836

1837
  if (mndFindSuperTableTagIndex(pOld, newTagName) >= 0) {
5✔
1838
    code = TSDB_CODE_MND_TAG_ALREADY_EXIST;
2✔
1839
    TAOS_RETURN(code);
2✔
1840
  }
1841

1842
  if (mndFindSuperTableColumnIndex(pOld, newTagName) >= 0) {
3✔
1843
    code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
1✔
1844
    TAOS_RETURN(code);
1✔
1845
  }
1846

1847
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
2!
1848

1849
  SSchema *pSchema = (SSchema *)(pNew->pTags + tag);
2✔
1850
  memcpy(pSchema->name, newTagName, TSDB_COL_NAME_LEN);
2✔
1851

1852
  pNew->tagVer++;
2✔
1853
  mInfo("stb:%s, start to modify tag %s to %s", pNew->name, oldTagName, newTagName);
2!
1854
  TAOS_RETURN(code);
2✔
1855
}
1856

1857
static int32_t mndAlterStbTagBytes(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const SField *pField) {
5✔
1858
  int32_t code = 0;
5✔
1859
  int32_t tag = mndFindSuperTableTagIndex(pOld, pField->name);
5✔
1860
  if (tag < 0) {
5✔
1861
    code = TSDB_CODE_MND_TAG_NOT_EXIST;
1✔
1862
    TAOS_RETURN(code);
1✔
1863
  }
1864

1865
  col_id_t colId = pOld->pTags[tag].colId;
4✔
1866
  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));
4!
1867

1868
  uint32_t nLen = 0;
4✔
1869
  for (int32_t i = 0; i < pOld->numOfTags; ++i) {
17✔
1870
    nLen += (pOld->pTags[i].colId == colId) ? pField->bytes : pOld->pTags[i].bytes;
13✔
1871
  }
1872

1873
  if (nLen > TSDB_MAX_TAGS_LEN) {
4!
1874
    code = TSDB_CODE_PAR_INVALID_TAGS_LENGTH;
×
1875
    TAOS_RETURN(code);
×
1876
  }
1877

1878
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
4!
1879

1880
  SSchema *pTag = pNew->pTags + tag;
4✔
1881

1882
  if (!(pTag->type == TSDB_DATA_TYPE_BINARY || pTag->type == TSDB_DATA_TYPE_VARBINARY ||
4!
1883
        pTag->type == TSDB_DATA_TYPE_NCHAR || pTag->type == TSDB_DATA_TYPE_GEOMETRY)) {
1!
1884
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
1✔
1885
    TAOS_RETURN(code);
1✔
1886
  }
1887

1888
  if (pField->bytes <= pTag->bytes) {
3✔
1889
    code = TSDB_CODE_MND_INVALID_ROW_BYTES;
1✔
1890
    TAOS_RETURN(code);
1✔
1891
  }
1892

1893
  pTag->bytes = pField->bytes;
2✔
1894
  pNew->tagVer++;
2✔
1895

1896
  mInfo("stb:%s, start to modify tag len %s to %d", pNew->name, pField->name, pField->bytes);
2!
1897
  TAOS_RETURN(code);
2✔
1898
}
1899

1900
static int32_t mndUpdateSuperTableColumnCompress(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, SArray *pField,
×
1901
                                                 int32_t nCols) {
1902
  // if (pColCmpr == NULL || colName == NULL) return -1;
1903

1904
  if (taosArrayGetSize(pField) != nCols) return TSDB_CODE_FAILED;
×
1905
  TAOS_FIELD *p = taosArrayGet(pField, 0);
×
1906

1907
  int32_t code = 0;
×
1908
  int32_t idx = mndFindSuperTableColumnIndex(pOld, p->name);
×
1909
  if (idx == -1) {
×
1910
    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;
×
1911
    TAOS_RETURN(code);
×
1912
  }
1913
  SSchema *pTarget = &pOld->pColumns[idx];
×
1914
  col_id_t colId = pTarget->colId;
×
1915

1916
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
×
1917
  code = validColCmprByType(pTarget->type, p->bytes);
×
1918
  if (code != TSDB_CODE_SUCCESS) {
×
1919
    TAOS_RETURN(code);
×
1920
  }
1921

1922
  int8_t updated = 0;
×
1923
  for (int i = 0; i < pNew->numOfColumns; i++) {
×
1924
    SColCmpr *pCmpr = &pNew->pCmpr[i];
×
1925
    if (pCmpr->id == colId) {
×
1926
      uint32_t dst = 0;
×
1927
      updated = tUpdateCompress(pCmpr->alg, p->bytes, TSDB_COLVAL_COMPRESS_DISABLED, TSDB_COLVAL_LEVEL_DISABLED,
×
1928
                                TSDB_COLVAL_LEVEL_MEDIUM, &dst);
1929
      if (updated > 0) pCmpr->alg = dst;
×
1930
      break;
×
1931
    }
1932
  }
1933

1934
  if (updated == 0) {
×
1935
    code = TSDB_CODE_MND_COLUMN_COMPRESS_ALREADY_EXIST;
×
1936
    TAOS_RETURN(code);
×
1937
  } else if (updated == -1) {
×
1938
    code = TSDB_CODE_TSC_COMPRESS_LEVEL_ERROR;
×
1939
    TAOS_RETURN(code);
×
1940
  }
1941

1942
  pNew->colVer++;
×
1943

1944
  TAOS_RETURN(code);
×
1945
}
1946
static int32_t mndAddSuperTableColumn(const SStbObj *pOld, SStbObj *pNew, const SMAlterStbReq* pReq, int32_t ncols,
11✔
1947
                                      int8_t withCompress) {
1948
  int32_t code = 0;
11✔
1949
  if (pOld->numOfColumns + ncols + pOld->numOfTags > TSDB_MAX_COLUMNS) {
11!
1950
    code = TSDB_CODE_MND_TOO_MANY_COLUMNS;
×
1951
    TAOS_RETURN(code);
×
1952
  }
1953

1954
  if ((code = grantCheck(TSDB_GRANT_TIMESERIES)) != 0) {
11!
1955
    TAOS_RETURN(code);
×
1956
  }
1957

1958
  if (!mndValidateSchema(pOld->pColumns, pOld->numOfColumns, pReq->pFields, TSDB_MAX_BYTES_PER_ROW)) {
11!
1959
    code = TSDB_CODE_PAR_INVALID_ROW_LENGTH;
×
1960
    TAOS_RETURN(code);
×
1961
  }
1962

1963
  pNew->numOfColumns = pNew->numOfColumns + ncols;
11✔
1964

1965
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
11!
1966

1967
  if (pNew->nextColId < 0 || pNew->nextColId >= 0x7fff - ncols) {
11!
1968
    code = TSDB_CODE_OUT_OF_RANGE;
×
1969
    TAOS_RETURN(code);
×
1970
  }
1971

1972
  for (int32_t i = 0; i < ncols; i++) {
18✔
1973
    if (withCompress) {
11!
1974
      SFieldWithOptions *pField = taosArrayGet(pReq->pFields, i);
×
1975
      if (mndFindSuperTableColumnIndex(pOld, pField->name) >= 0) {
×
1976
        code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
×
1977
        TAOS_RETURN(code);
×
1978
      }
1979

1980
      if (mndFindSuperTableTagIndex(pOld, pField->name) >= 0) {
×
1981
        code = TSDB_CODE_MND_TAG_ALREADY_EXIST;
×
1982
        TAOS_RETURN(code);
×
1983
      }
1984

1985
      SSchema *pSchema = &pNew->pColumns[pOld->numOfColumns + i];
×
1986
      pSchema->bytes = pField->bytes;
×
1987
      pSchema->type = pField->type;
×
1988
      memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
×
1989
      pSchema->colId = pNew->nextColId;
×
1990
      pNew->nextColId++;
×
1991

1992
      SColCmpr *pCmpr = &pNew->pCmpr[pOld->numOfColumns + i];
×
1993
      pCmpr->id = pSchema->colId;
×
1994
      pCmpr->alg = pField->compress;
×
1995
      mInfo("stb:%s, start to add column %s", pNew->name, pSchema->name);
×
1996
    } else {
1997
      SField *pField = taosArrayGet(pReq->pFields, i);
11✔
1998
      if (mndFindSuperTableColumnIndex(pOld, pField->name) >= 0) {
11✔
1999
        code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
3✔
2000
        TAOS_RETURN(code);
3✔
2001
      }
2002

2003
      if (mndFindSuperTableTagIndex(pOld, pField->name) >= 0) {
8✔
2004
        code = TSDB_CODE_MND_TAG_ALREADY_EXIST;
1✔
2005
        TAOS_RETURN(code);
1✔
2006
      }
2007

2008
      SSchema *pSchema = &pNew->pColumns[pOld->numOfColumns + i];
7✔
2009
      pSchema->bytes = pField->bytes;
7✔
2010
      pSchema->type = pField->type;
7✔
2011
      memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
7✔
2012
      pSchema->colId = pNew->nextColId;
7✔
2013
      pNew->nextColId++;
7✔
2014

2015
      SColCmpr *pCmpr = &pNew->pCmpr[pOld->numOfColumns + i];
7✔
2016
      pCmpr->id = pSchema->colId;
7✔
2017
      pCmpr->alg = createDefaultColCmprByType(pSchema->type);
7✔
2018
      mInfo("stb:%s, start to add column %s", pNew->name, pSchema->name);
7!
2019
    }
2020
  }
2021
  // 1. old schema already has extschemas
2022
  // 2. new schema has extschemas
2023
  if (pReq->pTypeMods || pOld->pExtSchemas) {
7!
2024
    if (!pNew->pExtSchemas) {
5✔
2025
      // all ext schemas reset to zero
2026
      pNew->pExtSchemas = taosMemoryCalloc(pNew->numOfColumns, sizeof(SExtSchema));
2!
2027
      if (!pNew->pExtSchemas) TAOS_RETURN(terrno);
2!
2028
    }
2029
    if (pOld->pExtSchemas) {
5✔
2030
      memcpy(pNew->pExtSchemas, pOld->pExtSchemas, pOld->numOfColumns * sizeof(SExtSchema));
3✔
2031
    }
2032
    if (taosArrayGetSize(pReq->pTypeMods) > 0) {
5!
2033
      // copy added column ext schema
2034
      for (int32_t i = 0; i < ncols; ++i) {
10✔
2035
        pNew->pColumns[pOld->numOfColumns + i].flags |= COL_HAS_TYPE_MOD;
5✔
2036
        pNew->pExtSchemas[pOld->numOfColumns + i].typeMod = *(STypeMod *)taosArrayGet(pReq->pTypeMods, i);
5✔
2037
      }
2038
    }
2039
  }
2040

2041
  pNew->colVer++;
7✔
2042
  TAOS_RETURN(code);
7✔
2043
}
2044

2045
static int32_t mndDropSuperTableColumn(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const char *colName) {
7✔
2046
  int32_t code = 0;
7✔
2047
  int32_t col = mndFindSuperTableColumnIndex(pOld, colName);
7✔
2048
  if (col < 0) {
7✔
2049
    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;
1✔
2050
    TAOS_RETURN(code);
1✔
2051
  }
2052

2053
  if (col == 0) {
6✔
2054
    code = TSDB_CODE_MND_INVALID_STB_ALTER_OPTION;
1✔
2055
    TAOS_RETURN(code);
1✔
2056
  }
2057

2058
  if (pOld->numOfColumns == 2) {
5✔
2059
    code = TSDB_CODE_PAR_INVALID_DROP_COL;
1✔
2060
    TAOS_RETURN(code);
1✔
2061
  }
2062

2063
  col_id_t colId = pOld->pColumns[col].colId;
4✔
2064
  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));
4!
2065

2066
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
4!
2067

2068
  int32_t sz = pNew->numOfColumns - col - 1;
4✔
2069
  memmove(pNew->pColumns + col, pNew->pColumns + col + 1, sizeof(SSchema) * sz);
4✔
2070
  memmove(pNew->pCmpr + col, pNew->pCmpr + col + 1, sizeof(SColCmpr) * sz);
4✔
2071
  if (pOld->pExtSchemas) {
4✔
2072
    memmove(pNew->pExtSchemas + col, pNew->pExtSchemas + col + 1, sizeof(SExtSchema) * sz);
3✔
2073
  }
2074
  pNew->numOfColumns--;
4✔
2075

2076
  pNew->colVer++;
4✔
2077
  mInfo("stb:%s, start to drop col %s", pNew->name, colName);
4!
2078
  TAOS_RETURN(code);
4✔
2079
}
2080

2081
static int32_t mndAlterStbColumnBytes(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const SField *pField) {
7✔
2082
  int32_t code = 0;
7✔
2083
  int32_t col = mndFindSuperTableColumnIndex(pOld, pField->name);
7✔
2084
  if (col < 0) {
7✔
2085
    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;
2✔
2086
    TAOS_RETURN(code);
2✔
2087
  }
2088

2089
  col_id_t colId = pOld->pColumns[col].colId;
5✔
2090

2091
  uint32_t nLen = 0;
5✔
2092
  for (int32_t i = 0; i < pOld->numOfColumns; ++i) {
17✔
2093
    nLen += (pOld->pColumns[i].colId == colId) ? pField->bytes : pOld->pColumns[i].bytes;
12✔
2094
  }
2095

2096
  if (nLen > TSDB_MAX_BYTES_PER_ROW) {
5✔
2097
    code = TSDB_CODE_MND_INVALID_ROW_BYTES;
1✔
2098
    TAOS_RETURN(code);
1✔
2099
  }
2100

2101
  TAOS_CHECK_RETURN(mndCheckColAndTagModifiable(pMnode, pOld->name, pOld->uid, colId));
4!
2102

2103
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
4!
2104

2105
  SSchema *pCol = pNew->pColumns + col;
4✔
2106
  if (!(pCol->type == TSDB_DATA_TYPE_BINARY || pCol->type == TSDB_DATA_TYPE_VARBINARY ||
4!
2107
        pCol->type == TSDB_DATA_TYPE_NCHAR || pCol->type == TSDB_DATA_TYPE_GEOMETRY)) {
1!
2108
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
1✔
2109
    TAOS_RETURN(code);
1✔
2110
  }
2111

2112
  if (pField->bytes <= pCol->bytes) {
3✔
2113
    code = TSDB_CODE_MND_INVALID_ROW_BYTES;
1✔
2114
    TAOS_RETURN(code);
1✔
2115
  }
2116

2117
  pCol->bytes = pField->bytes;
2✔
2118
  pNew->colVer++;
2✔
2119

2120
  mInfo("stb:%s, start to modify col len %s to %d", pNew->name, pField->name, pField->bytes);
2!
2121
  TAOS_RETURN(code);
2✔
2122
}
2123

2124
static int32_t mndSetAlterStbPrepareLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
26✔
2125
  int32_t  code = 0;
26✔
2126
  SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);
26✔
2127
  if (pRedoRaw == NULL) {
26!
2128
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2129
    if (terrno != 0) code = terrno;
×
2130
    TAOS_RETURN(code);
×
2131
  }
2132
  if ((code = mndTransAppendPrepareLog(pTrans, pRedoRaw)) != 0) {
26!
2133
    sdbFreeRaw(pRedoRaw);
×
2134
    TAOS_RETURN(code);
×
2135
  }
2136
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY));
26!
2137

2138
  TAOS_RETURN(code);
26✔
2139
}
2140

2141
static int32_t mndSetAlterStbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
26✔
2142
  int32_t  code = 0;
26✔
2143
  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);
26✔
2144
  if (pCommitRaw == NULL) {
26!
2145
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2146
    if (terrno != 0) code = terrno;
×
2147
    TAOS_RETURN(code);
×
2148
  }
2149
  if ((code = mndTransAppendCommitlog(pTrans, pCommitRaw)) != 0) {
26!
2150
    sdbFreeRaw(pCommitRaw);
×
2151
    TAOS_RETURN(code);
×
2152
  }
2153
  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));
26!
2154

2155
  TAOS_RETURN(code);
26✔
2156
}
2157

2158
static int32_t mndSetAlterStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb, void *alterOriData,
26✔
2159
                                         int32_t alterOriDataLen) {
2160
  int32_t code = 0;
26✔
2161
  SSdb   *pSdb = pMnode->pSdb;
26✔
2162
  SVgObj *pVgroup = NULL;
26✔
2163
  void   *pIter = NULL;
26✔
2164
  int32_t contLen;
2165

2166
  while (1) {
52✔
2167
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
78✔
2168
    if (pIter == NULL) break;
78✔
2169
    if (!mndVgroupInDb(pVgroup, pDb->uid)) {
52!
2170
      sdbRelease(pSdb, pVgroup);
×
2171
      continue;
×
2172
    }
2173

2174
    void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, alterOriData, alterOriDataLen);
52✔
2175
    if (pReq == NULL) {
52!
2176
      sdbCancelFetch(pSdb, pIter);
×
2177
      sdbRelease(pSdb, pVgroup);
×
2178
      code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2179
      if (terrno != 0) code = terrno;
×
2180
      TAOS_RETURN(code);
×
2181
    }
2182
    STransAction action = {0};
52✔
2183
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
52✔
2184
    action.pCont = pReq;
52✔
2185
    action.contLen = contLen;
52✔
2186
    action.msgType = TDMT_VND_ALTER_STB;
52✔
2187
    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
52!
2188
      taosMemoryFree(pReq);
×
2189
      sdbCancelFetch(pSdb, pIter);
×
2190
      sdbRelease(pSdb, pVgroup);
×
2191
      TAOS_RETURN(code);
×
2192
    }
2193
    sdbRelease(pSdb, pVgroup);
52✔
2194
  }
2195

2196
  TAOS_RETURN(code);
26✔
2197
}
2198

2199
static int32_t mndSetAlterStbRedoActions2(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb,
×
2200
                                          void *alterOriData, int32_t alterOriDataLen) {
2201
  int32_t code = 0;
×
2202
  SSdb   *pSdb = pMnode->pSdb;
×
2203
  SVgObj *pVgroup = NULL;
×
2204
  void   *pIter = NULL;
×
2205
  int32_t contLen;
2206

2207
  while (1) {
×
2208
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
×
2209
    if (pIter == NULL) break;
×
2210
    if (!mndVgroupInDb(pVgroup, pDb->uid)) {
×
2211
      sdbRelease(pSdb, pVgroup);
×
2212
      continue;
×
2213
    }
2214

2215
    void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, alterOriData, alterOriDataLen);
×
2216
    if (pReq == NULL) {
×
2217
      sdbCancelFetch(pSdb, pIter);
×
2218
      sdbRelease(pSdb, pVgroup);
×
2219
      code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2220
      if (terrno != 0) code = terrno;
×
2221
      TAOS_RETURN(code);
×
2222
    }
2223
    STransAction action = {0};
×
2224
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
×
2225
    action.pCont = pReq;
×
2226
    action.contLen = contLen;
×
2227
    action.msgType = TDMT_VND_CREATE_INDEX;
×
2228
    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
×
2229
      taosMemoryFree(pReq);
×
2230
      sdbCancelFetch(pSdb, pIter);
×
2231
      sdbRelease(pSdb, pVgroup);
×
2232
      TAOS_RETURN(code);
×
2233
    }
2234
    sdbRelease(pSdb, pVgroup);
×
2235
  }
2236

2237
  TAOS_RETURN(code);
×
2238
}
2239

2240
static int32_t mndBuildStbSchemaImp(SDbObj *pDb, SStbObj *pStb, const char *tbName, STableMetaRsp *pRsp) {
246✔
2241
  int32_t code = 0;
246✔
2242
  taosRLockLatch(&pStb->lock);
246✔
2243

2244
  int32_t totalCols = pStb->numOfColumns + pStb->numOfTags;
246✔
2245
  pRsp->pSchemas = taosMemoryCalloc(totalCols, sizeof(SSchema));
246!
2246
  if (pRsp->pSchemas == NULL) {
246!
2247
    taosRUnLockLatch(&pStb->lock);
×
2248
    code = terrno;
×
2249
    TAOS_RETURN(code);
×
2250
  }
2251
  pRsp->pSchemaExt = taosMemoryCalloc(pStb->numOfColumns, sizeof(SSchemaExt));
246!
2252
  if (pRsp->pSchemaExt == NULL) {
246!
2253
    taosRUnLockLatch(&pStb->lock);
×
2254
    code = terrno;
×
2255
    TAOS_RETURN(code);
×
2256
  }
2257
  pRsp->numOfColRefs = 0;
246✔
2258
  pRsp->pColRefs = NULL;
246✔
2259
  tstrncpy(pRsp->dbFName, pStb->db, sizeof(pRsp->dbFName));
246✔
2260
  tstrncpy(pRsp->tbName, tbName, sizeof(pRsp->tbName));
246✔
2261
  tstrncpy(pRsp->stbName, tbName, sizeof(pRsp->stbName));
246✔
2262
  pRsp->dbId = pDb->uid;
246✔
2263
  pRsp->numOfTags = pStb->numOfTags;
246✔
2264
  pRsp->numOfColumns = pStb->numOfColumns;
246✔
2265
  pRsp->precision = pDb->cfg.precision;
246✔
2266
  pRsp->tableType = TSDB_SUPER_TABLE;
246✔
2267
  pRsp->sversion = pStb->colVer;
246✔
2268
  pRsp->tversion = pStb->tagVer;
246✔
2269
  pRsp->suid = pStb->uid;
246✔
2270
  pRsp->tuid = pStb->uid;
246✔
2271
  pRsp->virtualStb = pStb->virtualStb;
246✔
2272

2273
  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
9,744✔
2274
    SSchema *pSchema = &pRsp->pSchemas[i];
9,498✔
2275
    SSchema *pSrcSchema = &pStb->pColumns[i];
9,498✔
2276
    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
9,498✔
2277
    pSchema->type = pSrcSchema->type;
9,498✔
2278
    pSchema->flags = pSrcSchema->flags;
9,498✔
2279
    pSchema->colId = pSrcSchema->colId;
9,498✔
2280
    pSchema->bytes = pSrcSchema->bytes;
9,498✔
2281
  }
2282

2283
  for (int32_t i = 0; i < pStb->numOfTags; ++i) {
1,167✔
2284
    SSchema *pSchema = &pRsp->pSchemas[i + pStb->numOfColumns];
921✔
2285
    SSchema *pSrcSchema = &pStb->pTags[i];
921✔
2286
    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
921✔
2287
    pSchema->type = pSrcSchema->type;
921✔
2288
    pSchema->flags = pSrcSchema->flags;
921✔
2289
    pSchema->colId = pSrcSchema->colId;
921✔
2290
    pSchema->bytes = pSrcSchema->bytes;
921✔
2291
  }
2292
  for (int32_t i = 0; i < pStb->numOfColumns; i++) {
9,744✔
2293
    SColCmpr   *pCmpr = &pStb->pCmpr[i];
9,498✔
2294
    SSchemaExt *pSchEx = &pRsp->pSchemaExt[i];
9,498✔
2295
    pSchEx->colId = pCmpr->id;
9,498✔
2296
    pSchEx->compress = pCmpr->alg;
9,498✔
2297
    if (pStb->pExtSchemas) {
9,498✔
2298
      pSchEx->typeMod = pStb->pExtSchemas[i].typeMod;
89✔
2299
    }
2300
  }
2301

2302
  taosRUnLockLatch(&pStb->lock);
246✔
2303
  TAOS_RETURN(code);
246✔
2304
}
2305

2306
static int32_t mndBuildStbCfgImp(SDbObj *pDb, SStbObj *pStb, const char *tbName, STableCfgRsp *pRsp) {
×
2307
  int32_t code = 0;
×
2308
  taosRLockLatch(&pStb->lock);
×
2309

2310
  int32_t totalCols = pStb->numOfColumns + pStb->numOfTags;
×
2311
  pRsp->pSchemas = taosMemoryCalloc(totalCols, sizeof(SSchema));
×
2312
  if (pRsp->pSchemas == NULL) {
×
2313
    taosRUnLockLatch(&pStb->lock);
×
2314
    code = terrno;
×
2315
    TAOS_RETURN(code);
×
2316
  }
2317

2318
  tstrncpy(pRsp->dbFName, pStb->db, sizeof(pRsp->dbFName));
×
2319
  tstrncpy(pRsp->tbName, tbName, sizeof(pRsp->tbName));
×
2320
  tstrncpy(pRsp->stbName, tbName, sizeof(pRsp->stbName));
×
2321
  pRsp->numOfTags = pStb->numOfTags;
×
2322
  pRsp->numOfColumns = pStb->numOfColumns;
×
2323
  pRsp->tableType = TSDB_SUPER_TABLE;
×
2324
  pRsp->delay1 = pStb->maxdelay[0];
×
2325
  pRsp->delay2 = pStb->maxdelay[1];
×
2326
  pRsp->watermark1 = pStb->watermark[0];
×
2327
  pRsp->watermark2 = pStb->watermark[1];
×
2328
  pRsp->ttl = pStb->ttl;
×
2329
  pRsp->keep = pStb->keep;
×
2330
  pRsp->commentLen = pStb->commentLen;
×
2331
  if (pStb->commentLen > 0) {
×
2332
    pRsp->pComment = taosStrdup(pStb->comment);
×
2333
  }
2334

2335
  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
×
2336
    SSchema *pSchema = &pRsp->pSchemas[i];
×
2337
    SSchema *pSrcSchema = &pStb->pColumns[i];
×
2338
    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
×
2339
    pSchema->type = pSrcSchema->type;
×
2340
    pSchema->flags = pSrcSchema->flags;
×
2341
    pSchema->colId = pSrcSchema->colId;
×
2342
    pSchema->bytes = pSrcSchema->bytes;
×
2343
  }
2344

2345
  for (int32_t i = 0; i < pStb->numOfTags; ++i) {
×
2346
    SSchema *pSchema = &pRsp->pSchemas[i + pStb->numOfColumns];
×
2347
    SSchema *pSrcSchema = &pStb->pTags[i];
×
2348
    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
×
2349
    pSchema->type = pSrcSchema->type;
×
2350
    pSchema->flags = pSrcSchema->flags;
×
2351
    pSchema->colId = pSrcSchema->colId;
×
2352
    pSchema->bytes = pSrcSchema->bytes;
×
2353
  }
2354

2355
  if (pStb->numOfFuncs > 0) {
×
2356
    pRsp->pFuncs = taosArrayDup(pStb->pFuncs, NULL);
×
2357
  }
2358

2359
  pRsp->pSchemaExt = taosMemoryCalloc(pStb->numOfColumns, sizeof(SSchemaExt));
×
2360
  for (int32_t i = 0; i < pStb->numOfColumns; i++) {
×
2361
    SColCmpr *pCmpr = &pStb->pCmpr[i];
×
2362

2363
    SSchemaExt *pSchExt = &pRsp->pSchemaExt[i];
×
2364
    pSchExt->colId = pCmpr->id;
×
2365
    pSchExt->compress = pCmpr->alg;
×
2366
    if (pStb->pExtSchemas) {
×
2367
      pSchExt->typeMod = pStb->pExtSchemas[i].typeMod;
×
2368
    }
2369
  }
2370
  pRsp->virtualStb = pStb->virtualStb;
×
2371
  pRsp->pColRefs = NULL;
×
2372

2373
  taosRUnLockLatch(&pStb->lock);
×
2374
  TAOS_RETURN(code);
×
2375
}
2376

2377
static int32_t mndValidateStbVersion(SMnode *pMnode, SSTableVersion *pStbVer, bool *schema, bool *sma) {
253✔
2378
  int32_t code = 0;
253✔
2379
  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};
253✔
2380
  snprintf(tbFName, sizeof(tbFName), "%s.%s", pStbVer->dbFName, pStbVer->stbName);
253✔
2381

2382
  SDbObj *pDb = mndAcquireDb(pMnode, pStbVer->dbFName);
253✔
2383
  if (pDb == NULL) {
253✔
2384
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
4✔
2385
    TAOS_RETURN(code);
4✔
2386
  }
2387

2388
  if (pDb->uid != pStbVer->dbId) {
249!
2389
    mndReleaseDb(pMnode, pDb);
×
2390
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
2391
    TAOS_RETURN(code);
×
2392
  }
2393

2394
  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);
249✔
2395
  if (pStb == NULL) {
249!
2396
    mndReleaseDb(pMnode, pDb);
×
2397
    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
×
2398
    TAOS_RETURN(code);
×
2399
  }
2400

2401
  taosRLockLatch(&pStb->lock);
249✔
2402

2403
  if (pStbVer->sversion != pStb->colVer || pStbVer->tversion != pStb->tagVer) {
249!
2404
    *schema = true;
×
2405
  } else {
2406
    *schema = false;
249✔
2407
  }
2408

2409
  if (pStbVer->smaVer && pStbVer->smaVer != pStb->smaVer) {
249!
2410
    *sma = true;
×
2411
  } else {
2412
    *sma = false;
249✔
2413
  }
2414

2415
  taosRUnLockLatch(&pStb->lock);
249✔
2416

2417
  mndReleaseDb(pMnode, pDb);
249✔
2418
  mndReleaseStb(pMnode, pStb);
249✔
2419
  return TSDB_CODE_SUCCESS;
249✔
2420
}
2421

2422
static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp) {
116✔
2423
  int32_t code = 0;
116✔
2424
  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};
116✔
2425
  snprintf(tbFName, sizeof(tbFName), "%s.%s", dbFName, tbName);
116✔
2426

2427
  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);
116✔
2428
  if (pDb == NULL) {
116!
2429
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
2430
    TAOS_RETURN(code);
×
2431
  }
2432

2433
  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);
116✔
2434
  if (pStb == NULL) {
116!
2435
    mndReleaseDb(pMnode, pDb);
×
2436
    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
×
2437
    TAOS_RETURN(code);
×
2438
  }
2439

2440
  code = mndBuildStbSchemaImp(pDb, pStb, tbName, pRsp);
116✔
2441
  mndReleaseDb(pMnode, pDb);
116✔
2442
  mndReleaseStb(pMnode, pStb);
116✔
2443
  TAOS_RETURN(code);
116✔
2444
}
2445

2446
static int32_t mndBuildStbCfg(SMnode *pMnode, const char *dbFName, const char *tbName, STableCfgRsp *pRsp) {
×
2447
  int32_t code = 0;
×
2448
  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};
×
2449
  snprintf(tbFName, sizeof(tbFName), "%s.%s", dbFName, tbName);
×
2450

2451
  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);
×
2452
  if (pDb == NULL) {
×
2453
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
2454
    TAOS_RETURN(code);
×
2455
  }
2456

2457
  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);
×
2458
  if (pStb == NULL) {
×
2459
    mndReleaseDb(pMnode, pDb);
×
2460
    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
×
2461
    TAOS_RETURN(code);
×
2462
  }
2463

2464
  code = mndBuildStbCfgImp(pDb, pStb, tbName, pRsp);
×
2465

2466
  mndReleaseDb(pMnode, pDb);
×
2467
  mndReleaseStb(pMnode, pStb);
×
2468
  TAOS_RETURN(code);
×
2469
}
2470

2471
static int32_t mndBuildSMAlterStbRsp(SDbObj *pDb, SStbObj *pObj, void **pCont, int32_t *pLen) {
25✔
2472
  int32_t       code = 0;
25✔
2473
  SEncoder      ec = {0};
25✔
2474
  uint32_t      contLen = 0;
25✔
2475
  SMAlterStbRsp alterRsp = {0};
25✔
2476
  SName         name = {0};
25✔
2477
  TAOS_CHECK_RETURN(tNameFromString(&name, pObj->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
25!
2478

2479
  alterRsp.pMeta = taosMemoryCalloc(1, sizeof(STableMetaRsp));
25!
2480
  if (NULL == alterRsp.pMeta) {
25!
2481
    code = terrno;
×
2482
    TAOS_RETURN(code);
×
2483
  }
2484

2485
  code = mndBuildStbSchemaImp(pDb, pObj, name.tname, alterRsp.pMeta);
25✔
2486
  if (code) {
25!
2487
    tFreeSMAlterStbRsp(&alterRsp);
×
2488
    return code;
×
2489
  }
2490

2491
  tEncodeSize(tEncodeSMAlterStbRsp, &alterRsp, contLen, code);
25!
2492
  if (code) {
25!
2493
    tFreeSMAlterStbRsp(&alterRsp);
×
2494
    return code;
×
2495
  }
2496

2497
  void *cont = taosMemoryMalloc(contLen);
25!
2498
  if (NULL == cont) {
25!
2499
    code = terrno;
×
2500
    tFreeSMAlterStbRsp(&alterRsp);
×
2501
    TAOS_RETURN(code);
×
2502
  }
2503
  tEncoderInit(&ec, cont, contLen);
25✔
2504
  code = tEncodeSMAlterStbRsp(&ec, &alterRsp);
25✔
2505
  tEncoderClear(&ec);
25✔
2506

2507
  tFreeSMAlterStbRsp(&alterRsp);
25✔
2508

2509
  if (code < 0) TAOS_RETURN(code);
25!
2510

2511
  *pCont = cont;
25✔
2512
  *pLen = contLen;
25✔
2513

2514
  TAOS_RETURN(code);
25✔
2515
}
2516

2517
int32_t mndBuildSMCreateStbRsp(SMnode *pMnode, char *dbFName, char *stbFName, void **pCont, int32_t *pLen) {
105✔
2518
  int32_t code = -1;
105✔
2519
  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);
105✔
2520
  if (NULL == pDb) {
105!
2521
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2522
    if (terrno != 0) code = terrno;
×
2523
    TAOS_RETURN(code);
×
2524
  }
2525

2526
  SStbObj *pObj = mndAcquireStb(pMnode, stbFName);
105✔
2527
  if (NULL == pObj) {
105!
2528
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2529
    if (terrno != 0) code = terrno;
×
2530
    goto _OVER;
×
2531
  }
2532

2533
  SEncoder       ec = {0};
105✔
2534
  uint32_t       contLen = 0;
105✔
2535
  SMCreateStbRsp stbRsp = {0};
105✔
2536
  SName          name = {0};
105✔
2537
  TAOS_CHECK_GOTO(tNameFromString(&name, pObj->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE), NULL, _OVER);
105!
2538

2539
  stbRsp.pMeta = taosMemoryCalloc(1, sizeof(STableMetaRsp));
105!
2540
  if (NULL == stbRsp.pMeta) {
105!
2541
    code = terrno;
×
2542
    goto _OVER;
×
2543
  }
2544

2545
  code = mndBuildStbSchemaImp(pDb, pObj, name.tname, stbRsp.pMeta);
105✔
2546
  if (code) {
105!
2547
    tFreeSMCreateStbRsp(&stbRsp);
×
2548
    goto _OVER;
×
2549
  }
2550

2551
  tEncodeSize(tEncodeSMCreateStbRsp, &stbRsp, contLen, code);
105!
2552
  if (code) {
105!
2553
    tFreeSMCreateStbRsp(&stbRsp);
×
2554
    goto _OVER;
×
2555
  }
2556

2557
  void *cont = taosMemoryMalloc(contLen);
105!
2558
  if (NULL == cont) {
105!
2559
    code = terrno;
×
2560
    tFreeSMCreateStbRsp(&stbRsp);
×
2561
    goto _OVER;
×
2562
  }
2563
  tEncoderInit(&ec, cont, contLen);
105✔
2564
  TAOS_CHECK_GOTO(tEncodeSMCreateStbRsp(&ec, &stbRsp), NULL, _OVER);
105!
2565
  tEncoderClear(&ec);
105✔
2566

2567
  tFreeSMCreateStbRsp(&stbRsp);
105✔
2568

2569
  *pCont = cont;
105✔
2570
  *pLen = contLen;
105✔
2571

2572
  code = 0;
105✔
2573

2574
_OVER:
105✔
2575
  if (pObj) {
105!
2576
    mndReleaseStb(pMnode, pObj);
105✔
2577
  }
2578

2579
  if (pDb) {
105!
2580
    mndReleaseDb(pMnode, pDb);
105✔
2581
  }
2582

2583
  TAOS_RETURN(code);
105✔
2584
}
2585

2586
static int32_t mndAlterStbImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp,
21✔
2587
                              void *alterOriData, int32_t alterOriDataLen) {
2588
  int32_t code = -1;
21✔
2589
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "alter-stb");
21✔
2590
  if (pTrans == NULL) {
21!
2591
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2592
    if (terrno != 0) code = terrno;
×
2593
    goto _OVER;
×
2594
  }
2595

2596
  mInfo("trans:%d, used to alter stb:%s", pTrans->id, pStb->name);
21!
2597
  mndTransSetDbName(pTrans, pDb->name, pStb->name);
21✔
2598
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
21!
2599

2600
  if (needRsp) {
21✔
2601
    void   *pCont = NULL;
20✔
2602
    int32_t contLen = 0;
20✔
2603
    TAOS_CHECK_GOTO(mndBuildSMAlterStbRsp(pDb, pStb, &pCont, &contLen), NULL, _OVER);
20!
2604
    mndTransSetRpcRsp(pTrans, pCont, contLen);
20✔
2605
  }
2606

2607
  TAOS_CHECK_GOTO(mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
21!
2608
  TAOS_CHECK_GOTO(mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
21!
2609
  TAOS_CHECK_GOTO(mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb, alterOriData, alterOriDataLen), NULL, _OVER);
21!
2610
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
21!
2611

2612
  code = 0;
21✔
2613

2614
_OVER:
21✔
2615
  mndTransDrop(pTrans);
21✔
2616
  TAOS_RETURN(code);
21✔
2617
}
2618

2619
static int32_t mndAlterStbAndUpdateTagIdxImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp,
5✔
2620
                                             void *alterOriData, int32_t alterOriDataLen, const SMAlterStbReq *pAlter) {
2621
  int32_t code = -1;
5✔
2622
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "alter-stb");
5✔
2623
  if (pTrans == NULL) {
5!
2624
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2625
    if (terrno != 0) code = terrno;
×
2626
    goto _OVER;
×
2627
  }
2628

2629
  mInfo("trans:%d, used to alter stb:%s", pTrans->id, pStb->name);
5!
2630
  mndTransSetDbName(pTrans, pDb->name, pStb->name);
5✔
2631

2632
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
5!
2633

2634
  if (needRsp) {
5!
2635
    void   *pCont = NULL;
5✔
2636
    int32_t contLen = 0;
5✔
2637
    TAOS_CHECK_GOTO(mndBuildSMAlterStbRsp(pDb, pStb, &pCont, &contLen), NULL, _OVER);
5!
2638
    mndTransSetRpcRsp(pTrans, pCont, contLen);
5✔
2639
  }
2640

2641
  if (pAlter->alterType == TSDB_ALTER_TABLE_DROP_TAG) {
5✔
2642
    SIdxObj idxObj = {0};
3✔
2643
    SField *pField0 = taosArrayGet(pAlter->pFields, 0);
3✔
2644
    bool    exist = false;
3✔
2645
    if (mndGetIdxsByTagName(pMnode, pStb, pField0->name, &idxObj) == 0) {
3!
2646
      exist = true;
×
2647
    }
2648
    TAOS_CHECK_GOTO(mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
3!
2649
    TAOS_CHECK_GOTO(mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
3!
2650

2651
    if (exist == true) {
3!
2652
      TAOS_CHECK_GOTO(mndSetDropIdxPrepareLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
×
2653
      TAOS_CHECK_GOTO(mndSetDropIdxCommitLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
×
2654
    }
2655

2656
    TAOS_CHECK_GOTO(mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb, alterOriData, alterOriDataLen), NULL, _OVER);
3!
2657
    TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
3!
2658

2659
  } else if (pAlter->alterType == TSDB_ALTER_TABLE_UPDATE_TAG_NAME) {
2!
2660
    SIdxObj     idxObj = {0};
2✔
2661
    SField     *pField0 = taosArrayGet(pAlter->pFields, 0);
2✔
2662
    SField     *pField1 = taosArrayGet(pAlter->pFields, 1);
2✔
2663
    const char *oTagName = pField0->name;
2✔
2664
    const char *nTagName = pField1->name;
2✔
2665
    bool        exist = false;
2✔
2666

2667
    if (mndGetIdxsByTagName(pMnode, pStb, pField0->name, &idxObj) == 0) {
2✔
2668
      exist = true;
1✔
2669
    }
2670

2671
    TAOS_CHECK_GOTO(mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
2!
2672
    TAOS_CHECK_GOTO(mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
2!
2673

2674
    if (exist == true) {
2✔
2675
      memcpy(idxObj.colName, nTagName, strlen(nTagName));
1✔
2676
      idxObj.colName[strlen(nTagName)] = 0;
1✔
2677
      TAOS_CHECK_GOTO(mndSetAlterIdxPrepareLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
1!
2678
      TAOS_CHECK_GOTO(mndSetAlterIdxCommitLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
1!
2679
    }
2680

2681
    TAOS_CHECK_GOTO(mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb, alterOriData, alterOriDataLen), NULL, _OVER);
2!
2682
    TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
2!
2683
  }
2684
  code = 0;
5✔
2685

2686
_OVER:
5✔
2687
  mndTransDrop(pTrans);
5✔
2688
  TAOS_RETURN(code);
5✔
2689
}
2690

2691
static int32_t mndAlterStb(SMnode *pMnode, SRpcMsg *pReq, const SMAlterStbReq *pAlter, SDbObj *pDb, SStbObj *pOld) {
49✔
2692
  bool    needRsp = true;
49✔
2693
  int32_t code = -1;
49✔
2694
  SField *pField0 = NULL;
49✔
2695

2696
  SStbObj stbObj = {0};
49✔
2697
  taosRLockLatch(&pOld->lock);
49✔
2698
  memcpy(&stbObj, pOld, sizeof(SStbObj));
49✔
2699
  taosRUnLockLatch(&pOld->lock);
49✔
2700
  stbObj.pColumns = NULL;
49✔
2701
  stbObj.pTags = NULL;
49✔
2702
  stbObj.pFuncs = NULL;
49✔
2703
  stbObj.pCmpr = NULL;
49✔
2704
  stbObj.pExtSchemas = NULL;
49✔
2705
  stbObj.updateTime = taosGetTimestampMs();
49✔
2706
  stbObj.lock = 0;
49✔
2707
  bool updateTagIndex = false;
49✔
2708
  switch (pAlter->alterType) {
49!
2709
    case TSDB_ALTER_TABLE_ADD_TAG:
7✔
2710
      code = mndAddSuperTableTag(pOld, &stbObj, pAlter->pFields, pAlter->numOfFields);
7✔
2711
      break;
7✔
2712
    case TSDB_ALTER_TABLE_DROP_TAG:
4✔
2713
      pField0 = taosArrayGet(pAlter->pFields, 0);
4✔
2714
      code = mndDropSuperTableTag(pMnode, pOld, &stbObj, pField0->name);
4✔
2715
      updateTagIndex = true;
4✔
2716
      break;
4✔
2717
    case TSDB_ALTER_TABLE_UPDATE_TAG_NAME:
7✔
2718
      code = mndAlterStbTagName(pMnode, pOld, &stbObj, pAlter->pFields);
7✔
2719
      updateTagIndex = true;
7✔
2720
      break;
7✔
2721
    case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES:
5✔
2722
      pField0 = taosArrayGet(pAlter->pFields, 0);
5✔
2723
      code = mndAlterStbTagBytes(pMnode, pOld, &stbObj, pField0);
5✔
2724
      break;
5✔
2725
    case TSDB_ALTER_TABLE_ADD_COLUMN:
11✔
2726
      code = mndAddSuperTableColumn(pOld, &stbObj, pAlter, pAlter->numOfFields, 0);
11✔
2727
      break;
11✔
2728
    case TSDB_ALTER_TABLE_DROP_COLUMN:
7✔
2729
      pField0 = taosArrayGet(pAlter->pFields, 0);
7✔
2730
      code = mndDropSuperTableColumn(pMnode, pOld, &stbObj, pField0->name);
7✔
2731
      break;
7✔
2732
    case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES:
7✔
2733
      pField0 = taosArrayGet(pAlter->pFields, 0);
7✔
2734
      code = mndAlterStbColumnBytes(pMnode, pOld, &stbObj, pField0);
7✔
2735
      break;
7✔
2736
    case TSDB_ALTER_TABLE_UPDATE_OPTIONS:
1✔
2737
      needRsp = false;
1✔
2738
      code = mndUpdateTableOptions(pOld, &stbObj, pAlter->comment, pAlter->commentLen, pAlter->ttl, pAlter->keep);
1✔
2739
      break;
1✔
2740
    case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS:
×
2741
      code = mndUpdateSuperTableColumnCompress(pMnode, pOld, &stbObj, pAlter->pFields, pAlter->numOfFields);
×
2742
      break;
×
2743
    case TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COMPRESS_OPTION:
×
2744
      code = mndAddSuperTableColumn(pOld, &stbObj, pAlter, pAlter->numOfFields, 1);
×
2745
      break;
×
2746
    default:
×
2747
      needRsp = false;
×
2748
      terrno = TSDB_CODE_OPS_NOT_SUPPORT;
×
2749
      break;
×
2750
  }
2751

2752
  if (code != 0) goto _OVER;
49✔
2753
  if (updateTagIndex == false) {
26✔
2754
    code = mndAlterStbImp(pMnode, pReq, pDb, &stbObj, needRsp, pReq->pCont, pReq->contLen);
21✔
2755
  } else {
2756
    code = mndAlterStbAndUpdateTagIdxImp(pMnode, pReq, pDb, &stbObj, needRsp, pReq->pCont, pReq->contLen, pAlter);
5✔
2757
  }
2758

2759
_OVER:
49✔
2760
  taosMemoryFreeClear(stbObj.pTags);
49!
2761
  taosMemoryFreeClear(stbObj.pColumns);
49!
2762
  taosMemoryFreeClear(stbObj.pCmpr);
49!
2763
  if (pAlter->commentLen > 0) {
49✔
2764
    taosMemoryFreeClear(stbObj.comment);
1!
2765
  }
2766
  taosMemoryFreeClear(stbObj.pExtSchemas);
49!
2767
  TAOS_RETURN(code);
49✔
2768
}
2769

2770
static int32_t mndProcessAlterStbReq(SRpcMsg *pReq) {
53✔
2771
  SMnode       *pMnode = pReq->info.node;
53✔
2772
  int32_t       code = -1;
53✔
2773
  SDbObj       *pDb = NULL;
53✔
2774
  SStbObj      *pStb = NULL;
53✔
2775
  SMAlterStbReq alterReq = {0};
53✔
2776

2777
  if (tDeserializeSMAlterStbReq(pReq->pCont, pReq->contLen, &alterReq) != 0) {
53!
2778
    code = TSDB_CODE_INVALID_MSG;
×
2779
    goto _OVER;
×
2780
  }
2781

2782
  mInfo("stb:%s, start to alter", alterReq.name);
53!
2783
  if (mndCheckAlterStbReq(&alterReq) != 0) goto _OVER;
53!
2784

2785
  pDb = mndAcquireDbByStb(pMnode, alterReq.name);
53✔
2786
  if (pDb == NULL) {
53✔
2787
    code = TSDB_CODE_MND_DB_NOT_EXIST;
2✔
2788
    goto _OVER;
2✔
2789
  }
2790

2791
  pStb = mndAcquireStb(pMnode, alterReq.name);
51✔
2792
  if (pStb == NULL) {
51✔
2793
    code = TSDB_CODE_MND_STB_NOT_EXIST;
2✔
2794
    goto _OVER;
2✔
2795
  }
2796

2797
  if ((code = mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pDb)) != 0) {
49!
2798
    goto _OVER;
×
2799
  }
2800

2801
  code = mndAlterStb(pMnode, pReq, &alterReq, pDb, pStb);
49✔
2802
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
49✔
2803

2804
  SName   name = {0};
49✔
2805
  int32_t ret = 0;
49✔
2806
  if ((ret = tNameFromString(&name, alterReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0)
49!
2807
    mError("stb:%s, failed to tNameFromString since %s", alterReq.name, tstrerror(ret));
×
2808

2809
  auditRecord(pReq, pMnode->clusterId, "alterStb", name.dbname, name.tname, alterReq.sql, alterReq.sqlLen);
49✔
2810

2811
_OVER:
53✔
2812
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
53!
2813
    mError("stb:%s, failed to alter since %s", alterReq.name, tstrerror(code));
27!
2814
  }
2815

2816
  mndReleaseStb(pMnode, pStb);
53✔
2817
  mndReleaseDb(pMnode, pDb);
53✔
2818
  tFreeSMAltertbReq(&alterReq);
53✔
2819

2820
  TAOS_RETURN(code);
53✔
2821
}
2822

2823
static int32_t mndSetDropStbPrepareLogs(SMnode *pMnode, STrans *pTrans, SStbObj *pStb) {
5✔
2824
  int32_t  code = 0;
5✔
2825
  SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);
5✔
2826
  if (pRedoRaw == NULL) {
5!
2827
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2828
    if (terrno != 0) code = terrno;
×
2829
    TAOS_RETURN(code);
×
2830
  }
2831
  if ((code = mndTransAppendPrepareLog(pTrans, pRedoRaw)) != 0) {
5!
2832
    sdbFreeRaw(pRedoRaw);
×
2833
    TAOS_RETURN(code);
×
2834
  }
2835
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING));
5!
2836

2837
  TAOS_RETURN(code);
5✔
2838
}
2839

2840
static int32_t mndSetDropStbCommitLogs(SMnode *pMnode, STrans *pTrans, SStbObj *pStb) {
5✔
2841
  int32_t  code = 0;
5✔
2842
  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);
5✔
2843
  if (pCommitRaw == NULL) {
5!
2844
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2845
    if (terrno != 0) code = terrno;
×
2846
    TAOS_RETURN(code);
×
2847
  }
2848
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
5!
2849
    sdbFreeRaw(pCommitRaw);
×
2850
    TAOS_RETURN(code);
×
2851
  }
2852
  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));
5!
2853

2854
  TAOS_RETURN(code);
5✔
2855
}
2856

2857
static int32_t mndSetDropStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
5✔
2858
  int32_t code = 0;
5✔
2859
  SSdb   *pSdb = pMnode->pSdb;
5✔
2860
  SVgObj *pVgroup = NULL;
5✔
2861
  void   *pIter = NULL;
5✔
2862

2863
  while (1) {
14✔
2864
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
19✔
2865
    if (pIter == NULL) break;
19✔
2866
    if (!mndVgroupInDb(pVgroup, pDb->uid)) {
14✔
2867
      sdbRelease(pSdb, pVgroup);
4✔
2868
      continue;
4✔
2869
    }
2870

2871
    int32_t contLen = 0;
10✔
2872
    void   *pReq = mndBuildVDropStbReq(pMnode, pVgroup, pStb, &contLen);
10✔
2873
    if (pReq == NULL) {
10!
2874
      sdbCancelFetch(pSdb, pIter);
×
2875
      sdbRelease(pSdb, pVgroup);
×
2876
      code = TSDB_CODE_OUT_OF_MEMORY;
×
2877
      TAOS_RETURN(code);
×
2878
    }
2879

2880
    STransAction action = {0};
10✔
2881
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
10✔
2882
    action.pCont = pReq;
10✔
2883
    action.contLen = contLen;
10✔
2884
    action.msgType = TDMT_VND_DROP_STB;
10✔
2885
    action.acceptableCode = TSDB_CODE_TDB_STB_NOT_EXIST;
10✔
2886
    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
10!
2887
      taosMemoryFree(pReq);
×
2888
      sdbCancelFetch(pSdb, pIter);
×
2889
      sdbRelease(pSdb, pVgroup);
×
2890
      TAOS_RETURN(code);
×
2891
    }
2892
    sdbRelease(pSdb, pVgroup);
10✔
2893
  }
2894

2895
  TAOS_RETURN(code);
5✔
2896
}
2897

2898
static int32_t mndDropStb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb) {
5✔
2899
  int32_t code = -1;
5✔
2900
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "drop-stb");
5✔
2901
  if (pTrans == NULL) {
5!
2902
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2903
    if (terrno != 0) code = terrno;
×
2904
    goto _OVER;
×
2905
  }
2906

2907
  mInfo("trans:%d, used to drop stb:%s", pTrans->id, pStb->name);
5!
2908
  mndTransSetDbName(pTrans, pDb->name, pStb->name);
5✔
2909
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
5!
2910

2911
  TAOS_CHECK_GOTO(mndSetDropStbPrepareLogs(pMnode, pTrans, pStb), NULL, _OVER);
5!
2912
  TAOS_CHECK_GOTO(mndSetDropStbCommitLogs(pMnode, pTrans, pStb), NULL, _OVER);
5!
2913
  TAOS_CHECK_GOTO(mndSetDropStbRedoActions(pMnode, pTrans, pDb, pStb), NULL, _OVER);
5!
2914
  TAOS_CHECK_GOTO(mndDropIdxsByStb(pMnode, pTrans, pDb, pStb), NULL, _OVER);
5!
2915
  TAOS_CHECK_GOTO(mndDropSmasByStb(pMnode, pTrans, pDb, pStb), NULL, _OVER);
5!
2916
  TAOS_CHECK_GOTO(mndUserRemoveStb(pMnode, pTrans, pStb->name), NULL, _OVER);
5!
2917
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
5!
2918
  code = 0;
5✔
2919

2920
_OVER:
5✔
2921
  mndTransDrop(pTrans);
5✔
2922
  TAOS_RETURN(code);
5✔
2923
}
2924

2925
static int32_t mndCheckDropStbForTopic(SMnode *pMnode, const char *stbFullName, int64_t suid) {
5✔
2926
  int32_t code = 0;
5✔
2927
  SSdb   *pSdb = pMnode->pSdb;
5✔
2928
  void   *pIter = NULL;
5✔
2929
  while (1) {
×
2930
    SMqTopicObj *pTopic = NULL;
5✔
2931
    pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);
5✔
2932
    if (pIter == NULL) break;
5!
2933

2934
    if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {
×
2935
      if (pTopic->stbUid == suid) {
×
2936
        sdbRelease(pSdb, pTopic);
×
2937
        sdbCancelFetch(pSdb, pIter);
×
2938
        TAOS_RETURN(-1);
×
2939
      }
2940
    }
2941

2942
    if (pTopic->ast == NULL) {
×
2943
      sdbRelease(pSdb, pTopic);
×
2944
      continue;
×
2945
    }
2946

2947
    SNode *pAst = NULL;
×
2948
    if (nodesStringToNode(pTopic->ast, &pAst) != 0) {
×
2949
      code = TSDB_CODE_MND_INVALID_TOPIC_OPTION;
×
2950
      mError("topic:%s, create ast error", pTopic->name);
×
2951
      sdbRelease(pSdb, pTopic);
×
2952
      sdbCancelFetch(pSdb, pIter);
×
2953
      TAOS_RETURN(code);
×
2954
    }
2955

2956
    SNodeList *pNodeList = NULL;
×
2957
    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=
×
2958
        0) {
2959
      sdbRelease(pSdb, pTopic);
×
2960
      sdbCancelFetch(pSdb, pIter);
×
2961
      TAOS_RETURN(code);
×
2962
    }
2963
    SNode *pNode = NULL;
×
2964
    FOREACH(pNode, pNodeList) {
×
2965
      SColumnNode *pCol = (SColumnNode *)pNode;
×
2966

2967
      if (pCol->tableId == suid) {
×
2968
        sdbRelease(pSdb, pTopic);
×
2969
        nodesDestroyNode(pAst);
×
2970
        nodesDestroyList(pNodeList);
×
2971
        sdbCancelFetch(pSdb, pIter);
×
2972
        TAOS_RETURN(-1);
×
2973
      } else {
2974
        goto NEXT;
×
2975
      }
2976
    }
2977
  NEXT:
×
2978
    sdbRelease(pSdb, pTopic);
×
2979
    nodesDestroyNode(pAst);
×
2980
    nodesDestroyList(pNodeList);
×
2981
  }
2982
  TAOS_RETURN(code);
5✔
2983
}
2984

2985
static int32_t mndCheckDropStbForStream(SMnode *pMnode, const char *stbFullName, int64_t suid) {
5✔
2986
  int32_t code = 0;
5✔
2987
  SSdb   *pSdb = pMnode->pSdb;
5✔
2988
  void   *pIter = NULL;
5✔
2989
  while (1) {
×
2990
    SStreamObj *pStream = NULL;
5✔
2991
    pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);
5✔
2992
    if (pIter == NULL) break;
5!
2993

2994
    if (pStream->targetStbUid == suid) {
×
2995
      sdbCancelFetch(pSdb, pIter);
×
2996
      sdbRelease(pSdb, pStream);
×
2997
      TAOS_RETURN(-1);
×
2998
    }
2999

3000
    SNode *pAst = NULL;
×
3001
    if (nodesStringToNode(pStream->ast, &pAst) != 0) {
×
3002
      code = TSDB_CODE_MND_INVALID_STREAM_OPTION;
×
3003
      mError("stream:%s, create ast error", pStream->name);
×
3004
      sdbCancelFetch(pSdb, pIter);
×
3005
      sdbRelease(pSdb, pStream);
×
3006
      TAOS_RETURN(code);
×
3007
    }
3008

3009
    SNodeList *pNodeList = NULL;
×
3010
    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=
×
3011
        0) {
3012
      sdbCancelFetch(pSdb, pIter);
×
3013
      sdbRelease(pSdb, pStream);
×
3014
      TAOS_RETURN(code);
×
3015
    }
3016
    SNode *pNode = NULL;
×
3017
    FOREACH(pNode, pNodeList) {
×
3018
      SColumnNode *pCol = (SColumnNode *)pNode;
×
3019

3020
      if (pCol->tableId == suid) {
×
3021
        sdbCancelFetch(pSdb, pIter);
×
3022
        sdbRelease(pSdb, pStream);
×
3023
        nodesDestroyNode(pAst);
×
3024
        nodesDestroyList(pNodeList);
×
3025
        TAOS_RETURN(-1);
×
3026
      } else {
3027
        goto NEXT;
×
3028
      }
3029
    }
3030
  NEXT:
×
3031
    sdbRelease(pSdb, pStream);
×
3032
    nodesDestroyNode(pAst);
×
3033
    nodesDestroyList(pNodeList);
×
3034
  }
3035
  TAOS_RETURN(code);
5✔
3036
}
3037

3038
static int32_t mndProcessDropTtltbRsp(SRpcMsg *pRsp) { return 0; }
×
3039
static int32_t mndProcessTrimDbRsp(SRpcMsg *pRsp) { return 0; }
×
3040
static int32_t mndProcessS3MigrateDbRsp(SRpcMsg *pRsp) { return 0; }
×
3041

3042
static int32_t mndProcessDropStbReq(SRpcMsg *pReq) {
15✔
3043
  SMnode      *pMnode = pReq->info.node;
15✔
3044
  int32_t      code = -1;
15✔
3045
  SDbObj      *pDb = NULL;
15✔
3046
  SStbObj     *pStb = NULL;
15✔
3047
  SMDropStbReq dropReq = {0};
15✔
3048

3049
  TAOS_CHECK_GOTO(tDeserializeSMDropStbReq(pReq->pCont, pReq->contLen, &dropReq), NULL, _OVER);
15!
3050

3051
  mInfo("stb:%s, start to drop", dropReq.name);
15!
3052

3053
  pStb = mndAcquireStb(pMnode, dropReq.name);
15✔
3054
  if (pStb == NULL) {
15✔
3055
    if (dropReq.igNotExists) {
10✔
3056
      mInfo("stb:%s, not exist, ignore not exist is set", dropReq.name);
9!
3057
      code = 0;
9✔
3058
      goto _OVER;
9✔
3059
    } else {
3060
      code = TSDB_CODE_MND_STB_NOT_EXIST;
1✔
3061
      goto _OVER;
1✔
3062
    }
3063
  }
3064

3065
  if ((dropReq.source == TD_REQ_FROM_TAOX_OLD || dropReq.source == TD_REQ_FROM_TAOX) && pStb->uid != dropReq.suid) {
5!
3066
    code = 0;
×
3067
    goto _OVER;
×
3068
  }
3069

3070
  pDb = mndAcquireDbByStb(pMnode, dropReq.name);
5✔
3071
  if (pDb == NULL) {
5!
3072
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
3073
    goto _OVER;
×
3074
  }
3075

3076
  if ((code = mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pDb)) != 0) {
5!
3077
    goto _OVER;
×
3078
  }
3079

3080
  if (mndCheckDropStbForTopic(pMnode, dropReq.name, pStb->uid) < 0) {
5!
3081
    code = TSDB_CODE_MND_TOPIC_MUST_BE_DELETED;
×
3082
    goto _OVER;
×
3083
  }
3084

3085
  if (mndCheckDropStbForStream(pMnode, dropReq.name, pStb->uid) < 0) {
5!
3086
    code = TSDB_CODE_MND_STREAM_MUST_BE_DELETED;
×
3087
    goto _OVER;
×
3088
  }
3089

3090
  code = mndDropStb(pMnode, pReq, pDb, pStb);
5✔
3091
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
5!
3092

3093
  SName   name = {0};
5✔
3094
  int32_t ret = 0;
5✔
3095
  if ((ret = tNameFromString(&name, dropReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0)
5!
3096
    mError("stb:%s, failed to tNameFromString since %s", dropReq.name, tstrerror(ret));
×
3097

3098
  auditRecord(pReq, pMnode->clusterId, "dropStb", name.dbname, name.tname, dropReq.sql, dropReq.sqlLen);
5✔
3099

3100
_OVER:
15✔
3101
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
15✔
3102
    mError("stb:%s, failed to drop since %s", dropReq.name, tstrerror(code));
1!
3103
  }
3104

3105
  mndReleaseDb(pMnode, pDb);
15✔
3106
  mndReleaseStb(pMnode, pStb);
15✔
3107
  tFreeSMDropStbReq(&dropReq);
15✔
3108
  TAOS_RETURN(code);
15✔
3109
}
3110

3111
static int32_t mndProcessTableMetaReq(SRpcMsg *pReq) {
217✔
3112
  SMnode       *pMnode = pReq->info.node;
217✔
3113
  int32_t       code = -1;
217✔
3114
  STableInfoReq infoReq = {0};
217✔
3115
  STableMetaRsp metaRsp = {0};
217✔
3116
  SUserObj     *pUser = NULL;
217✔
3117

3118
  code = mndAcquireUser(pMnode, pReq->info.conn.user, &pUser);
217✔
3119
  if (pUser == NULL) return 0;
217!
3120
  bool sysinfo = pUser->sysInfo;
217✔
3121

3122
  TAOS_CHECK_GOTO(tDeserializeSTableInfoReq(pReq->pCont, pReq->contLen, &infoReq), NULL, _OVER);
217!
3123

3124
  if (0 == strcmp(infoReq.dbFName, TSDB_INFORMATION_SCHEMA_DB)) {
217✔
3125
    mInfo("information_schema table:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);
99!
3126
    TAOS_CHECK_GOTO(mndBuildInsTableSchema(pMnode, infoReq.dbFName, infoReq.tbName, sysinfo, &metaRsp), NULL, _OVER);
99!
3127
  } else if (0 == strcmp(infoReq.dbFName, TSDB_PERFORMANCE_SCHEMA_DB)) {
118✔
3128
    mInfo("performance_schema table:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);
2!
3129
    TAOS_CHECK_GOTO(mndBuildPerfsTableSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp), NULL, _OVER);
2!
3130
  } else {
3131
    mInfo("stb:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);
116!
3132
    TAOS_CHECK_GOTO(mndBuildStbSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp), NULL, _OVER);
116!
3133
  }
3134

3135
  int32_t rspLen = tSerializeSTableMetaRsp(NULL, 0, &metaRsp);
217✔
3136
  if (rspLen < 0) {
217!
3137
    code = TSDB_CODE_INVALID_MSG;
×
3138
    goto _OVER;
×
3139
  }
3140

3141
  void *pRsp = rpcMallocCont(rspLen);
217✔
3142
  if (pRsp == NULL) {
217!
3143
    code = terrno;
×
3144
    goto _OVER;
×
3145
  }
3146

3147
  if ((rspLen = tSerializeSTableMetaRsp(pRsp, rspLen, &metaRsp)) < 0) {
217!
3148
    code = rspLen;
×
3149
    goto _OVER;
×
3150
  }
3151
  pReq->info.rsp = pRsp;
217✔
3152
  pReq->info.rspLen = rspLen;
217✔
3153
  code = 0;
217✔
3154

3155
  mTrace("%s.%s, meta is retrieved", infoReq.dbFName, infoReq.tbName);
217✔
3156

3157
_OVER:
150✔
3158
  if (code != 0) {
217!
3159
    mError("stb:%s.%s, failed to retrieve meta since %s", infoReq.dbFName, infoReq.tbName, tstrerror(code));
×
3160
  }
3161

3162
  mndReleaseUser(pMnode, pUser);
217✔
3163
  tFreeSTableMetaRsp(&metaRsp);
217✔
3164
  // TODO change to TAOS_RETURN
3165
  return code;
217✔
3166
}
3167

3168
static int32_t mndProcessTableCfgReq(SRpcMsg *pReq) {
×
3169
  SMnode      *pMnode = pReq->info.node;
×
3170
  int32_t      code = -1;
×
3171
  STableCfgReq cfgReq = {0};
×
3172
  STableCfgRsp cfgRsp = {0};
×
3173

3174
  TAOS_CHECK_GOTO(tDeserializeSTableCfgReq(pReq->pCont, pReq->contLen, &cfgReq), NULL, _OVER);
×
3175

3176
  char dbName[TSDB_DB_NAME_LEN] = {0};
×
3177
  TAOS_CHECK_GOTO(mndExtractShortDbNameFromDbFullName(cfgReq.dbFName, dbName), NULL, _OVER);
×
3178
  if (0 == strcmp(dbName, TSDB_INFORMATION_SCHEMA_DB)) {
×
3179
    mInfo("information_schema table:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
×
3180
    TAOS_CHECK_GOTO(mndBuildInsTableCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp), NULL, _OVER);
×
3181
  } else if (0 == strcmp(dbName, TSDB_PERFORMANCE_SCHEMA_DB)) {
×
3182
    mInfo("performance_schema table:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
×
3183
    TAOS_CHECK_GOTO(mndBuildPerfsTableCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp), NULL, _OVER);
×
3184
  } else {
3185
    mInfo("stb:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
×
3186
    TAOS_CHECK_GOTO(mndBuildStbCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp), NULL, _OVER);
×
3187
  }
3188

3189
  int32_t rspLen = tSerializeSTableCfgRsp(NULL, 0, &cfgRsp);
×
3190
  if (rspLen < 0) {
×
3191
    code = TSDB_CODE_INVALID_MSG;
×
3192
    goto _OVER;
×
3193
  }
3194

3195
  void *pRsp = rpcMallocCont(rspLen);
×
3196
  if (pRsp == NULL) {
×
3197
    code = terrno;
×
3198
    goto _OVER;
×
3199
  }
3200

3201
  if ((rspLen = tSerializeSTableCfgRsp(pRsp, rspLen, &cfgRsp)) < 0) {
×
3202
    code = rspLen;
×
3203
    goto _OVER;
×
3204
  }
3205
  pReq->info.rsp = pRsp;
×
3206
  pReq->info.rspLen = rspLen;
×
3207
  code = 0;
×
3208

3209
  mTrace("%s.%s, cfg is retrieved", cfgReq.dbFName, cfgReq.tbName);
×
3210

3211
_OVER:
×
3212
  if (code != 0) {
×
3213
    mError("stb:%s.%s, failed to retrieve cfg since %s", cfgReq.dbFName, cfgReq.tbName, tstrerror(code));
×
3214
  }
3215

3216
  tFreeSTableCfgRsp(&cfgRsp);
×
3217
  TAOS_RETURN(code);
×
3218
}
3219

3220
int32_t mndValidateStbInfo(SMnode *pMnode, SSTableVersion *pStbVersions, int32_t numOfStbs, void **ppRsp,
222✔
3221
                           int32_t *pRspLen) {
3222
  int32_t   code = 0;
222✔
3223
  SSTbHbRsp hbRsp = {0};
222✔
3224
  hbRsp.pMetaRsp = taosArrayInit(numOfStbs, sizeof(STableMetaRsp));
222✔
3225
  if (hbRsp.pMetaRsp == NULL) {
222!
3226
    code = terrno;
×
3227
    TAOS_RETURN(code);
×
3228
  }
3229

3230
  hbRsp.pIndexRsp = taosArrayInit(numOfStbs, sizeof(STableIndexRsp));
222✔
3231
  if (NULL == hbRsp.pIndexRsp) {
222!
3232
    taosArrayDestroy(hbRsp.pMetaRsp);
×
3233
    code = terrno;
×
3234
    TAOS_RETURN(code);
×
3235
  }
3236

3237
  for (int32_t i = 0; i < numOfStbs; ++i) {
475✔
3238
    SSTableVersion *pStbVersion = &pStbVersions[i];
253✔
3239
    pStbVersion->suid = be64toh(pStbVersion->suid);
253✔
3240
    pStbVersion->sversion = ntohl(pStbVersion->sversion);
253✔
3241
    pStbVersion->tversion = ntohl(pStbVersion->tversion);
253✔
3242
    pStbVersion->smaVer = ntohl(pStbVersion->smaVer);
253✔
3243

3244
    bool    schema = false;
253✔
3245
    bool    sma = false;
253✔
3246
    int32_t code = mndValidateStbVersion(pMnode, pStbVersion, &schema, &sma);
253✔
3247
    if (TSDB_CODE_SUCCESS != code) {
253✔
3248
      STableMetaRsp metaRsp = {0};
4✔
3249
      metaRsp.numOfColumns = -1;
4✔
3250
      metaRsp.suid = pStbVersion->suid;
4✔
3251
      tstrncpy(metaRsp.dbFName, pStbVersion->dbFName, sizeof(metaRsp.dbFName));
4✔
3252
      tstrncpy(metaRsp.tbName, pStbVersion->stbName, sizeof(metaRsp.tbName));
4✔
3253
      tstrncpy(metaRsp.stbName, pStbVersion->stbName, sizeof(metaRsp.stbName));
4✔
3254
      if (taosArrayPush(hbRsp.pMetaRsp, &metaRsp) == NULL) {
8!
3255
        code = terrno;
×
3256
        return code;
×
3257
      }
3258
      continue;
4✔
3259
    }
3260

3261
    if (schema) {
249!
3262
      STableMetaRsp metaRsp = {0};
×
3263
      mInfo("stb:%s.%s, start to retrieve meta", pStbVersion->dbFName, pStbVersion->stbName);
×
3264
      if (mndBuildStbSchema(pMnode, pStbVersion->dbFName, pStbVersion->stbName, &metaRsp) != 0) {
×
3265
        metaRsp.numOfColumns = -1;
×
3266
        metaRsp.suid = pStbVersion->suid;
×
3267
        tstrncpy(metaRsp.dbFName, pStbVersion->dbFName, sizeof(metaRsp.dbFName));
×
3268
        tstrncpy(metaRsp.tbName, pStbVersion->stbName, sizeof(metaRsp.tbName));
×
3269
        tstrncpy(metaRsp.stbName, pStbVersion->stbName, sizeof(metaRsp.stbName));
×
3270
        if (taosArrayPush(hbRsp.pMetaRsp, &metaRsp) == NULL) {
×
3271
          code = terrno;
×
3272
          return code;
×
3273
        }
3274
        continue;
×
3275
      }
3276

3277
      if (taosArrayPush(hbRsp.pMetaRsp, &metaRsp) == NULL) {
×
3278
        code = terrno;
×
3279
        return code;
×
3280
      }
3281
    }
3282

3283
    if (sma) {
249!
3284
      bool           exist = false;
×
3285
      char           tbFName[TSDB_TABLE_FNAME_LEN];
3286
      STableIndexRsp indexRsp = {0};
×
3287
      indexRsp.pIndex = taosArrayInit(10, sizeof(STableIndexInfo));
×
3288
      if (NULL == indexRsp.pIndex) {
×
3289
        code = terrno;
×
3290
        TAOS_RETURN(code);
×
3291
      }
3292

3293
      (void)tsnprintf(tbFName, sizeof(tbFName), "%s.%s", pStbVersion->dbFName, pStbVersion->stbName);
×
3294
      int32_t code = mndGetTableSma(pMnode, tbFName, &indexRsp, &exist);
×
3295
      if (code || !exist) {
×
3296
        indexRsp.suid = pStbVersion->suid;
×
3297
        indexRsp.version = -1;
×
3298
        indexRsp.pIndex = NULL;
×
3299
      }
3300

3301
      tstrncpy(indexRsp.dbFName, pStbVersion->dbFName, sizeof(indexRsp.dbFName));
×
3302
      tstrncpy(indexRsp.tbName, pStbVersion->stbName, sizeof(indexRsp.tbName));
×
3303

3304
      if (taosArrayPush(hbRsp.pIndexRsp, &indexRsp) == NULL) {
×
3305
        code = terrno;
×
3306
        return code;
×
3307
      }
3308
    }
3309
  }
3310

3311
  int32_t rspLen = tSerializeSSTbHbRsp(NULL, 0, &hbRsp);
222✔
3312
  if (rspLen < 0) {
222!
3313
    tFreeSSTbHbRsp(&hbRsp);
×
3314
    code = TSDB_CODE_INVALID_MSG;
×
3315
    TAOS_RETURN(code);
×
3316
  }
3317

3318
  void *pRsp = taosMemoryMalloc(rspLen);
222!
3319
  if (pRsp == NULL) {
222!
3320
    tFreeSSTbHbRsp(&hbRsp);
×
3321
    code = terrno;
×
3322
    TAOS_RETURN(code);
×
3323
  }
3324

3325
  rspLen = tSerializeSSTbHbRsp(pRsp, rspLen, &hbRsp);
222✔
3326
  tFreeSSTbHbRsp(&hbRsp);
222✔
3327
  if (rspLen < 0) return rspLen;
222!
3328
  *ppRsp = pRsp;
222✔
3329
  *pRspLen = rspLen;
222✔
3330
  TAOS_RETURN(code);
222✔
3331
}
3332

3333
int32_t mndGetNumOfStbs(SMnode *pMnode, char *dbName, int32_t *pNumOfStbs) {
94✔
3334
  int32_t code = 0;
94✔
3335
  SSdb   *pSdb = pMnode->pSdb;
94✔
3336
  SDbObj *pDb = mndAcquireDb(pMnode, dbName);
94✔
3337
  if (pDb == NULL) {
94!
3338
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
3339
    TAOS_RETURN(code);
×
3340
  }
3341

3342
  int32_t numOfStbs = 0;
94✔
3343
  void   *pIter = NULL;
94✔
3344
  while (1) {
131✔
3345
    SStbObj *pStb = NULL;
225✔
3346
    pIter = sdbFetch(pSdb, SDB_STB, pIter, (void **)&pStb);
225✔
3347
    if (pIter == NULL) break;
225✔
3348

3349
    if (pStb->dbUid == pDb->uid) {
131✔
3350
      numOfStbs++;
45✔
3351
    }
3352

3353
    sdbRelease(pSdb, pStb);
131✔
3354
  }
3355

3356
  *pNumOfStbs = numOfStbs;
94✔
3357
  mndReleaseDb(pMnode, pDb);
94✔
3358
  TAOS_RETURN(code);
94✔
3359
}
3360

3361
int32_t mndExtractDbNameFromStbFullName(const char *stbFullName, char *dst) {
×
3362
  SName name = {0};
×
3363
  TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
×
3364

3365
  TAOS_CHECK_RETURN(tNameGetFullDbName(&name, dst));
×
3366

3367
  return 0;
×
3368
}
3369

3370
int32_t mndExtractShortDbNameFromStbFullName(const char *stbFullName, char *dst) {
×
3371
  SName name = {0};
×
3372
  TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
×
3373

3374
  TAOS_CHECK_RETURN(tNameGetDbName(&name, dst));
×
3375

3376
  return 0;
×
3377
}
3378

3379
int32_t mndExtractShortDbNameFromDbFullName(const char *stbFullName, char *dst) {
×
3380
  SName name = {0};
×
3381
  TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB));
×
3382

3383
  TAOS_CHECK_RETURN(tNameGetDbName(&name, dst));
×
3384

3385
  return 0;
×
3386
}
3387

3388
void mndExtractTbNameFromStbFullName(const char *stbFullName, char *dst, int32_t dstSize) {
79✔
3389
  int32_t pos = -1;
79✔
3390
  int32_t num = 0;
79✔
3391
  for (pos = 0; stbFullName[pos] != 0; ++pos) {
414!
3392
    if (stbFullName[pos] == TS_PATH_DELIMITER[0]) num++;
414✔
3393
    if (num == 2) break;
414✔
3394
  }
3395

3396
  if (num == 2) {
79!
3397
    tstrncpy(dst, stbFullName + pos + 1, dstSize);
79✔
3398
  }
3399
}
79✔
3400

3401
static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
38✔
3402
  SMnode  *pMnode = pReq->info.node;
38✔
3403
  SSdb    *pSdb = pMnode->pSdb;
38✔
3404
  int32_t  numOfRows = 0;
38✔
3405
  SStbObj *pStb = NULL;
38✔
3406
  int32_t  cols = 0;
38✔
3407
  int32_t  lino = 0;
38✔
3408
  int32_t  code = 0;
38✔
3409

3410
  SDbObj *pDb = NULL;
38✔
3411
  if (strlen(pShow->db) > 0) {
38✔
3412
    pDb = mndAcquireDb(pMnode, pShow->db);
33✔
3413
    if (pDb == NULL) return terrno;
33!
3414
  }
3415

3416
  while (numOfRows < rows) {
125!
3417
    pShow->pIter = sdbFetch(pSdb, SDB_STB, pShow->pIter, (void **)&pStb);
125✔
3418
    if (pShow->pIter == NULL) break;
125✔
3419

3420
    if (pDb != NULL && pStb->dbUid != pDb->uid) {
87✔
3421
      sdbRelease(pSdb, pStb);
8✔
3422
      continue;
8✔
3423
    }
3424

3425
    if (isTsmaResSTb(pStb->name)) {
79!
3426
      sdbRelease(pSdb, pStb);
×
3427
      continue;
×
3428
    }
3429

3430
    cols = 0;
79✔
3431

3432
    SName name = {0};
79✔
3433

3434
    char stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
79✔
3435
    mndExtractTbNameFromStbFullName(pStb->name, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN);
79✔
3436
    varDataSetLen(stbName, strlen(&stbName[VARSTR_HEADER_SIZE]));
79✔
3437
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
79✔
3438
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbName, false), pStb, &lino, _ERROR);
79!
3439

3440
    char db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
79✔
3441
    RETRIEVE_CHECK_GOTO(tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB), pStb, &lino, _ERROR);
79!
3442
    RETRIEVE_CHECK_GOTO(tNameGetDbName(&name, varDataVal(db)), pStb, &lino, _ERROR);
79!
3443
    varDataSetLen(db, strlen(varDataVal(db)));
79✔
3444
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
79✔
3445
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)db, false), pStb, &lino, _ERROR);
79!
3446

3447
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
79✔
3448
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->createdTime, false), pStb, &lino,
79!
3449
                        _ERROR);
3450

3451
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
79✔
3452
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->numOfColumns, false), pStb, &lino,
79!
3453
                        _ERROR);
3454

3455
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
79✔
3456
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->numOfTags, false), pStb, &lino, _ERROR);
79!
3457

3458
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
79✔
3459
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->updateTime, false), pStb, &lino,
79!
3460
                        _ERROR);  // number of tables
3461

3462
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
79✔
3463
    if (pStb->commentLen > 0) {
79✔
3464
      char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0};
1✔
3465
      STR_TO_VARSTR(comment, pStb->comment);
1✔
3466
      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, comment, false), pStb, &lino, _ERROR);
1!
3467
    } else if (pStb->commentLen == 0) {
78✔
3468
      char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};
11✔
3469
      STR_TO_VARSTR(comment, "");
11✔
3470
      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, comment, false), pStb, &lino, _ERROR);
11!
3471
    } else {
3472
      colDataSetNULL(pColInfo, numOfRows);
67!
3473
    }
3474

3475
    char watermark[64 + VARSTR_HEADER_SIZE] = {0};
79✔
3476
    (void)tsnprintf(varDataVal(watermark), sizeof(watermark) - VARSTR_HEADER_SIZE, "%" PRId64 "a,%" PRId64 "a",
79✔
3477
              pStb->watermark[0], pStb->watermark[1]);
79✔
3478
    varDataSetLen(watermark, strlen(varDataVal(watermark)));
79✔
3479

3480
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
79✔
3481
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)watermark, false), pStb, &lino, _ERROR);
79!
3482

3483
    char maxDelay[64 + VARSTR_HEADER_SIZE] = {0};
79✔
3484
    (void)tsnprintf(varDataVal(maxDelay), sizeof(maxDelay) - VARSTR_HEADER_SIZE, "%" PRId64 "a,%" PRId64 "a",
79✔
3485
              pStb->maxdelay[0], pStb->maxdelay[1]);
79✔
3486
    varDataSetLen(maxDelay, strlen(varDataVal(maxDelay)));
79✔
3487

3488
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
79✔
3489
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)maxDelay, false), pStb, &lino, _ERROR);
79!
3490

3491
    char    rollup[160 + VARSTR_HEADER_SIZE] = {0};
79✔
3492
    int32_t rollupNum = (int32_t)taosArrayGetSize(pStb->pFuncs);
79✔
3493
    char   *sep = ", ";
79✔
3494
    int32_t sepLen = strlen(sep);
79✔
3495
    int32_t rollupLen = sizeof(rollup) - VARSTR_HEADER_SIZE - 2;
79✔
3496
    for (int32_t i = 0; i < rollupNum; ++i) {
79!
3497
      char *funcName = taosArrayGet(pStb->pFuncs, i);
×
3498
      if (i) {
×
3499
        (void)strncat(varDataVal(rollup), sep, rollupLen);
×
3500
        rollupLen -= sepLen;
×
3501
      }
3502
      (void)strncat(varDataVal(rollup), funcName, rollupLen);
×
3503
      rollupLen -= strlen(funcName);
×
3504
    }
3505
    varDataSetLen(rollup, strlen(varDataVal(rollup)));
79✔
3506

3507
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
79✔
3508
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)rollup, false), pStb, &lino, _ERROR);
79!
3509

3510
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
79✔
3511
    if (pColInfo) {
79!
3512
      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)(&pStb->uid), false), pStb, &lino, _ERROR);
79!
3513
    }
3514

3515
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
79✔
3516
    if (pColInfo) {
79!
3517
      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)(&pStb->virtualStb), false), pStb, &lino, _ERROR);
79!
3518
    }
3519
    numOfRows++;
79✔
3520
    sdbRelease(pSdb, pStb);
79✔
3521
  }
3522

3523
  if (pDb != NULL) {
38✔
3524
    mndReleaseDb(pMnode, pDb);
33✔
3525
  }
3526

3527
  goto _OVER;
38✔
3528

3529
_ERROR:
×
3530
  mError("show:0x%" PRIx64 ", failed to retrieve data at %s:%d since %s", pShow->id, __FUNCTION__, lino,
×
3531
         tstrerror(code));
3532

3533
_OVER:
×
3534
  pShow->numOfRows += numOfRows;
38✔
3535
  return numOfRows;
38✔
3536
}
3537

3538
static int32_t buildDbColsInfoBlock(const SSDataBlock *p, const SSysTableMeta *pSysDbTableMeta, size_t size,
×
3539
                                    const char *dbName, const char *tbName) {
3540
  char    tName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3541
  char    dName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3542
  char    typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3543
  int32_t numOfRows = p->info.rows;
×
3544
  int32_t lino = 0;
×
3545
  int32_t code = 0;
×
3546

3547
  STR_TO_VARSTR(dName, dbName);
×
3548
  STR_TO_VARSTR(typeName, "SYSTEM_TABLE");
×
3549

3550
  for (int32_t i = 0; i < size; ++i) {
×
3551
    const SSysTableMeta *pm = &pSysDbTableMeta[i];
×
3552
    //    if (pm->sysInfo) {
3553
    //      continue;
3554
    //    }
3555
    if (tbName[0] && strncmp(tbName, pm->name, TSDB_TABLE_NAME_LEN) != 0) {
×
3556
      continue;
×
3557
    }
3558

3559
    STR_TO_VARSTR(tName, pm->name);
×
3560

3561
    for (int32_t j = 0; j < pm->colNum; j++) {
×
3562
      // table name
3563
      SColumnInfoData *pColInfoData = taosArrayGet(p->pDataBlock, 0);
×
3564
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, tName, false), &lino, _OVER);
×
3565

3566
      // database name
3567
      pColInfoData = taosArrayGet(p->pDataBlock, 1);
×
3568
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, dName, false), &lino, _OVER);
×
3569

3570
      pColInfoData = taosArrayGet(p->pDataBlock, 2);
×
3571
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, typeName, false), &lino, _OVER);
×
3572

3573
      // col name
3574
      char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3575
      STR_TO_VARSTR(colName, pm->schema[j].name);
×
3576
      pColInfoData = taosArrayGet(p->pDataBlock, 3);
×
3577
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, colName, false), &lino, _OVER);
×
3578

3579
      // col type
3580
      int8_t colType = pm->schema[j].type;
×
3581
      pColInfoData = taosArrayGet(p->pDataBlock, 4);
×
3582
      char colTypeStr[VARSTR_HEADER_SIZE + 32];
3583
      int  colTypeLen =
×
3584
          tsnprintf(varDataVal(colTypeStr), sizeof(colTypeStr) - VARSTR_HEADER_SIZE, "%s", tDataTypes[colType].name);
×
3585
      if (colType == TSDB_DATA_TYPE_VARCHAR) {
×
3586
        colTypeLen +=
×
3587
            tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE, "(%d)",
×
3588
                      (int32_t)(pm->schema[j].bytes - VARSTR_HEADER_SIZE));
×
3589
      } else if (colType == TSDB_DATA_TYPE_NCHAR) {
×
3590
        colTypeLen +=
×
3591
            tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE, "(%d)",
×
3592
                      (int32_t)((pm->schema[j].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
×
3593
      }
3594
      varDataSetLen(colTypeStr, colTypeLen);
×
3595
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, (char *)colTypeStr, false), &lino, _OVER);
×
3596

3597
      pColInfoData = taosArrayGet(p->pDataBlock, 5);
×
3598
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, (const char *)&pm->schema[j].bytes, false), &lino, _OVER);
×
3599
      for (int32_t k = 6; k <= 9; ++k) {
×
3600
        pColInfoData = taosArrayGet(p->pDataBlock, k);
×
3601
        colDataSetNULL(pColInfoData, numOfRows);
×
3602
      }
3603

3604
      numOfRows += 1;
×
3605
    }
3606
  }
3607
  return numOfRows;
×
3608
_OVER:
×
3609
  mError("failed at %s:%d since %s", __FUNCTION__, lino, tstrerror(code));
×
3610
  return numOfRows;
×
3611
}
3612
#define BUILD_COL_FOR_INFO_DB 1
3613
#define BUILD_COL_FOR_PERF_DB 1 << 1
3614
#define BUILD_COL_FOR_USER_DB 1 << 2
3615
#define BUILD_COL_FOR_ALL_DB  (BUILD_COL_FOR_INFO_DB | BUILD_COL_FOR_PERF_DB | BUILD_COL_FOR_USER_DB)
3616

3617
static int32_t buildSysDbColsInfo(SSDataBlock *p, int8_t buildWhichDBs, char *tb) {
×
3618
  size_t               size = 0;
×
3619
  const SSysTableMeta *pSysDbTableMeta = NULL;
×
3620

3621
  if (buildWhichDBs & BUILD_COL_FOR_INFO_DB) {
×
3622
    getInfosDbMeta(&pSysDbTableMeta, &size);
×
3623
    p->info.rows = buildDbColsInfoBlock(p, pSysDbTableMeta, size, TSDB_INFORMATION_SCHEMA_DB, tb);
×
3624
  }
3625

3626
  if (buildWhichDBs & BUILD_COL_FOR_PERF_DB) {
×
3627
    getPerfDbMeta(&pSysDbTableMeta, &size);
×
3628
    p->info.rows = buildDbColsInfoBlock(p, pSysDbTableMeta, size, TSDB_PERFORMANCE_SCHEMA_DB, tb);
×
3629
  }
3630

3631
  return p->info.rows;
×
3632
}
3633

3634
static int8_t determineBuildColForWhichDBs(const char *db) {
×
3635
  int8_t buildWhichDBs;
3636
  if (!db[0])
×
3637
    buildWhichDBs = BUILD_COL_FOR_ALL_DB;
×
3638
  else {
3639
    char *p = strchr(db, '.');
×
3640
    if (p && strcmp(p + 1, TSDB_INFORMATION_SCHEMA_DB) == 0) {
×
3641
      buildWhichDBs = BUILD_COL_FOR_INFO_DB;
×
3642
    } else if (p && strcmp(p + 1, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {
×
3643
      buildWhichDBs = BUILD_COL_FOR_PERF_DB;
×
3644
    } else {
3645
      buildWhichDBs = BUILD_COL_FOR_USER_DB;
×
3646
    }
3647
  }
3648
  return buildWhichDBs;
×
3649
}
3650

3651
static int32_t mndRetrieveStbCol(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
×
3652
  uint8_t  buildWhichDBs;
3653
  SMnode  *pMnode = pReq->info.node;
×
3654
  SSdb    *pSdb = pMnode->pSdb;
×
3655
  SStbObj *pStb = NULL;
×
3656
  int32_t  numOfRows = 0;
×
3657
  int32_t  lino = 0;
×
3658
  int32_t  code = 0;
×
3659

3660
  buildWhichDBs = determineBuildColForWhichDBs(pShow->db);
×
3661

3662
  if (!pShow->sysDbRsp) {
×
3663
    numOfRows = buildSysDbColsInfo(pBlock, buildWhichDBs, pShow->filterTb);
×
3664
    mDebug("mndRetrieveStbCol get system table cols, rows:%d, db:%s", numOfRows, pShow->db);
×
3665
    pShow->sysDbRsp = true;
×
3666
  }
3667

3668
  if (buildWhichDBs & BUILD_COL_FOR_USER_DB) {
×
3669
    SDbObj *pDb = NULL;
×
3670
    if (strlen(pShow->db) > 0) {
×
3671
      pDb = mndAcquireDb(pMnode, pShow->db);
×
3672
      if (pDb == NULL && TSDB_CODE_MND_DB_NOT_EXIST != terrno && pBlock->info.rows == 0) return terrno;
×
3673
    }
3674

3675
    char typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3676
    STR_TO_VARSTR(typeName, "SUPER_TABLE");
×
3677
    bool fetch = pShow->restore ? false : true;
×
3678
    pShow->restore = false;
×
3679
    while (numOfRows < rows) {
×
3680
      if (fetch) {
×
3681
        pShow->pIter = sdbFetch(pSdb, SDB_STB, pShow->pIter, (void **)&pStb);
×
3682
        if (pShow->pIter == NULL) break;
×
3683
      } else {
3684
        fetch = true;
×
3685
        void *pKey = taosHashGetKey(pShow->pIter, NULL);
×
3686
        pStb = sdbAcquire(pSdb, SDB_STB, pKey);
×
3687
        if (!pStb) continue;
×
3688
      }
3689

3690
      if (pDb != NULL && pStb->dbUid != pDb->uid) {
×
3691
        sdbRelease(pSdb, pStb);
×
3692
        continue;
×
3693
      }
3694

3695
      SName name = {0};
×
3696
      char  stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3697
      mndExtractTbNameFromStbFullName(pStb->name, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN);
×
3698
      if (pShow->filterTb[0] && strncmp(pShow->filterTb, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN) != 0) {
×
3699
        sdbRelease(pSdb, pStb);
×
3700
        continue;
×
3701
      }
3702

3703
      if ((numOfRows + pStb->numOfColumns) > rows) {
×
3704
        pShow->restore = true;
×
3705
        if (numOfRows == 0) {
×
3706
          mError("mndRetrieveStbCol failed to get stable cols since buf:%d less than result:%d, stable name:%s, db:%s",
×
3707
                 rows, pStb->numOfColumns, pStb->name, pStb->db);
3708
        }
3709
        sdbRelease(pSdb, pStb);
×
3710
        break;
×
3711
      }
3712

3713
      varDataSetLen(stbName, strlen(&stbName[VARSTR_HEADER_SIZE]));
×
3714

3715
      mDebug("mndRetrieveStbCol get stable cols, stable name:%s, db:%s", pStb->name, pStb->db);
×
3716

3717
      char db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3718
      RETRIEVE_CHECK_GOTO(tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB), pStb, &lino, _OVER);
×
3719
      RETRIEVE_CHECK_GOTO(tNameGetDbName(&name, varDataVal(db)), pStb, &lino, _OVER);
×
3720
      varDataSetLen(db, strlen(varDataVal(db)));
×
3721

3722
      for (int i = 0; i < pStb->numOfColumns; i++) {
×
3723
        int32_t          cols = 0;
×
3724
        SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
3725
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbName, false), pStb, &lino, _OVER);
×
3726

3727
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
3728
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)db, false), pStb, &lino, _OVER);
×
3729

3730
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
3731
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, typeName, false), pStb, &lino, _OVER);
×
3732

3733
        // col name
3734
        char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3735
        STR_TO_VARSTR(colName, pStb->pColumns[i].name);
×
3736
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
3737
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, colName, false), pStb, &lino, _OVER);
×
3738

3739
        // col type
3740
        int8_t colType = pStb->pColumns[i].type;
×
3741
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
3742
        char colTypeStr[VARSTR_HEADER_SIZE + 32];
3743
        int  colTypeLen =
×
3744
            tsnprintf(varDataVal(colTypeStr), sizeof(colTypeStr) - VARSTR_HEADER_SIZE, "%s", tDataTypes[colType].name);
×
3745
        if (colType == TSDB_DATA_TYPE_VARCHAR) {
×
3746
          colTypeLen +=
×
3747
              tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE,
×
3748
                        "(%d)", (int32_t)(pStb->pColumns[i].bytes - VARSTR_HEADER_SIZE));
×
3749
        } else if (colType == TSDB_DATA_TYPE_NCHAR) {
×
3750
          colTypeLen +=
×
3751
              tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE,
×
3752
                        "(%d)", (int32_t)((pStb->pColumns[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
×
3753
        } else if (IS_DECIMAL_TYPE(colType)) {
×
3754
          STypeMod typeMod = pStb->pExtSchemas[i].typeMod;
×
3755
          uint8_t prec = 0, scale = 0;
×
3756
          decimalFromTypeMod(typeMod, &prec, &scale);
×
3757
          colTypeLen += sprintf(varDataVal(colTypeStr) + colTypeLen, "(%d,%d)", prec, scale);
×
3758
        }
3759
        varDataSetLen(colTypeStr, colTypeLen);
×
3760
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (char *)colTypeStr, false), pStb, &lino, _OVER);
×
3761

3762
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
3763
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->pColumns[i].bytes, false), pStb,
×
3764
                            &lino, _OVER);
3765
        while (cols < pShow->numOfColumns) {
×
3766
          pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
3767
          colDataSetNULL(pColInfo, numOfRows);
×
3768
        }
3769
        numOfRows++;
×
3770
      }
3771

3772
      sdbRelease(pSdb, pStb);
×
3773
    }
3774

3775
    if (pDb != NULL) {
×
3776
      mndReleaseDb(pMnode, pDb);
×
3777
    }
3778
  }
3779

3780
  mDebug("mndRetrieveStbCol success, rows:%d, pShow->numOfRows:%d", numOfRows, pShow->numOfRows);
×
3781
  goto _OVER;
×
3782

3783
_ERROR:
3784
  mError("failed to mndRetrieveStbCol, rows:%d, pShow->numOfRows:%d, at %s:%d since %s", numOfRows, pShow->numOfRows,
3785
         __FUNCTION__, lino, tstrerror(code));
3786

3787
_OVER:
×
3788
  pShow->numOfRows += numOfRows;
×
3789
  return numOfRows;
×
3790
}
3791

3792
static void mndCancelGetNextStb(SMnode *pMnode, void *pIter) {
×
3793
  SSdb *pSdb = pMnode->pSdb;
×
3794
  sdbCancelFetchByType(pSdb, pIter, SDB_STB);
×
3795
}
×
3796

3797
const char *mndGetStbStr(const char *src) {
8✔
3798
  char *posDb = strstr(src, TS_PATH_DELIMITER);
8✔
3799
  if (posDb != NULL) ++posDb;
8!
3800
  if (posDb == NULL) return src;
8!
3801

3802
  char *posStb = strstr(posDb, TS_PATH_DELIMITER);
8✔
3803
  if (posStb != NULL) ++posStb;
8!
3804
  if (posStb == NULL) return posDb;
8!
3805
  return posStb;
8✔
3806
}
3807

3808
static int32_t mndCheckIndexReq(SCreateTagIndexReq *pReq) {
×
3809
  // impl
3810
  return TSDB_CODE_SUCCESS;
×
3811
}
3812

3813
/*int32_t mndAddIndexImpl(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp, void *sql,
3814
                        int32_t len) {
3815
  // impl later
3816
  int32_t code = 0;
3817
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "create-stb-index");
3818
  if (pTrans == NULL) goto _OVER;
3819

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

3824
  if (mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb) != 0) goto _OVER;
3825
  if (mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb) != 0) goto _OVER;
3826
  if (mndSetAlterStbRedoActions2(pMnode, pTrans, pDb, pStb, sql, len) != 0) goto _OVER;
3827
  if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
3828

3829
  return code;
3830

3831
_OVER:
3832
  mndTransDrop(pTrans);
3833
  return code;
3834
}
3835
static int32_t mndAddIndex(SMnode *pMnode, SRpcMsg *pReq, SCreateTagIndexReq *tagIdxReq, SDbObj *pDb, SStbObj *pOld) {
3836
  bool    needRsp = true;
3837
  int32_t code = -1;
3838
  SField *pField0 = NULL;
3839

3840
  SStbObj  stbObj = {0};
3841
  SStbObj *pNew = &stbObj;
3842

3843
  taosRLockLatch(&pOld->lock);
3844
  memcpy(&stbObj, pOld, sizeof(SStbObj));
3845
  taosRUnLockLatch(&pOld->lock);
3846

3847
  stbObj.pColumns = NULL;
3848
  stbObj.pTags = NULL;
3849
  stbObj.updateTime = taosGetTimestampMs();
3850
  stbObj.lock = 0;
3851

3852
  int32_t tag = mndFindSuperTableTagIndex(pOld, tagIdxReq->colName);
3853
  if (tag < 0) {
3854
    terrno = TSDB_CODE_MND_TAG_NOT_EXIST;
3855
    return -1;
3856
  }
3857
  if (mndAllocStbSchemas(pOld, pNew) != 0) {
3858
    return -1;
3859
  }
3860

3861
  SSchema *pTag = pNew->pTags + tag;
3862
  if (IS_IDX_ON(pTag)) {
3863
    terrno = TSDB_CODE_MND_TAG_INDEX_ALREADY_EXIST;
3864
    return -1;
3865
  } else {
3866
    pTag->flags |= COL_IDX_ON;
3867
  }
3868
  pNew->tagVer++;
3869

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

3872
  return code;
3873
}
3874
static int32_t mndProcessCreateIndexReq(SRpcMsg *pReq) {
3875
  SMnode            *pMnode = pReq->info.node;
3876
  int32_t            code = -1;
3877
  SDbObj            *pDb = NULL;
3878
  SStbObj           *pStb = NULL;
3879
  SCreateTagIndexReq tagIdxReq = {0};
3880

3881
  if (tDeserializeSCreateTagIdxReq(pReq->pCont, pReq->contLen, &tagIdxReq) != 0) {
3882
    terrno = TSDB_CODE_INVALID_MSG;
3883
    goto _OVER;
3884
  }
3885

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

3888
  if (mndCheckIndexReq(&tagIdxReq) != TSDB_CODE_SUCCESS) {
3889
    goto _OVER;
3890
  }
3891

3892
  pDb = mndAcquireDbByStb(pMnode, tagIdxReq.dbFName);
3893
  if (pDb == NULL) {
3894
    terrno = TSDB_CODE_MND_DB_NOT_EXIST;
3895
    goto _OVER;
3896
  }
3897

3898
  pStb = mndAcquireStb(pMnode, tagIdxReq.stbName);
3899
  if (pStb == NULL) {
3900
    terrno = TSDB_CODE_MND_STB_NOT_EXIST;
3901
    goto _OVER;
3902
  }
3903
  if (mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_WRITE_DB, pDb) != 0) {
3904
    goto _OVER;
3905
  }
3906

3907
  code = mndAddIndex(pMnode, pReq, &tagIdxReq, pDb, pStb);
3908
  if (terrno == TSDB_CODE_MND_TAG_INDEX_ALREADY_EXIST || terrno == TSDB_CODE_MND_TAG_NOT_EXIST) {
3909
    return terrno;
3910
  } else {
3911
    if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
3912
  }
3913
_OVER:
3914
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
3915
    mError("stb:%s, failed to create index since %s", tagIdxReq.stbName, terrstr());
3916
  }
3917
  mndReleaseStb(pMnode, pStb);
3918
  mndReleaseDb(pMnode, pDb);
3919
  return code;
3920
}
3921
static int32_t mndProcessDropIndexReq(SRpcMsg *pReq) {
3922
  SMnode          *pMnode = pReq->info.node;
3923
  int32_t          code = -1;
3924
  SDbObj          *pDb = NULL;
3925
  SStbObj         *pStb = NULL;
3926
  SDropTagIndexReq dropReq = {0};
3927
  if (tDeserializeSDropTagIdxReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
3928
    terrno = TSDB_CODE_INVALID_MSG;
3929
    goto _OVER;
3930
  }
3931
  //
3932
  return TSDB_CODE_SUCCESS;
3933
_OVER:
3934
  return code;
3935
}*/
3936

3937
static int32_t mndProcessDropStbReqFromMNode(SRpcMsg *pReq) {
×
3938
  int32_t code = mndProcessDropStbReq(pReq);
×
3939
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
×
3940
    pReq->info.rsp = rpcMallocCont(1);
×
3941
    pReq->info.rspLen = 1;
×
3942
    pReq->info.noResp = false;
×
3943
    pReq->code = code;
×
3944
  }
3945
  return code;
×
3946
}
3947

3948
typedef struct SVDropTbVgReqs {
3949
  SArray     *pBatchReqs;
3950
  SVgroupInfo info;
3951
} SVDropTbVgReqs;
3952

3953
typedef struct SMDropTbDbInfo {
3954
  SArray *dbVgInfos;
3955
  int32_t hashPrefix;
3956
  int32_t hashSuffix;
3957
  int32_t hashMethod;
3958
} SMDropTbDbInfo;
3959

3960
typedef struct SMDropTbTsmaInfo {
3961
  char           tsmaResTbDbFName[TSDB_DB_FNAME_LEN];
3962
  char           tsmaResTbNamePrefix[TSDB_TABLE_FNAME_LEN];
3963
  int32_t        suid;
3964
  SMDropTbDbInfo dbInfo;  // reference to DbInfo in pDbMap
3965
} SMDropTbTsmaInfo;
3966

3967
typedef struct SMDropTbTsmaInfos {
3968
  SArray *pTsmaInfos;  // SMDropTbTsmaInfo
3969
} SMDropTbTsmaInfos;
3970

3971
typedef struct SMndDropTbsWithTsmaCtx {
3972
  SHashObj *pVgMap;  // <vgId, SVDropTbVgReqs>
3973
} SMndDropTbsWithTsmaCtx;
3974

3975
static int32_t mndDropTbForSingleVg(SMnode *pMnode, SMndDropTbsWithTsmaCtx *pCtx, SArray *pTbs, int32_t vgId);
3976

3977
static void destroySVDropTbBatchReqs(void *p);
3978
static void mndDestroyDropTbsWithTsmaCtx(SMndDropTbsWithTsmaCtx *p) {
×
3979
  if (!p) return;
×
3980

3981
  if (p->pVgMap) {
×
3982
    void *pIter = taosHashIterate(p->pVgMap, NULL);
×
3983
    while (pIter) {
×
3984
      SVDropTbVgReqs *pReqs = pIter;
×
3985
      taosArrayDestroyEx(pReqs->pBatchReqs, destroySVDropTbBatchReqs);
×
3986
      pIter = taosHashIterate(p->pVgMap, pIter);
×
3987
    }
3988
    taosHashCleanup(p->pVgMap);
×
3989
  }
3990
  taosMemoryFree(p);
×
3991
}
3992

3993
static int32_t mndInitDropTbsWithTsmaCtx(SMndDropTbsWithTsmaCtx **ppCtx) {
×
3994
  int32_t                 code = 0;
×
3995
  SMndDropTbsWithTsmaCtx *pCtx = taosMemoryCalloc(1, sizeof(SMndDropTbsWithTsmaCtx));
×
3996
  if (!pCtx) return terrno;
×
3997

3998
  pCtx->pVgMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
×
3999
  if (!pCtx->pVgMap) {
×
4000
    code = terrno;
×
4001
    goto _end;
×
4002
  }
4003

4004
  *ppCtx = pCtx;
×
4005
_end:
×
4006
  if (code) mndDestroyDropTbsWithTsmaCtx(pCtx);
×
4007
  return code;
×
4008
}
4009

4010
static void *mndBuildVDropTbsReq(SMnode *pMnode, const SVgroupInfo *pVgInfo, const SVDropTbBatchReq *pReq,
×
4011
                                 int32_t *len) {
4012
  int32_t   contLen = 0;
×
4013
  int32_t   ret = 0;
×
4014
  SMsgHead *pHead = NULL;
×
4015
  SEncoder  encoder = {0};
×
4016

4017
  tEncodeSize(tEncodeSVDropTbBatchReq, pReq, contLen, ret);
×
4018
  if (ret < 0) return NULL;
×
4019

4020
  contLen += sizeof(SMsgHead);
×
4021
  pHead = taosMemoryMalloc(contLen);
×
4022
  if (pHead == NULL) {
×
4023
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
4024
    return NULL;
×
4025
  }
4026

4027
  pHead->contLen = htonl(contLen);
×
4028
  pHead->vgId = htonl(pVgInfo->vgId);
×
4029

4030
  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
×
4031

4032
  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
×
4033
  int32_t code = tEncodeSVDropTbBatchReq(&encoder, pReq);
×
4034
  tEncoderClear(&encoder);
×
4035
  if (code != 0) return NULL;
×
4036

4037
  *len = contLen;
×
4038
  return pHead;
×
4039
}
4040

4041
static int32_t mndSetDropTbsRedoActions(SMnode *pMnode, STrans *pTrans, const SVDropTbVgReqs *pVgReqs, void *pCont,
×
4042
                                        int32_t contLen, tmsg_t msgType) {
4043
  STransAction action = {0};
×
4044
  action.epSet = pVgReqs->info.epSet;
×
4045
  action.pCont = pCont;
×
4046
  action.contLen = contLen;
×
4047
  action.msgType = msgType;
×
4048
  action.acceptableCode = TSDB_CODE_TDB_TABLE_NOT_EXIST;
×
4049
  return mndTransAppendRedoAction(pTrans, &action);
×
4050
}
4051

4052
static int32_t mndBuildDropTbRedoActions(SMnode *pMnode, STrans *pTrans, SHashObj *pVgMap, tmsg_t msgType) {
×
4053
  int32_t code = 0;
×
4054
  void   *pIter = taosHashIterate(pVgMap, NULL);
×
4055
  while (pIter) {
×
4056
    const SVDropTbVgReqs *pVgReqs = pIter;
×
4057
    int32_t               len = 0;
×
4058
    for (int32_t i = 0; i < taosArrayGetSize(pVgReqs->pBatchReqs) && code == TSDB_CODE_SUCCESS; ++i) {
×
4059
      SVDropTbBatchReq *pBatchReq = taosArrayGet(pVgReqs->pBatchReqs, i);
×
4060
      void             *p = mndBuildVDropTbsReq(pMnode, &pVgReqs->info, pBatchReq, &len);
×
4061
      if (!p) {
×
4062
        code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
4063
        if (terrno != 0) code = terrno;
×
4064
        break;
×
4065
      }
4066
      if ((code = mndSetDropTbsRedoActions(pMnode, pTrans, pVgReqs, p, len, msgType)) != 0) {
×
4067
        break;
×
4068
      }
4069
    }
4070
    if (TSDB_CODE_SUCCESS != code) {
×
4071
      taosHashCancelIterate(pVgMap, pIter);
×
4072
      break;
×
4073
    }
4074
    pIter = taosHashIterate(pVgMap, pIter);
×
4075
  }
4076
  return code;
×
4077
}
4078

4079
static int32_t mndCreateDropTbsTxnPrepare(SRpcMsg *pRsp, SMndDropTbsWithTsmaCtx *pCtx) {
×
4080
  int32_t code = 0;
×
4081
  SMnode *pMnode = pRsp->info.node;
×
4082
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, pRsp, "drop-tbs");
×
4083
  mndTransSetChangeless(pTrans);
×
4084
  mndTransSetSerial(pTrans);
×
4085
  if (pTrans == NULL) {
×
4086
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
4087
    if (terrno != 0) code = terrno;
×
4088
    goto _OVER;
×
4089
  }
4090

4091
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
×
4092

4093
  if ((code = mndBuildDropTbRedoActions(pMnode, pTrans, pCtx->pVgMap, TDMT_VND_DROP_TABLE)) != 0) goto _OVER;
×
4094
  if ((code = mndTransPrepare(pMnode, pTrans)) != 0) goto _OVER;
×
4095

4096
_OVER:
×
4097
  mndTransDrop(pTrans);
×
4098
  TAOS_RETURN(code);
×
4099
}
4100

4101
static int32_t mndProcessDropTbWithTsma(SRpcMsg *pReq) {
×
4102
  int32_t      code = -1;
×
4103
  SMnode      *pMnode = pReq->info.node;
×
4104
  SDbObj      *pDb = NULL;
×
4105
  SStbObj     *pStb = NULL;
×
4106
  SMDropTbsReq dropReq = {0};
×
4107
  bool         locked = false;
×
4108
  if (tDeserializeSMDropTbsReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
×
4109
    code = TSDB_CODE_INVALID_MSG;
×
4110
    goto _OVER;
×
4111
  }
4112

4113
  SMndDropTbsWithTsmaCtx *pCtx = NULL;
×
4114
  code = mndInitDropTbsWithTsmaCtx(&pCtx);
×
4115
  if (code) goto _OVER;
×
4116
  for (int32_t i = 0; i < dropReq.pVgReqs->size; ++i) {
×
4117
    SMDropTbReqsOnSingleVg *pReq = taosArrayGet(dropReq.pVgReqs, i);
×
4118
    code = mndDropTbForSingleVg(pMnode, pCtx, pReq->pTbs, pReq->vgInfo.vgId);
×
4119
    if (code) goto _OVER;
×
4120
  }
4121
  code = mndCreateDropTbsTxnPrepare(pReq, pCtx);
×
4122
  if (code == 0) {
×
4123
    code = TSDB_CODE_ACTION_IN_PROGRESS;
×
4124
  }
4125
_OVER:
×
4126
  tFreeSMDropTbsReq(&dropReq);
×
4127
  if (pCtx) mndDestroyDropTbsWithTsmaCtx(pCtx);
×
4128
  TAOS_RETURN(code);
×
4129
}
4130

4131
static int32_t createDropTbBatchReq(const SVDropTbReq *pReq, SVDropTbBatchReq *pBatchReq) {
×
4132
  pBatchReq->nReqs = 1;
×
4133
  pBatchReq->pArray = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SVDropTbReq));
×
4134
  if (!pBatchReq->pArray) return terrno;
×
4135
  if (taosArrayPush(pBatchReq->pArray, pReq) == NULL) {
×
4136
    taosArrayDestroy(pBatchReq->pArray);
×
4137
    pBatchReq->pArray = NULL;
×
4138
    return terrno;
×
4139
  }
4140
  return TSDB_CODE_SUCCESS;
×
4141
}
4142

4143
static void destroySVDropTbBatchReqs(void *p) {
×
4144
  SVDropTbBatchReq *pReq = p;
×
4145
  taosArrayDestroy(pReq->pArray);
×
4146
  pReq->pArray = NULL;
×
4147
}
×
4148

4149
static int32_t mndDropTbAdd(SMnode *pMnode, SHashObj *pVgHashMap, const SVgroupInfo *pVgInfo, char *name, tb_uid_t suid,
×
4150
                            bool ignoreNotExists) {
4151
  SVDropTbReq req = {.name = name, .suid = suid, .igNotExists = ignoreNotExists, .uid = 0};
×
4152

4153
  SVDropTbVgReqs *pVgReqs = taosHashGet(pVgHashMap, &pVgInfo->vgId, sizeof(pVgInfo->vgId));
×
4154
  SVDropTbVgReqs  vgReqs = {0};
×
4155
  if (pVgReqs == NULL) {
×
4156
    vgReqs.info = *pVgInfo;
×
4157
    vgReqs.pBatchReqs = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SVDropTbBatchReq));
×
4158
    if (!vgReqs.pBatchReqs) return terrno;
×
4159
    SVDropTbBatchReq batchReq = {0};
×
4160
    int32_t          code = createDropTbBatchReq(&req, &batchReq);
×
4161
    if (TSDB_CODE_SUCCESS != code) return code;
×
4162
    if (taosArrayPush(vgReqs.pBatchReqs, &batchReq) == NULL) {
×
4163
      taosArrayDestroy(batchReq.pArray);
×
4164
      return terrno;
×
4165
    }
4166
    if (taosHashPut(pVgHashMap, &pVgInfo->vgId, sizeof(pVgInfo->vgId), &vgReqs, sizeof(vgReqs)) != 0) {
×
4167
      taosArrayDestroyEx(vgReqs.pBatchReqs, destroySVDropTbBatchReqs);
×
4168
      return terrno;
×
4169
    }
4170
  } else {
4171
    SVDropTbBatchReq batchReq = {0};
×
4172
    int32_t          code = createDropTbBatchReq(&req, &batchReq);
×
4173
    if (TSDB_CODE_SUCCESS != code) return code;
×
4174
    if (taosArrayPush(pVgReqs->pBatchReqs, &batchReq) == NULL) {
×
4175
      taosArrayDestroy(batchReq.pArray);
×
4176
      return terrno;
×
4177
    }
4178
  }
4179
  return 0;
×
4180
}
4181

4182
static int32_t mndDropTbForSingleVg(SMnode *pMnode, SMndDropTbsWithTsmaCtx *pCtx, SArray *pTbs, int32_t vgId) {
×
4183
  int32_t code = 0;
×
4184

4185
  SVgObj *pVgObj = mndAcquireVgroup(pMnode, vgId);
×
4186
  if (!pVgObj) {
×
4187
    code = 0;
×
4188
    goto _end;
×
4189
  }
4190
  SVgroupInfo vgInfo = {.hashBegin = pVgObj->hashBegin,
×
4191
                        .hashEnd = pVgObj->hashEnd,
×
4192
                        .numOfTable = pVgObj->numOfTables,
×
4193
                        .vgId = pVgObj->vgId};
×
4194
  vgInfo.epSet = mndGetVgroupEpset(pMnode, pVgObj);
×
4195
  mndReleaseVgroup(pMnode, pVgObj);
×
4196

4197
  for (int32_t i = 0; i < pTbs->size; ++i) {
×
4198
    SVDropTbReq *pTb = taosArrayGet(pTbs, i);
×
4199
    TAOS_CHECK_GOTO(mndDropTbAdd(pMnode, pCtx->pVgMap, &vgInfo, pTb->name, pTb->suid, pTb->igNotExists), NULL, _end);
×
4200
  }
4201
_end:
×
4202
  return code;
×
4203
}
4204

4205
static int32_t mndProcessFetchTtlExpiredTbs(SRpcMsg *pRsp) {
511✔
4206
  int32_t                 code = -1;
511✔
4207
  SDecoder                decoder = {0};
511✔
4208
  SMnode                 *pMnode = pRsp->info.node;
511✔
4209
  SVFetchTtlExpiredTbsRsp rsp = {0};
511✔
4210
  SMndDropTbsWithTsmaCtx *pCtx = NULL;
511✔
4211
  if (pRsp->code != TSDB_CODE_SUCCESS) {
511✔
4212
    code = pRsp->code;
10✔
4213
    goto _end;
10✔
4214
  }
4215
  if (pRsp->contLen == 0) {
501!
4216
    code = 0;
501✔
4217
    goto _end;
501✔
4218
  }
4219

4220
  tDecoderInit(&decoder, pRsp->pCont, pRsp->contLen);
×
4221
  code = tDecodeVFetchTtlExpiredTbsRsp(&decoder, &rsp);
×
4222
  if (code) goto _end;
×
4223

4224
  code = mndInitDropTbsWithTsmaCtx(&pCtx);
×
4225
  if (code) goto _end;
×
4226

4227
  code = mndDropTbForSingleVg(pMnode, pCtx, rsp.pExpiredTbs, rsp.vgId);
×
4228
  if (code) goto _end;
×
4229
  code = mndCreateDropTbsTxnPrepare(pRsp, pCtx);
×
4230
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
×
4231
_end:
×
4232
  if (pCtx) mndDestroyDropTbsWithTsmaCtx(pCtx);
511!
4233
  tDecoderClear(&decoder);
511✔
4234
  tFreeFetchTtlExpiredTbsRsp(&rsp);
511✔
4235
  TAOS_RETURN(code);
511✔
4236
}
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