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

taosdata / TDengine / #4986

15 Mar 2026 08:32AM UTC coverage: 37.305% (-31.3%) from 68.601%
#4986

push

travis-ci

tomchon
test: keep docs and unit test

125478 of 336361 relevant lines covered (37.3%)

1134847.06 hits per line

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

47.91
/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 "mndRsma.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 "mndStream.h"
35
#include "tname.h"
36

37
#define STB_VER_SUPPORT_COMP    2
38
#define STB_VER_SUPPORT_VIRTUAL 3
39
#define STB_VER_SUPPORT_OWNER   4
40
#define STB_VER_NUMBER          STB_VER_SUPPORT_OWNER
41
#define STB_RESERVE_SIZE        56
42

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

64
static int32_t mndProcessCreateIndexReq(SRpcMsg *pReq);
65
static int32_t mndProcessDropIndexReq(SRpcMsg *pReq);
66

67
static int32_t mndProcessDropStbReqFromMNode(SRpcMsg *pReq);
68
static int32_t mndProcessDropTbWithTsma(SRpcMsg *pReq);
69
static int32_t mndProcessFetchTtlExpiredTbs(SRpcMsg *pReq);
70
static int32_t mndProcessAuditRecordRsp(SRpcMsg *pRsp);
71

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

83
  mndSetMsgHandle(pMnode, TDMT_MND_CREATE_STB, mndProcessCreateStbReq);
16✔
84
  mndSetMsgHandle(pMnode, TDMT_MND_ALTER_STB, mndProcessAlterStbReq);
16✔
85
  mndSetMsgHandle(pMnode, TDMT_MND_DROP_STB, mndProcessDropStbReq);
16✔
86
  mndSetMsgHandle(pMnode, TDMT_VND_CREATE_STB_RSP, mndTransProcessRsp);
16✔
87
  mndSetMsgHandle(pMnode, TDMT_VND_DROP_TTL_TABLE_RSP, mndProcessDropTtltbRsp);
16✔
88
  mndSetMsgHandle(pMnode, TDMT_VND_TRIM_RSP, mndTransProcessRsp);
16✔
89
  mndSetMsgHandle(pMnode, TDMT_VND_TRIM_WAL_RSP, mndProcessTrimDbWalRsp);
16✔
90
  mndSetMsgHandle(pMnode, TDMT_VND_ALTER_STB_RSP, mndTransProcessRsp);
16✔
91
  mndSetMsgHandle(pMnode, TDMT_VND_DROP_STB_RSP, mndTransProcessRsp);
16✔
92
  mndSetMsgHandle(pMnode, TDMT_MND_TABLE_META, mndProcessTableMetaReq);
16✔
93
  mndSetMsgHandle(pMnode, TDMT_MND_TTL_TIMER, mndProcessTtlTimer);
16✔
94
  // mndSetMsgHandle(pMnode, TDMT_MND_TRIM_DB_TIMER, mndProcessTrimDbTimer);
95
  mndSetMsgHandle(pMnode, TDMT_MND_TABLE_CFG, mndProcessTableCfgReq);
16✔
96
  mndSetMsgHandle(pMnode, TDMT_MND_STB_DROP, mndProcessDropStbReqFromMNode);
16✔
97
  mndSetMsgHandle(pMnode, TDMT_MND_STB_DROP_RSP, mndTransProcessRsp);
16✔
98
  mndSetMsgHandle(pMnode, TDMT_MND_DROP_TB_WITH_TSMA, mndProcessDropTbWithTsma);
16✔
99
  mndSetMsgHandle(pMnode, TDMT_VND_FETCH_TTL_EXPIRED_TBS_RSP, mndProcessFetchTtlExpiredTbs);
16✔
100
  mndSetMsgHandle(pMnode, TDMT_VND_DROP_TABLE_RSP, mndTransProcessRsp);
16✔
101
  //  mndSetMsgHandle(pMnode, TDMT_MND_SYSTABLE_RETRIEVE, mndProcessRetrieveStbReq);
102

103
  // mndSetMsgHandle(pMnode, TDMT_MND_CREATE_INDEX, mndProcessCreateIndexReq);
104
  // mndSetMsgHandle(pMnode, TDMT_MND_DROP_INDEX, mndProcessDropIndexReq);
105
  // mndSetMsgHandle(pMnode, TDMT_VND_CREATE_INDEX_RSP, mndTransProcessRsp);
106
  // mndSetMsgHandle(pMnode, TDMT_VND_DROP_INDEX_RSP, mndTransProcessRsp);
107
  mndSetMsgHandle(pMnode, TDMT_VND_AUDIT_RECORD_RSP, mndProcessAuditRecordRsp);
16✔
108

109
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_STB, mndRetrieveStb);
16✔
110
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_STB, mndCancelGetNextStb);
16✔
111

112
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_COL, mndRetrieveStbCol);
16✔
113
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_COL, mndCancelGetNextStb);
16✔
114

115
  return sdbSetTable(pMnode->pSdb, table);
16✔
116
}
117

118
void mndCleanupStb(SMnode *pMnode) {}
16✔
119

120
SSdbRaw *mndStbActionEncode(SStbObj *pStb) {
72✔
121
  int32_t code = 0;
72✔
122
  int32_t lino = 0;
72✔
123
  bool    hasTypeMod = false;
72✔
124
  terrno = TSDB_CODE_OUT_OF_MEMORY;
72✔
125

126
  int32_t size = sizeof(SStbObj) + (pStb->numOfColumns + pStb->numOfTags) * sizeof(SSchema) + pStb->commentLen +
144✔
127
                 pStb->ast1Len + pStb->ast2Len + pStb->numOfColumns * sizeof(SColCmpr) + STB_RESERVE_SIZE +
144✔
128
                 taosArrayGetSize(pStb->pFuncs) * TSDB_FUNC_NAME_LEN + sizeof(int32_t) * pStb->numOfColumns;
72✔
129
  SSdbRaw *pRaw = sdbAllocRaw(SDB_STB, STB_VER_NUMBER, size);
72✔
130
  if (pRaw == NULL) goto _OVER;
72✔
131

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

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

165
  for (int32_t i = 0; i < pStb->numOfTags; ++i) {
276✔
166
    SSchema *pSchema = &pStb->pTags[i];
204✔
167
    SDB_SET_INT8(pRaw, dataPos, pSchema->type, _OVER)
204✔
168
    SDB_SET_INT8(pRaw, dataPos, pSchema->flags, _OVER)
204✔
169
    SDB_SET_INT16(pRaw, dataPos, pSchema->colId, _OVER)
204✔
170
    SDB_SET_INT32(pRaw, dataPos, pSchema->bytes, _OVER)
204✔
171
    SDB_SET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, _OVER)
204✔
172
  }
173

174
  for (int32_t i = 0; i < pStb->numOfFuncs; ++i) {
72✔
175
    char *func = taosArrayGet(pStb->pFuncs, i);
×
176
    SDB_SET_BINARY(pRaw, dataPos, func, TSDB_FUNC_NAME_LEN, _OVER)
×
177
  }
178

179
  if (pStb->commentLen > 0) {
72✔
180
    SDB_SET_BINARY(pRaw, dataPos, pStb->comment, pStb->commentLen + 1, _OVER)
×
181
  }
182

183
  if (pStb->ast1Len > 0) {
72✔
184
    SDB_SET_BINARY(pRaw, dataPos, pStb->pAst1, pStb->ast1Len, _OVER)
×
185
  }
186

187
  if (pStb->ast2Len > 0) {
72✔
188
    SDB_SET_BINARY(pRaw, dataPos, pStb->pAst2, pStb->ast2Len, _OVER)
×
189
  }
190

191
  if (pStb->pCmpr != NULL) {
72✔
192
    for (int i = 0; i < pStb->numOfColumns; i++) {
232✔
193
      SColCmpr *p = &pStb->pCmpr[i];
160✔
194
      SDB_SET_INT16(pRaw, dataPos, p->id, _OVER)
160✔
195
      SDB_SET_INT32(pRaw, dataPos, p->alg, _OVER)
160✔
196
    }
197
  }
198
  SDB_SET_INT64(pRaw, dataPos, pStb->keep, _OVER)
72✔
199

200
  if (hasTypeMod) {
72✔
201
    for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
×
202
      SDB_SET_INT32(pRaw, dataPos, pStb->pExtSchemas[i].typeMod, _OVER);
×
203
    }
204
  }
205

206
  SDB_SET_INT8(pRaw, dataPos, pStb->virtualStb, _OVER)
72✔
207
  // since 3.4.0.0 - STB_VER_SUPPORT_OWNER
208
  SDB_SET_BINARY(pRaw, dataPos, pStb->createUser, TSDB_USER_LEN, _OVER)
72✔
209
  SDB_SET_INT64(pRaw, dataPos, pStb->ownerId, _OVER)
72✔
210
  SDB_SET_RESERVE(pRaw, dataPos, STB_RESERVE_SIZE, _OVER)
72✔
211
  SDB_SET_DATALEN(pRaw, dataPos, _OVER)
72✔
212

213
  terrno = 0;
72✔
214

215
_OVER:
72✔
216
  if (terrno != 0) {
72✔
217
    mError("stb:%s, failed to encode to raw:%p since %s", pStb->name, pRaw, terrstr());
×
218
    sdbFreeRaw(pRaw);
×
219
    return NULL;
×
220
  }
221

222
  mTrace("stb:%s, encode to raw:%p, row:%p", pStb->name, pRaw, pStb);
72✔
223
  return pRaw;
72✔
224
}
225

226
SSdbRow *mndStbActionDecode(SSdbRaw *pRaw) {
72✔
227
  int32_t code = 0;
72✔
228
  int32_t lino = 0;
72✔
229
  terrno = TSDB_CODE_OUT_OF_MEMORY;
72✔
230
  SSdbRow *pRow = NULL;
72✔
231
  SStbObj *pStb = NULL;
72✔
232
  bool     hasExtSchemas = false;
72✔
233

234
  int8_t sver = 0;
72✔
235
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto _OVER;
72✔
236

237
  if (sver > STB_VER_NUMBER) {
72✔
238
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
×
239
    goto _OVER;
×
240
  }
241

242
  pRow = sdbAllocRow(sizeof(SStbObj));
72✔
243
  if (pRow == NULL) goto _OVER;
72✔
244

245
  pStb = sdbGetRowObj(pRow);
72✔
246
  if (pStb == NULL) goto _OVER;
72✔
247

248
  int32_t dataPos = 0;
72✔
249
  SDB_GET_BINARY(pRaw, dataPos, pStb->name, TSDB_TABLE_FNAME_LEN, _OVER)
72✔
250
  SDB_GET_BINARY(pRaw, dataPos, pStb->db, TSDB_DB_FNAME_LEN, _OVER)
72✔
251
  SDB_GET_INT64(pRaw, dataPos, &pStb->createdTime, _OVER)
72✔
252
  SDB_GET_INT64(pRaw, dataPos, &pStb->updateTime, _OVER)
72✔
253
  SDB_GET_INT64(pRaw, dataPos, &pStb->uid, _OVER)
72✔
254
  SDB_GET_INT64(pRaw, dataPos, &pStb->dbUid, _OVER)
72✔
255
  SDB_GET_INT32(pRaw, dataPos, &pStb->tagVer, _OVER)
72✔
256
  SDB_GET_INT32(pRaw, dataPos, &pStb->colVer, _OVER)
72✔
257
  SDB_GET_INT32(pRaw, dataPos, &pStb->smaVer, _OVER)
72✔
258
  SDB_GET_INT32(pRaw, dataPos, &pStb->nextColId, _OVER)
72✔
259
  SDB_GET_INT64(pRaw, dataPos, &pStb->maxdelay[0], _OVER)
72✔
260
  SDB_GET_INT64(pRaw, dataPos, &pStb->maxdelay[1], _OVER)
72✔
261
  SDB_GET_INT64(pRaw, dataPos, &pStb->watermark[0], _OVER)
72✔
262
  SDB_GET_INT64(pRaw, dataPos, &pStb->watermark[1], _OVER)
72✔
263
  SDB_GET_INT32(pRaw, dataPos, &pStb->ttl, _OVER)
72✔
264
  SDB_GET_INT32(pRaw, dataPos, &pStb->numOfColumns, _OVER)
72✔
265
  SDB_GET_INT32(pRaw, dataPos, &pStb->numOfTags, _OVER)
72✔
266
  SDB_GET_INT32(pRaw, dataPos, &pStb->numOfFuncs, _OVER)
72✔
267
  SDB_GET_INT32(pRaw, dataPos, &pStb->commentLen, _OVER)
72✔
268
  SDB_GET_INT32(pRaw, dataPos, &pStb->ast1Len, _OVER)
72✔
269
  SDB_GET_INT32(pRaw, dataPos, &pStb->ast2Len, _OVER)
72✔
270

271
  pStb->pColumns = taosMemoryCalloc(pStb->numOfColumns, sizeof(SSchema));
72✔
272
  pStb->pTags = taosMemoryCalloc(pStb->numOfTags, sizeof(SSchema));
72✔
273
  pStb->pFuncs = taosArrayInit(pStb->numOfFuncs, TSDB_FUNC_NAME_LEN);
72✔
274
  if (pStb->pColumns == NULL || pStb->pTags == NULL || pStb->pFuncs == NULL) {
72✔
275
    goto _OVER;
×
276
  }
277

278
  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
232✔
279
    SSchema *pSchema = &pStb->pColumns[i];
160✔
280
    SDB_GET_INT8(pRaw, dataPos, &pSchema->type, _OVER)
160✔
281
    SDB_GET_INT8(pRaw, dataPos, &pSchema->flags, _OVER)
160✔
282
    SDB_GET_INT16(pRaw, dataPos, &pSchema->colId, _OVER)
160✔
283
    SDB_GET_INT32(pRaw, dataPos, &pSchema->bytes, _OVER)
160✔
284
    SDB_GET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, _OVER)
160✔
285
    hasExtSchemas = hasExtSchemas || HAS_TYPE_MOD(pSchema);
160✔
286
  }
287

288
  for (int32_t i = 0; i < pStb->numOfTags; ++i) {
276✔
289
    SSchema *pSchema = &pStb->pTags[i];
204✔
290
    SDB_GET_INT8(pRaw, dataPos, &pSchema->type, _OVER)
204✔
291
    SDB_GET_INT8(pRaw, dataPos, &pSchema->flags, _OVER)
204✔
292
    SDB_GET_INT16(pRaw, dataPos, &pSchema->colId, _OVER)
204✔
293
    SDB_GET_INT32(pRaw, dataPos, &pSchema->bytes, _OVER)
204✔
294
    SDB_GET_BINARY(pRaw, dataPos, pSchema->name, TSDB_COL_NAME_LEN, _OVER)
204✔
295
  }
296

297
  for (int32_t i = 0; i < pStb->numOfFuncs; ++i) {
72✔
298
    char funcName[TSDB_FUNC_NAME_LEN] = {0};
×
299
    SDB_GET_BINARY(pRaw, dataPos, funcName, TSDB_FUNC_NAME_LEN, _OVER)
×
300
    if (taosArrayPush(pStb->pFuncs, funcName) == NULL) goto _OVER;
×
301
  }
302

303
  if (pStb->commentLen > 0) {
72✔
304
    pStb->comment = taosMemoryCalloc(pStb->commentLen + 1, 1);
×
305
    if (pStb->comment == NULL) goto _OVER;
×
306
    SDB_GET_BINARY(pRaw, dataPos, pStb->comment, pStb->commentLen + 1, _OVER)
×
307
  }
308

309
  if (pStb->ast1Len > 0) {
72✔
310
    pStb->pAst1 = taosMemoryCalloc(pStb->ast1Len, 1);
×
311
    if (pStb->pAst1 == NULL) goto _OVER;
×
312
    SDB_GET_BINARY(pRaw, dataPos, pStb->pAst1, pStb->ast1Len, _OVER)
×
313
  }
314

315
  if (pStb->ast2Len > 0) {
72✔
316
    pStb->pAst2 = taosMemoryCalloc(pStb->ast2Len, 1);
×
317
    if (pStb->pAst2 == NULL) goto _OVER;
×
318
    SDB_GET_BINARY(pRaw, dataPos, pStb->pAst2, pStb->ast2Len, _OVER)
×
319
  }
320

321
  pStb->pCmpr = taosMemoryCalloc(pStb->numOfColumns, sizeof(SColCmpr));
72✔
322
  if (sver < STB_VER_SUPPORT_COMP) {
72✔
323
    // compatible with old data, setup default compress value
324
    // impl later
325
    for (int i = 0; i < pStb->numOfColumns; i++) {
×
326
      SSchema  *pSchema = &pStb->pColumns[i];
×
327
      SColCmpr *pCmpr = &pStb->pCmpr[i];
×
328
      pCmpr->id = pSchema->colId;
×
329
      pCmpr->alg = createDefaultColCmprByType(pSchema->type);
×
330
    }
331
  } else {
332
    for (int i = 0; i < pStb->numOfColumns; i++) {
232✔
333
      SColCmpr *pCmpr = &pStb->pCmpr[i];
160✔
334
      SDB_GET_INT16(pRaw, dataPos, &pCmpr->id, _OVER)
160✔
335
      SDB_GET_INT32(pRaw, dataPos, (int32_t *)&pCmpr->alg, _OVER)  // compatiable
160✔
336
    }
337
  }
338
  SDB_GET_INT64(pRaw, dataPos, &pStb->keep, _OVER)
72✔
339

340
  // type mod
341
  if (hasExtSchemas) {
72✔
342
    pStb->pExtSchemas = taosMemoryCalloc(pStb->numOfColumns, sizeof(SExtSchema));
×
343
    if (!pStb->pExtSchemas) goto _OVER;
×
344
    for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
×
345
      SSchema *pSchema = &pStb->pColumns[i];
×
346
      SDB_GET_INT32(pRaw, dataPos, &pStb->pExtSchemas[i].typeMod, _OVER)
×
347
    }
348
  }
349

350
  if (sver < STB_VER_SUPPORT_VIRTUAL) {
72✔
351
    pStb->virtualStb = 0;
×
352
  } else {
353
    SDB_GET_INT8(pRaw, dataPos, &pStb->virtualStb, _OVER)
72✔
354
  }
355

356
  if (sver < STB_VER_SUPPORT_OWNER) {
72✔
357
    pStb->createUser[0] = 0;
×
358
  } else {
359
    SDB_GET_BINARY(pRaw, dataPos, pStb->createUser, TSDB_USER_LEN, _OVER)
72✔
360
    SDB_GET_INT64(pRaw, dataPos, &pStb->ownerId, _OVER)
72✔
361
  }
362

363
  SDB_GET_RESERVE(pRaw, dataPos, STB_RESERVE_SIZE, _OVER)
72✔
364

365
  terrno = 0;
72✔
366

367
_OVER:
72✔
368
  if (terrno != 0) {
72✔
369
    mError("stb:%s, failed to decode from raw:%p since %s", pStb == NULL ? "null" : pStb->name, pRaw, terrstr());
×
370
    if (pStb != NULL) {
×
371
      taosMemoryFreeClear(pStb->pColumns);
×
372
      taosMemoryFreeClear(pStb->pTags);
×
373
      taosMemoryFreeClear(pStb->comment);
×
374
      taosMemoryFree(pStb->pCmpr);
×
375
      taosMemoryFreeClear(pStb->pExtSchemas);
×
376
    }
377
    taosMemoryFreeClear(pRow);
×
378
    return NULL;
×
379
  }
380

381
  mTrace("stb:%s, decode from raw:%p, row:%p", pStb->name, pRaw, pStb);
72✔
382
  return pRow;
72✔
383
}
384

385
void mndFreeStb(SStbObj *pStb) {
90✔
386
  taosArrayDestroy(pStb->pFuncs);
90✔
387
  taosMemoryFreeClear(pStb->pColumns);
90✔
388
  taosMemoryFreeClear(pStb->pTags);
90✔
389
  taosMemoryFreeClear(pStb->comment);
90✔
390
  taosMemoryFreeClear(pStb->pAst1);
90✔
391
  taosMemoryFreeClear(pStb->pAst2);
90✔
392
  taosMemoryFreeClear(pStb->pCmpr);
90✔
393
  taosMemoryFreeClear(pStb->pExtSchemas);
90✔
394
}
90✔
395

396
static int32_t mndStbActionInsert(SSdb *pSdb, SStbObj *pStb) {
18✔
397
  mTrace("stb:%s, perform insert action, row:%p", pStb->name, pStb);
18✔
398
  return 0;
18✔
399
}
400

401
static int32_t mndStbActionDelete(SSdb *pSdb, SStbObj *pStb) {
90✔
402
  mTrace("stb:%s, perform delete action, row:%p", pStb->name, pStb);
90✔
403
  mndFreeStb(pStb);
90✔
404
  return 0;
90✔
405
}
406

407
static int32_t mndStbActionUpdate(SSdb *pSdb, SStbObj *pOld, SStbObj *pNew) {
36✔
408
  terrno = 0;
36✔
409
  mTrace("stb:%s, perform update action, old row:%p new row:%p", pOld->name, pOld, pNew);
36✔
410

411
  taosWLockLatch(&pOld->lock);
36✔
412
  int32_t numOfColumns = pOld->numOfColumns;
36✔
413
  if (pOld->numOfColumns < pNew->numOfColumns) {
36✔
414
    void *pColumns = taosMemoryMalloc(pNew->numOfColumns * sizeof(SSchema));
4✔
415
    if (pColumns == NULL) {
4✔
416
      goto END;
×
417
    }
418
    taosMemoryFree(pOld->pColumns);
4✔
419
    pOld->pColumns = pColumns;
4✔
420
  }
421

422
  if (pOld->numOfTags < pNew->numOfTags) {
36✔
423
    void *pTags = taosMemoryMalloc(pNew->numOfTags * sizeof(SSchema));
2✔
424
    if (pTags == NULL) {
2✔
425
      goto END;
×
426
    }
427
    taosMemoryFree(pOld->pTags);
2✔
428
    pOld->pTags = pTags;
2✔
429
  }
430

431
  if (pOld->commentLen < pNew->commentLen && pNew->commentLen > 0) {
36✔
432
    void *comment = taosMemoryMalloc(pNew->commentLen + 1);
×
433
    if (comment == NULL) {
×
434
      goto END;
×
435
    }
436
    taosMemoryFree(pOld->comment);
×
437
    pOld->comment = comment;
×
438
  }
439
  pOld->commentLen = pNew->commentLen;
36✔
440

441
  if (pOld->ast1Len < pNew->ast1Len) {
36✔
442
    void *pAst1 = taosMemoryMalloc(pNew->ast1Len + 1);
×
443
    if (pAst1 == NULL) {
×
444
      goto END;
×
445
    }
446
    taosMemoryFree(pOld->pAst1);
×
447
    pOld->pAst1 = pAst1;
×
448
  }
449

450
  if (pOld->ast2Len < pNew->ast2Len) {
36✔
451
    void *pAst2 = taosMemoryMalloc(pNew->ast2Len + 1);
×
452
    if (pAst2 == NULL) {
×
453
      goto END;
×
454
    }
455
    taosMemoryFree(pOld->pAst2);
×
456
    pOld->pAst2 = pAst2;
×
457
  }
458

459
  pOld->updateTime = pNew->updateTime;
36✔
460
  pOld->tagVer = pNew->tagVer;
36✔
461
  pOld->colVer = pNew->colVer;
36✔
462
  pOld->smaVer = pNew->smaVer;
36✔
463
  pOld->nextColId = pNew->nextColId;
36✔
464
  pOld->ttl = pNew->ttl;
36✔
465
  pOld->keep = pNew->keep;
36✔
466
  pOld->ownerId = pNew->ownerId;
36✔
467
  
468
  if (pNew->numOfColumns > 0) {
36✔
469
    pOld->numOfColumns = pNew->numOfColumns;
36✔
470
    memcpy(pOld->pColumns, pNew->pColumns, pOld->numOfColumns * sizeof(SSchema));
36✔
471
  }
472
  if (pNew->numOfTags > 0) {
36✔
473
    pOld->numOfTags = pNew->numOfTags;
36✔
474
    memcpy(pOld->pTags, pNew->pTags, pOld->numOfTags * sizeof(SSchema));
36✔
475
  }
476
  if (pNew->commentLen > 0) {
36✔
477
    memcpy(pOld->comment, pNew->comment, pNew->commentLen + 1);
×
478
    pOld->commentLen = pNew->commentLen;
×
479
  }
480
  if (pNew->ast1Len != 0) {
36✔
481
    memcpy(pOld->pAst1, pNew->pAst1, pNew->ast1Len);
×
482
    pOld->ast1Len = pNew->ast1Len;
×
483
  }
484
  if (pNew->ast2Len != 0) {
36✔
485
    memcpy(pOld->pAst2, pNew->pAst2, pNew->ast2Len);
×
486
    pOld->ast2Len = pNew->ast2Len;
×
487
  }
488
  if (numOfColumns < pNew->numOfColumns) {
36✔
489
    taosMemoryFree(pOld->pCmpr);
4✔
490
    pOld->pCmpr = taosMemoryCalloc(pNew->numOfColumns, sizeof(SColCmpr));
4✔
491
    if (pOld->pCmpr == NULL){
4✔
492
      goto END;
×
493
    }
494
    memcpy(pOld->pCmpr, pNew->pCmpr, pNew->numOfColumns * sizeof(SColCmpr));
4✔
495
  } else {
496
    memcpy(pOld->pCmpr, pNew->pCmpr, pNew->numOfColumns * sizeof(SColCmpr));
32✔
497
  }
498

499
  if (pNew->pExtSchemas) {
36✔
500
    taosMemoryFreeClear(pOld->pExtSchemas);
×
501
    pOld->pExtSchemas = taosMemoryCalloc(pNew->numOfColumns, sizeof(SExtSchema));
×
502
    if (pOld->pExtSchemas == NULL){
×
503
      goto END;
×
504
    }
505
    memcpy(pOld->pExtSchemas, pNew->pExtSchemas, pNew->numOfColumns * sizeof(SExtSchema));
×
506
  }
507

508
END:
36✔
509
  taosWUnLockLatch(&pOld->lock);
36✔
510
  return terrno;
36✔
511
}
512

513
SStbObj *mndAcquireStb(SMnode *pMnode, char *stbName) {
110✔
514
  SSdb    *pSdb = pMnode->pSdb;
110✔
515
  SStbObj *pStb = sdbAcquire(pSdb, SDB_STB, stbName);
110✔
516
  if (pStb == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
110✔
517
    terrno = TSDB_CODE_MND_STB_NOT_EXIST;
26✔
518
  }
519
  return pStb;
110✔
520
}
521

522
void mndReleaseStb(SMnode *pMnode, SStbObj *pStb) {
114✔
523
  SSdb *pSdb = pMnode->pSdb;
114✔
524
  sdbRelease(pSdb, pStb);
114✔
525
}
114✔
526

527
SDbObj *mndAcquireDbByStb(SMnode *pMnode, const char *stbName) {
88✔
528
  SName name = {0};
88✔
529
  if ((terrno = tNameFromString(&name, stbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) return NULL;
88✔
530

531
  char db[TSDB_TABLE_FNAME_LEN] = {0};
88✔
532
  if ((terrno = tNameGetFullDbName(&name, db)) != 0) return NULL;
88✔
533

534
  return mndAcquireDb(pMnode, db);
88✔
535
}
536

537
static FORCE_INLINE int32_t schemaExColIdCompare(const void *colId, const void *pSchema) {
538
  if (*(col_id_t *)colId < ((SSchema *)pSchema)->colId) {
539
    return -1;
540
  } else if (*(col_id_t *)colId > ((SSchema *)pSchema)->colId) {
541
    return 1;
542
  }
543
  return 0;
544
}
545

546
void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen, void *alterOriData,
68✔
547
                            int32_t alterOriDataLen) {
548
  SEncoder       encoder = {0};
68✔
549
  int32_t        contLen;
550
  SName          name = {0};
68✔
551
  SVCreateStbReq req = {0};
68✔
552

553
  if ((terrno = tNameFromString(&name, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) {
68✔
554
    goto _err;
×
555
  }
556
  char dbFName[TSDB_DB_FNAME_LEN] = {0};
68✔
557
  if ((terrno = tNameGetFullDbName(&name, dbFName)) != 0) {
68✔
558
    goto _err;
×
559
  };
560

561
  req.name = (char *)tNameGetTableName(&name);
68✔
562
  req.suid = pStb->uid;
68✔
563
  req.rollup = pStb->ast1Len > 0 ? 1 : 0;
68✔
564
  req.alterOriData = alterOriData;
68✔
565
  req.alterOriDataLen = alterOriDataLen;
68✔
566
  req.source = pStb->source;
68✔
567
  req.virtualStb = pStb->virtualStb;
68✔
568
  // todo
569
  req.schemaRow.nCols = pStb->numOfColumns;
68✔
570
  req.schemaRow.version = pStb->colVer;
68✔
571
  req.schemaRow.pSchema = pStb->pColumns;
68✔
572
  req.schemaTag.nCols = pStb->numOfTags;
68✔
573
  req.schemaTag.version = pStb->tagVer;
68✔
574
  req.schemaTag.pSchema = pStb->pTags;
68✔
575

576
  req.colCmpred = 1;
68✔
577
  SColCmprWrapper *pCmpr = &req.colCmpr;
68✔
578
  req.keep = pStb->keep;
68✔
579
  pCmpr->version = pStb->colVer;
68✔
580
  pCmpr->nCols = pStb->numOfColumns;
68✔
581

582
  req.colCmpr.pColCmpr = taosMemoryCalloc(pCmpr->nCols, sizeof(SColCmpr));
68✔
583
  for (int32_t i = 0; i < pStb->numOfColumns; i++) {
216✔
584
    SColCmpr *p = &pCmpr->pColCmpr[i];
148✔
585
    p->alg = pStb->pCmpr[i].alg;
148✔
586
    p->id = pStb->pCmpr[i].id;
148✔
587
  }
588

589
  req.pExtSchemas = pStb->pExtSchemas; // only reference to it.
68✔
590
  // get length
591
  int32_t ret = 0;
68✔
592
  tEncodeSize(tEncodeSVCreateStbReq, &req, contLen, ret);
68✔
593
  if (ret < 0) {
68✔
594
    goto _err;
×
595
  }
596

597
  contLen += sizeof(SMsgHead);
68✔
598

599
  SMsgHead *pHead = taosMemoryCalloc(1, contLen);
68✔
600
  if (pHead == NULL) {
68✔
601
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
602
    goto _err;
×
603
  }
604

605
  pHead->contLen = htonl(contLen);
68✔
606
  pHead->vgId = htonl(pVgroup->vgId);
68✔
607

608
  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
68✔
609
  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
68✔
610
  if (tEncodeSVCreateStbReq(&encoder, &req) < 0) {
68✔
611
    taosMemoryFreeClear(pHead);
×
612
    tEncoderClear(&encoder);
×
613
    goto _err;
×
614
  }
615
  tEncoderClear(&encoder);
68✔
616

617
  *pContLen = contLen;
68✔
618
  taosMemoryFreeClear(req.rsmaParam.name);
68✔
619
  taosMemoryFreeClear(req.rsmaParam.funcColIds);
68✔
620
  taosMemoryFreeClear(req.rsmaParam.funcIds);
68✔
621
  taosMemoryFreeClear(req.colCmpr.pColCmpr);
68✔
622
  return pHead;
68✔
623
_err:
×
624
  taosMemoryFreeClear(req.rsmaParam.name);
×
625
  taosMemoryFreeClear(req.rsmaParam.funcColIds);
×
626
  taosMemoryFreeClear(req.rsmaParam.funcIds);
×
627
  taosMemoryFreeClear(req.colCmpr.pColCmpr);
×
628
  return NULL;
×
629
}
630

631
static void *mndBuildVDropStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pStb, int32_t *pContLen) {
44✔
632
  SName        name = {0};
44✔
633
  SVDropStbReq req = {0};
44✔
634
  int32_t      contLen = 0;
44✔
635
  int32_t      ret = 0;
44✔
636
  SMsgHead    *pHead = NULL;
44✔
637
  SEncoder     encoder = {0};
44✔
638

639
  if ((terrno = tNameFromString(&name, pStb->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0) {
44✔
640
    return NULL;
×
641
  }
642

643
  req.name = (char *)tNameGetTableName(&name);
44✔
644
  req.suid = pStb->uid;
44✔
645

646
  tEncodeSize(tEncodeSVDropStbReq, &req, contLen, ret);
44✔
647
  if (ret < 0) return NULL;
44✔
648

649
  contLen += sizeof(SMsgHead);
44✔
650
  pHead = taosMemoryMalloc(contLen);
44✔
651
  if (pHead == NULL) {
44✔
652
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
653
    return NULL;
×
654
  }
655

656
  pHead->contLen = htonl(contLen);
44✔
657
  pHead->vgId = htonl(pVgroup->vgId);
44✔
658

659
  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
44✔
660

661
  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
44✔
662
  int32_t code = tEncodeSVDropStbReq(&encoder, &req);
44✔
663
  tEncoderClear(&encoder);
44✔
664
  if (code != 0) {
44✔
665
    terrno = code;
×
666
    return NULL;
×
667
  }
668

669
  *pContLen = contLen;
44✔
670
  return pHead;
44✔
671
}
672

673
int32_t mndCheckCreateStbReq(SMCreateStbReq *pCreate) {
22✔
674
  int32_t code = 0;
22✔
675
  if (pCreate->igExists < 0 || pCreate->igExists > 1) {
22✔
676
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
677
    TAOS_RETURN(code);
×
678
  }
679

680
  if (pCreate->virtualStb != 0 && pCreate->virtualStb != 1) {
22✔
681
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
682
    TAOS_RETURN(code);
×
683
  }
684

685
  int32_t maxColumns = pCreate->virtualStb ? TSDB_MAX_COLUMNS : TSDB_MAX_COLUMNS_NON_VIRTUAL;
22✔
686
  if (pCreate->numOfColumns < TSDB_MIN_COLUMNS || pCreate->numOfTags + pCreate->numOfColumns > maxColumns) {
22✔
687
    code = TSDB_CODE_PAR_INVALID_COLUMNS_NUM;
×
688
    TAOS_RETURN(code);
×
689
  }
690

691
  if (pCreate->numOfTags <= 0 || pCreate->numOfTags > TSDB_MAX_TAGS) {
22✔
692
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
693
    TAOS_RETURN(code);
×
694
  }
695

696
  SField *pField = taosArrayGet(pCreate->pColumns, 0);
22✔
697
  if (pField->type != TSDB_DATA_TYPE_TIMESTAMP) {
22✔
698
    code = TSDB_CODE_PAR_INVALID_FIRST_COLUMN;
×
699
    TAOS_RETURN(code);
×
700
  }
701

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

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

734
  TAOS_RETURN(code);
22✔
735
}
736

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

751
  TAOS_RETURN(code);
×
752
}
753

754
int32_t mndSetCreateStbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
18✔
755
  int32_t  code = 0;
18✔
756
  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);
18✔
757
  if (pCommitRaw == NULL) {
18✔
758
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
759
    if (terrno != 0) code = terrno;
×
760
    TAOS_RETURN(code);
×
761
  }
762
  mInfo("trans:%d, add stb to commit log", pTrans->id);
18✔
763
  if ((code = mndTransAppendCommitlog(pTrans, pCommitRaw)) != 0) {
18✔
764
    sdbFreeRaw(pCommitRaw);
×
765
    TAOS_RETURN(code);
×
766
  }
767
  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));
18✔
768

769
  TAOS_RETURN(code);
18✔
770
}
771

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

779
  while (1) {
36✔
780
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
54✔
781
    if (pIter == NULL) break;
54✔
782
    if (!mndVgroupInDb(pVgroup, pDb->uid)) {
36✔
783
      sdbRelease(pSdb, pVgroup);
×
784
      continue;
×
785
    }
786

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

796
    STransAction action = {0};
36✔
797
    action.mTraceId = pTrans->mTraceId;
36✔
798
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
36✔
799
    action.pCont = pReq;
36✔
800
    action.contLen = contLen;
36✔
801
    action.msgType = TDMT_VND_CREATE_STB;
36✔
802
    action.acceptableCode = TSDB_CODE_TDB_STB_ALREADY_EXIST;
36✔
803
    action.retryCode = TSDB_CODE_TDB_STB_NOT_EXIST;
36✔
804
    mInfo("trans:%d, add create stb to redo action", pTrans->id);
36✔
805
    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
36✔
806
      taosMemoryFree(pReq);
×
807
      sdbCancelFetch(pSdb, pIter);
×
808
      sdbRelease(pSdb, pVgroup);
×
809
      TAOS_RETURN(code);
×
810
    }
811
    sdbRelease(pSdb, pVgroup);
36✔
812
  }
813

814
  TAOS_RETURN(code);
18✔
815
}
816

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

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

829
  STransAction action = {0};
×
830
  action.mTraceId = pTrans->mTraceId;
×
831
  action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
×
832
  action.pCont = pReq;
×
833
  action.contLen = contLen;
×
834
  action.msgType = TDMT_VND_CREATE_STB;
×
835
  action.acceptableCode = TSDB_CODE_TDB_STB_ALREADY_EXIST;
×
836
  action.retryCode = TSDB_CODE_TDB_STB_NOT_EXIST;
×
837
  action.groupId = pVgroup->vgId;
×
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) {
18✔
847
  int32_t code = 0;
18✔
848
  SSdb   *pSdb = pMnode->pSdb;
18✔
849
  SVgObj *pVgroup = NULL;
18✔
850
  void   *pIter = NULL;
18✔
851

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

860
    int32_t contLen = 0;
36✔
861
    void   *pReq = mndBuildVDropStbReq(pMnode, pVgroup, pStb, &contLen);
36✔
862
    if (pReq == NULL) {
36✔
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};
36✔
870
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
36✔
871
    action.pCont = pReq;
36✔
872
    action.contLen = contLen;
36✔
873
    action.msgType = TDMT_VND_DROP_STB;
36✔
874
    action.acceptableCode = TSDB_CODE_TDB_STB_NOT_EXIST;
36✔
875
    mInfo("trans:%d, add drop stb to undo action", pTrans->id);
36✔
876
    if ((code = mndTransAppendUndoAction(pTrans, &action)) != 0) {
36✔
877
      taosMemoryFree(pReq);
×
878
      sdbCancelFetch(pSdb, pIter);
×
879
      sdbRelease(pSdb, pVgroup);
×
880
      TAOS_RETURN(code);
×
881
    }
882
    sdbRelease(pSdb, pVgroup);
36✔
883
  }
884

885
  TAOS_RETURN(code);
18✔
886
}
887

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

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

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

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

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

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

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

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

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

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

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

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

1029
  char fullIdxName[TSDB_INDEX_FNAME_LEN * 2] = {0};
18✔
1030

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

1038
  mInfo("trans:%d, used to create stb:%s", pTrans->id, pCreate->name);
18✔
1039
  TAOS_CHECK_GOTO(mndBuildStbFromReq(pMnode, &stbObj, pCreate, pDb), NULL, _OVER);
18✔
1040
  memcpy(stbObj.createUser, pOperUser->name, TSDB_USER_LEN);
18✔
1041
  stbObj.ownerId = pOperUser->uid;
18✔
1042

1043

1044
  SSchema *pSchema = &(stbObj.pTags[0]);
18✔
1045
  if (mndGenIdxNameForFirstTag(fullIdxName, pDb->name, stbObj.name, pSchema->name) < 0) {
18✔
1046
    code = terrno;
×
1047
    goto _OVER;
×
1048
  }
1049
  SSIdx idx = {0};
18✔
1050
  if (mndAcquireGlobalIdx(pMnode, fullIdxName, SDB_IDX, &idx) == 0 && idx.pIdx != NULL) {
18✔
1051
    code = TSDB_CODE_MND_TAG_INDEX_ALREADY_EXIST;
×
1052
    mndReleaseIdx(pMnode, idx.pIdx);
×
1053
    goto _OVER;
×
1054
  }
1055

1056
  SIdxObj idxObj = {0};
18✔
1057
  memcpy(idxObj.name, fullIdxName, TSDB_INDEX_FNAME_LEN);
18✔
1058
  memcpy(idxObj.stb, stbObj.name, TSDB_TABLE_FNAME_LEN);
18✔
1059
  memcpy(idxObj.db, stbObj.db, TSDB_DB_FNAME_LEN);
18✔
1060
  memcpy(idxObj.colName, pSchema->name, TSDB_COL_NAME_LEN);
18✔
1061
  memcpy(idxObj.createUser, pOperUser->name, TSDB_USER_LEN);
18✔
1062
  idxObj.ownerId = pOperUser->uid;
18✔
1063
  idxObj.createdTime = taosGetTimestampMs();
18✔
1064
  idxObj.uid = mndGenerateUid(fullIdxName, strlen(fullIdxName));
18✔
1065
  idxObj.stbUid = stbObj.uid;
18✔
1066
  idxObj.dbUid = stbObj.dbUid;
18✔
1067

1068
  TAOS_CHECK_GOTO(mndSetCreateIdxCommitLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
18✔
1069

1070
  TAOS_CHECK_GOTO(mndAddStbToTrans(pMnode, pTrans, pDb, &stbObj), NULL, _OVER);
18✔
1071
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
18✔
1072
  code = 0;
18✔
1073

1074
_OVER:
18✔
1075
  mndTransDrop(pTrans);
18✔
1076
  if (mndStbActionDelete(pMnode->pSdb, &stbObj) != 0) mError("failed to mndStbActionDelete");
18✔
1077
  TAOS_RETURN(code);
18✔
1078
}
1079

1080
typedef struct {
1081
  const char *name;
1082
  uint8_t     type;
1083
  int32_t     bytes;
1084
  uint32_t    alg;
1085
} AuditColumnDef;
1086

1087
// column is consistent with vnodePrepareRow process in vnodeSvr.c
1088
static const AuditColumnDef audit_columns[] = {
1089
    {"ts", TSDB_DATA_TYPE_TIMESTAMP, 8, 0x2000102},
1090
    {"details", TSDB_DATA_TYPE_VARCHAR, 50000 + VARSTR_HEADER_SIZE, 0xFF000302},
1091
    {"user_name", TSDB_DATA_TYPE_VARCHAR, 25 + VARSTR_HEADER_SIZE, 0xFF000302},
1092
    {"operation", TSDB_DATA_TYPE_VARCHAR, 20 + VARSTR_HEADER_SIZE, 0xFF000302},
1093
    {"db", TSDB_DATA_TYPE_VARCHAR, TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE, 0xFF000302},
1094
    {"resource", TSDB_DATA_TYPE_VARCHAR, TSDB_STREAM_NAME_LEN + VARSTR_HEADER_SIZE, 0xFF000302},
1095
    {"client_address", TSDB_DATA_TYPE_VARCHAR, AUDIT_CLIENT_ADD_LEN + VARSTR_HEADER_SIZE, 0xFF000302},
1096
    {"duration", TSDB_DATA_TYPE_DOUBLE, 8, 0x5000102},
1097
    {"affected_rows", TSDB_DATA_TYPE_UBIGINT, 8, 0x1000102}};
1098

1099
static int32_t mndBuildAuditStb(SMnode *pMnode, SStbObj *pDst, SDbObj *pDb) {
×
1100
  int32_t code = 0;
×
1101
  char   *name = AUDIT_STABLE_NAME;
×
1102
  (void)tsnprintf(pDst->name, TSDB_TABLE_FNAME_LEN, "%s.%s", pDb->name, name);
×
1103
  memcpy(pDst->db, pDb->name, TSDB_DB_FNAME_LEN);
×
1104
  pDst->createdTime = taosGetTimestampMs();
×
1105
  pDst->updateTime = pDst->createdTime;
×
1106
  pDst->uid = mndGenerateUid(pDst->name, strlen(pDst->name));
×
1107
  pDst->dbUid = pDb->uid;
×
1108
  pDst->tagVer = 1;
×
1109
  pDst->colVer = 1;
×
1110
  pDst->smaVer = 1;
×
1111
  pDst->nextColId = 1;
×
1112
  pDst->maxdelay[0] = -1;
×
1113
  pDst->maxdelay[1] = -1;
×
1114
  pDst->watermark[0] = 5000;
×
1115
  pDst->watermark[1] = 5000;
×
1116
  pDst->ttl = 0;
×
1117
  pDst->keep = -1;
×
1118
  pDst->source = 0;
×
1119
  pDst->virtualStb = 0;
×
1120
  pDst->numOfColumns = sizeof(audit_columns) / sizeof(AuditColumnDef);
×
1121
  pDst->numOfTags = 1;
×
1122
  pDst->numOfFuncs = 0;
×
1123
  pDst->commentLen = -1;
×
1124
  pDst->pFuncs = NULL;
×
1125

1126
  pDst->ast1Len = 0;
×
1127
  pDst->ast2Len = 0;
×
1128

1129
  pDst->pColumns = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SSchema));
×
1130
  pDst->pTags = taosMemoryCalloc(1, pDst->numOfTags * sizeof(SSchema));
×
1131
  pDst->pCmpr = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SCmprObj));
×
1132
  if (pDst->pColumns == NULL || pDst->pTags == NULL || pDst->pCmpr == NULL) {
×
1133
    code = terrno;
×
1134
    TAOS_RETURN(code);
×
1135
  }
1136

1137
  if (pDst->nextColId < 0 || pDst->nextColId >= 0x7fff - pDst->numOfColumns - pDst->numOfTags) {
×
1138
    code = TSDB_CODE_OUT_OF_RANGE;
×
1139
    TAOS_RETURN(code);
×
1140
  }
1141

1142
  SSchema *pSchema = NULL;
×
1143
  for (int32_t i = 0; i < sizeof(audit_columns) / sizeof(AuditColumnDef); ++i) {
×
1144
    pSchema = &pDst->pColumns[pDst->nextColId - 1];
×
1145
    pSchema->type = audit_columns[i].type;
×
1146
    pSchema->bytes = audit_columns[i].bytes;
×
1147
    pSchema->flags = 1;
×
1148
    tstrncpy(pSchema->name, audit_columns[i].name, TSDB_COL_NAME_LEN);
×
1149
    pSchema->colId = pDst->nextColId;
×
1150
    // hasTypeMods = hasTypeMods || HAS_TYPE_MOD(pSchema);
1151
    SColCmpr *pColCmpr = &pDst->pCmpr[pDst->nextColId - 1];
×
1152
    pColCmpr->id = pSchema->colId;
×
1153
    pColCmpr->alg = audit_columns[i].alg;
×
1154
    pDst->nextColId++;
×
1155
  }
1156

1157
  // tag
1158
  pSchema = &pDst->pTags[0];
×
1159
  pSchema->type = TSDB_DATA_TYPE_VARCHAR;
×
1160
  pSchema->bytes = 64 + VARSTR_HEADER_SIZE;
×
1161
  SSCHMEA_SET_IDX_ON(pSchema);
×
1162
  tstrncpy(pSchema->name, "cluster_id", TSDB_COL_NAME_LEN);
×
1163
  pSchema->colId = pDst->nextColId;
×
1164
  pDst->nextColId++;
×
1165

1166
  /*
1167
    if (hasTypeMods) {
1168
      pDst->pExtSchemas = taosMemoryCalloc(pDst->numOfColumns, sizeof(SExtSchema));
1169
      if (!pDst->pExtSchemas) {
1170
        code = terrno;
1171
        TAOS_RETURN(code);
1172
      }
1173
      for (int32_t i = 0; i < pDst->numOfColumns; ++i) {
1174
        SFieldWithOptions *pField = taosArrayGet(pCreate->pColumns, i);
1175
        pDst->pExtSchemas[i].typeMod = pField->typeMod;
1176
      }
1177
    }
1178
  */
1179
  TAOS_RETURN(code);
×
1180
}
1181

1182
static int32_t mndSetCreateAuditStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb,
×
1183
                                               SVgObj *pVgroup) {
1184
  int32_t code = 0;
×
1185
  SSdb   *pSdb = pMnode->pSdb;
×
1186

1187
  int32_t contLen;
1188

1189
  if (pVgroup == NULL) {
×
1190
    code = TSDB_CODE_INVALID_PARA;
×
1191
    TAOS_RETURN(code);
×
1192
  }
1193

1194
  void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, NULL, 0);
×
1195
  if (pReq == NULL) {
×
1196
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
1197
    if (terrno != 0) code = terrno;
×
1198
    TAOS_RETURN(code);
×
1199
  }
1200

1201
  STransAction action = {0};
×
1202
  action.mTraceId = pTrans->mTraceId;
×
1203
  action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
×
1204
  action.pCont = pReq;
×
1205
  action.contLen = contLen;
×
1206
  action.msgType = TDMT_VND_CREATE_STB;
×
1207
  action.acceptableCode = TSDB_CODE_TDB_STB_ALREADY_EXIST;
×
1208
  action.retryCode = TSDB_CODE_TDB_STB_NOT_EXIST;
×
1209
  mInfo("trans:%d, add create stb to redo action", pTrans->id);
×
1210
  if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
×
1211
    taosMemoryFree(pReq);
×
1212
    TAOS_RETURN(code);
×
1213
  }
1214

1215
  TAOS_RETURN(code);
×
1216
}
1217

1218
// Note: pVgroup is expected to point to the first element of a vgroup array (e.g. pVgroups at the call site).
1219
// Only the first element is used for creating the audit super table.
1220
int32_t mndCreateAuditStb(SMnode *pMnode, SDbObj *pDb, SUserObj *pOperUser, STrans *pTrans, SVgObj *pVgroup) {
×
1221
  SStbObj stbObj = {0};
×
1222
  int32_t code = -1;
×
1223

1224
  char fullIdxName[TSDB_INDEX_FNAME_LEN * 2] = {0};
×
1225

1226
  TAOS_CHECK_GOTO(mndBuildAuditStb(pMnode, &stbObj, pDb), NULL, _OVER);
×
1227
  memcpy(stbObj.createUser, pOperUser->name, TSDB_USER_LEN);
×
1228
  stbObj.ownerId = pOperUser->uid;
×
1229

1230
  SSchema *pSchema = &(stbObj.pTags[0]);
×
1231
  if (mndGenIdxNameForFirstTag(fullIdxName, pDb->name, stbObj.name, pSchema->name) < 0) {
×
1232
    code = terrno;
×
1233
    goto _OVER;
×
1234
  }
1235

1236
  SSIdx idx = {0};
×
1237
  if (mndAcquireGlobalIdx(pMnode, fullIdxName, SDB_IDX, &idx) == 0 && idx.pIdx != NULL) {
×
1238
    code = TSDB_CODE_MND_TAG_INDEX_ALREADY_EXIST;
×
1239
    mndReleaseIdx(pMnode, idx.pIdx);
×
1240
    goto _OVER;
×
1241
  }
1242

1243
  SIdxObj idxObj = {0};
×
1244
  memcpy(idxObj.name, fullIdxName, TSDB_INDEX_FNAME_LEN);
×
1245
  memcpy(idxObj.stb, stbObj.name, TSDB_TABLE_FNAME_LEN);
×
1246
  memcpy(idxObj.db, stbObj.db, TSDB_DB_FNAME_LEN);
×
1247
  memcpy(idxObj.colName, pSchema->name, TSDB_COL_NAME_LEN);
×
1248
  memcpy(idxObj.createUser, pOperUser->name, TSDB_USER_LEN);
×
1249
  idxObj.ownerId = pOperUser->uid;
×
1250
  idxObj.createdTime = taosGetTimestampMs();
×
1251
  idxObj.uid = mndGenerateUid(fullIdxName, strlen(fullIdxName));
×
1252
  idxObj.stbUid = stbObj.uid;
×
1253
  idxObj.dbUid = stbObj.dbUid;
×
1254

1255
  mndTransSetDbName(pTrans, pDb->name, stbObj.name);
×
1256
  TAOS_CHECK_RETURN(mndTransCheckConflict(pMnode, pTrans));
×
1257
  TAOS_CHECK_GOTO(mndSetCreateIdxCommitLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
×
1258
  TAOS_CHECK_GOTO(mndSetCreateStbCommitLogs(pMnode, pTrans, pDb, &stbObj), NULL, _OVER);
×
1259
  TAOS_CHECK_GOTO(mndSetCreateAuditStbRedoActions(pMnode, pTrans, pDb, &stbObj, pVgroup), NULL, _OVER);
×
1260

1261
  code = 0;
×
1262

1263
_OVER:
×
1264
  if (mndStbActionDelete(pMnode->pSdb, &stbObj) != 0) mError("failed to mndStbActionDelete");
×
1265
  TAOS_RETURN(code);
×
1266
}
1267

1268
static int32_t mndProcessAuditRecordRsp(SRpcMsg *pRsp) {
×
1269
  int32_t code = 0;
×
1270

1271
  if (pRsp == NULL) {
×
1272
    mError("audit record rsp, null response message");
×
1273
    return -1;
×
1274
  }
1275

1276
  if (pRsp->code != 0) {
×
1277
    mError("audit record rsp failed, code:%d", pRsp->code);
×
1278
    return pRsp->code;
×
1279
  }
1280

1281
  SMnode *pMnode = pRsp->info.node;
×
1282
  SSdb   *pSdb = pMnode->pSdb;
×
1283
  (void)pMnode;  // currently unused, kept for potential future use
1284
  (void)pSdb;    // currently unused, kept for potential future use
1285

1286
  mDebug("audit record rsp succeeded, code:%d", pRsp->code);
×
1287

1288
  // no need to implement this rsp, since we do not care about the result of audit record insertion
1289

1290
  return code;
×
1291
}
1292

1293
int32_t mndAddStbToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
18✔
1294
  mndTransSetDbName(pTrans, pDb->name, pStb->name);
18✔
1295
  TAOS_CHECK_RETURN(mndTransCheckConflict(pMnode, pTrans));
18✔
1296
  TAOS_CHECK_RETURN(mndSetCreateStbCommitLogs(pMnode, pTrans, pDb, pStb));
18✔
1297
  TAOS_CHECK_RETURN(mndSetCreateStbRedoActions(pMnode, pTrans, pDb, pStb));
18✔
1298
  TAOS_CHECK_RETURN(mndSetCreateStbUndoActions(pMnode, pTrans, pDb, pStb));
18✔
1299
  return 0;
18✔
1300
}
1301

1302
static int32_t mndProcessTtlTimer(SRpcMsg *pReq) {
6✔
1303
  SMnode           *pMnode = pReq->info.node;
6✔
1304
  SSdb             *pSdb = pMnode->pSdb;
6✔
1305
  SVgObj           *pVgroup = NULL;
6✔
1306
  void             *pIter = NULL;
6✔
1307
  SVDropTtlTableReq ttlReq = {
6✔
1308
      .timestampSec = taosGetTimestampSec(), .ttlDropMaxCount = tsTtlBatchDropNum, .nUids = 0, .pTbUids = NULL};
6✔
1309
  int32_t reqLen = tSerializeSVDropTtlTableReq(NULL, 0, &ttlReq);
6✔
1310
  int32_t contLen = reqLen + sizeof(SMsgHead);
6✔
1311

1312
  mDebug("start to process ttl timer");
6✔
1313

1314
  while (1) {
2✔
1315
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
8✔
1316
    if (pIter == NULL) break;
8✔
1317

1318
    if (pVgroup->mountVgId) {
2✔
1319
      sdbRelease(pSdb, pVgroup);
×
1320
      continue;
×
1321
    }
1322

1323
    int32_t   code = 0;
2✔
1324
    SMsgHead *pHead = rpcMallocCont(contLen);
2✔
1325
    if (pHead == NULL) {
2✔
1326
      sdbRelease(pSdb, pVgroup);
×
1327
      continue;
×
1328
    }
1329
    pHead->contLen = htonl(contLen);
2✔
1330
    pHead->vgId = htonl(pVgroup->vgId);
2✔
1331
    if ((code = tSerializeSVDropTtlTableReq((char *)pHead + sizeof(SMsgHead), reqLen, &ttlReq)) < 0) {
2✔
1332
      mError("vgId:%d, failed to serialize drop ttl table request since %s", pVgroup->vgId, tstrerror(code));
×
1333
      sdbRelease(pSdb, pVgroup);
×
1334
      continue;
×
1335
    }
1336

1337
    SRpcMsg rpcMsg = {
2✔
1338
        .msgType = TDMT_VND_FETCH_TTL_EXPIRED_TBS, .pCont = pHead, .contLen = contLen, .info = pReq->info};
1339
    SEpSet epSet = mndGetVgroupEpset(pMnode, pVgroup);
2✔
1340
    code = tmsgSendReq(&epSet, &rpcMsg);
2✔
1341
    if (code != 0) {
2✔
1342
      mError("vgId:%d, failed to send drop ttl table request to vnode since 0x%x", pVgroup->vgId, code);
×
1343
    } else {
1344
      mDebug("vgId:%d, send drop ttl table request to vnode, time:%" PRId32, pVgroup->vgId, ttlReq.timestampSec);
2✔
1345
    }
1346
    sdbRelease(pSdb, pVgroup);
2✔
1347
  }
1348

1349
  return 0;
6✔
1350
}
1351

1352
#if 0
1353
static int32_t mndProcessTrimDbTimer(SRpcMsg *pReq) {
1354
  SMnode     *pMnode = pReq->info.node;
1355
  SSdb       *pSdb = pMnode->pSdb;
1356
  SVgObj     *pVgroup = NULL;
1357
  void       *pIter = NULL;
1358
  SVTrimDbReq trimReq = {0};
1359
  trimReq.compactStartTime = taosGetTimestampMs();
1360
  trimReq.tw.skey = INT64_MIN;
1361
  trimReq.tw.ekey = trimReq.compactStartTime;
1362
  trimReq.compactId = 0;  // TODO: use the real value
1363
  trimReq.metaOnly = 0;
1364
  trimReq.triggerType = TSDB_TRIGGER_AUTO;
1365
  int32_t reqLen = tSerializeSVTrimDbReq(NULL, 0, &trimReq);
1366
  int32_t contLen = reqLen + sizeof(SMsgHead);
1367

1368
  while (1) {
1369
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
1370
    if (pIter == NULL) break;
1371
    if (pVgroup->mountVgId) {
1372
      sdbRelease(pSdb, pVgroup);
1373
      continue;
1374
    }
1375

1376
    int32_t code = 0;
1377

1378
    SMsgHead *pHead = rpcMallocCont(contLen);
1379
    if (pHead == NULL) {
1380
      sdbCancelFetch(pSdb, pVgroup);
1381
      sdbRelease(pSdb, pVgroup);
1382
      continue;
1383
    }
1384
    pHead->contLen = htonl(contLen);
1385
    pHead->vgId = htonl(pVgroup->vgId);
1386
    trimReq.dbUid = pVgroup->dbUid;
1387
    (void)snprintf(trimReq.db, sizeof(trimReq.db), "%s", pVgroup->dbName);
1388
    if ((code = tSerializeSVTrimDbReq((char *)pHead + sizeof(SMsgHead), reqLen, &trimReq)) < 0) {
1389
      mError("vgId:%d, failed to serialize trim db request since %s", pVgroup->vgId, tstrerror(code));
1390
    }
1391

1392
    SRpcMsg rpcMsg = {.msgType = TDMT_VND_TRIM, .pCont = pHead, .contLen = contLen};
1393
    SEpSet  epSet = mndGetVgroupEpset(pMnode, pVgroup);
1394
    code = tmsgSendReq(&epSet, &rpcMsg);
1395
    if (code != 0) {
1396
      mError("vgId:%d, timer failed to send vnode-trim request to vnode since 0x%x", pVgroup->vgId, code);
1397
    } else {
1398
      mInfo("vgId:%d, timer send vnode-trim request to vnode, time:%" PRIi64 " ms", pVgroup->vgId, trimReq.tw.ekey);
1399
    }
1400
    sdbRelease(pSdb, pVgroup);
1401
  }
1402

1403
  return 0;
1404
}
1405
#endif
1406

1407
static int32_t mndFindSuperTableTagIndex(const SStbObj *pStb, const char *tagName) {
42✔
1408
  for (int32_t tag = 0; tag < pStb->numOfTags; tag++) {
136✔
1409
    if (strcmp(pStb->pTags[tag].name, tagName) == 0) {
118✔
1410
      return tag;
24✔
1411
    }
1412
  }
1413

1414
  return -1;
18✔
1415
}
1416

1417
static int32_t mndFindSuperTableColumnIndex(const SStbObj *pStb, const char *colName) {
38✔
1418
  for (int32_t col = 0; col < pStb->numOfColumns; col++) {
90✔
1419
    if (strcmp(pStb->pColumns[col].name, colName) == 0) {
72✔
1420
      return col;
20✔
1421
    }
1422
  }
1423

1424
  return -1;
18✔
1425
}
1426

1427
static bool mndValidateSchema(SSchema *pSchemas, int32_t nSchema, SArray *pFields, int32_t maxLen) {
14✔
1428
  int32_t rowLen = 0;
14✔
1429
  for (int32_t i = 0; i < nSchema; ++i) {
48✔
1430
    rowLen += (pSchemas + i)->bytes;
34✔
1431
  }
1432

1433
  int32_t nField = taosArrayGetSize(pFields);
14✔
1434
  for (int32_t i = 0; i < nField; ++i) {
28✔
1435
    rowLen += ((SField *)TARRAY_GET_ELEM(pFields, i))->bytes;
14✔
1436
  }
1437

1438
  return rowLen <= maxLen;
14✔
1439
}
1440

1441
static int32_t mndBuildStbFromAlter(SStbObj *pStb, SStbObj *pDst, SMCreateStbReq *createReq) {
×
1442
  int32_t code = 0;
×
1443
  taosRLockLatch(&pStb->lock);
×
1444
  memcpy(pDst, pStb, sizeof(SStbObj));
×
1445
  taosRUnLockLatch(&pStb->lock);
×
1446

1447
  pDst->source = createReq->source;
×
1448
  pDst->updateTime = taosGetTimestampMs();
×
1449
  pDst->numOfColumns = createReq->numOfColumns;
×
1450
  pDst->numOfTags = createReq->numOfTags;
×
1451
  pDst->pColumns = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SSchema));
×
1452
  pDst->pTags = taosMemoryCalloc(1, pDst->numOfTags * sizeof(SSchema));
×
1453
  pDst->pCmpr = taosMemoryCalloc(1, pDst->numOfColumns * sizeof(SColCmpr));
×
1454
  pDst->pExtSchemas = taosMemoryCalloc(pDst->numOfColumns, sizeof(SExtSchema));
×
1455

1456
  if (pDst->pColumns == NULL || pDst->pTags == NULL || pDst->pCmpr == NULL || pDst->pExtSchemas == NULL) {
×
1457
    code = terrno;
×
1458
    TAOS_RETURN(code);
×
1459
  }
1460

1461
  if (pDst->nextColId < 0 || pDst->nextColId >= 0x7fff - pDst->numOfColumns - pDst->numOfTags) {
×
1462
    code = TSDB_CODE_OUT_OF_RANGE;
×
1463
    TAOS_RETURN(code);
×
1464
  }
1465

1466
  for (int32_t i = 0; i < pDst->numOfColumns; ++i) {
×
1467
    SFieldWithOptions *pField = taosArrayGet(createReq->pColumns, i);
×
1468
    SSchema           *pSchema = &pDst->pColumns[i];
×
1469
    pSchema->type = pField->type;
×
1470
    pSchema->bytes = pField->bytes;
×
1471
    pSchema->flags = pField->flags;
×
1472
    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
×
1473
    int32_t cIndex = mndFindSuperTableColumnIndex(pStb, pField->name);
×
1474
    if (cIndex >= 0) {
×
1475
      pSchema->colId = pStb->pColumns[cIndex].colId;
×
1476
    } else {
1477
      pSchema->colId = pDst->nextColId++;
×
1478
    }
1479
  }
1480

1481
  for (int32_t i = 0; i < pDst->numOfTags; ++i) {
×
1482
    SField  *pField = taosArrayGet(createReq->pTags, i);
×
1483
    SSchema *pSchema = &pDst->pTags[i];
×
1484
    pSchema->type = pField->type;
×
1485
    pSchema->bytes = pField->bytes;
×
1486
    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
×
1487
    int32_t cIndex = mndFindSuperTableTagIndex(pStb, pField->name);
×
1488
    if (cIndex >= 0) {
×
1489
      pSchema->colId = pStb->pTags[cIndex].colId;
×
1490
    } else {
1491
      pSchema->colId = pDst->nextColId++;
×
1492
    }
1493
  }
1494
  for (int32_t i = 0; i < pDst->numOfColumns; i++) {
×
1495
    SColCmpr          *p = pDst->pCmpr + i;
×
1496
    SFieldWithOptions *pField = taosArrayGet(createReq->pColumns, i);
×
1497
    SSchema           *pSchema = &pDst->pColumns[i];
×
1498
    p->id = pSchema->colId;
×
1499
    if (pField->compress == 0) {
×
1500
      p->alg = createDefaultColCmprByType(pSchema->type);
×
1501
    } else {
1502
      p->alg = pField->compress;
×
1503
    }
1504
    if (pField->flags & COL_HAS_TYPE_MOD) {
×
1505
      pDst->pExtSchemas[i].typeMod = pField->typeMod;
×
1506
    }
1507
  }
1508
  pDst->tagVer = createReq->tagVer;
×
1509
  pDst->colVer = createReq->colVer;
×
1510
  return TSDB_CODE_SUCCESS;
×
1511
}
1512

1513
// used for tmq_get_json_meta to build alter msg
1514
static void buildAlterMsg(SStbObj *pStb, SStbObj *pDst, void** pAlterBuf, int32_t* len){
×
1515
  SMAlterStbReq alterReq = {0};
×
1516
  alterReq.pFields = taosArrayInit(2, sizeof(SField));
×
1517
  if (NULL == alterReq.pFields) {
×
1518
    mError("failed to init alter fields array");
×
1519
    goto END;
×
1520
  }
1521
  tstrncpy(alterReq.name, pStb->name, TSDB_TABLE_FNAME_LEN);
×
1522
  for (int32_t i = 0; i < pDst->numOfColumns && taosArrayGetSize(alterReq.pFields) == 0; ++i) {
×
1523
    SSchema           *pSchema = &pDst->pColumns[i];
×
1524
    int32_t cIndex = mndFindSuperTableColumnIndex(pStb, pSchema->name);
×
1525
    if (cIndex >= 0 && pSchema->bytes == pStb->pColumns[cIndex].bytes) {
×
1526
      continue;
×
1527
    }
1528
    if (cIndex < 0) {
×
1529
      alterReq.alterType = TSDB_ALTER_TABLE_ADD_COLUMN;
×
1530
    } else if (pSchema->bytes > pStb->pColumns[cIndex].bytes){
×
1531
      alterReq.alterType = TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES;
×
1532
    }
1533
    SField *pAlterField = taosArrayReserve(alterReq.pFields, 1);
×
1534
    pAlterField->type = pSchema->type;
×
1535
    pAlterField->bytes = pSchema->bytes;
×
1536
    tstrncpy(pAlterField->name, pSchema->name, TSDB_COL_NAME_LEN);
×
1537
    mDebug("alter column name:%s, type:%d, bytes:%d", pAlterField->name, pAlterField->type, pAlterField->bytes);
×
1538
  }
1539

1540
  for (int32_t i = 0; i < pDst->numOfTags && taosArrayGetSize(alterReq.pFields) == 0; ++i) {
×
1541
    SSchema *pSchema = &pDst->pTags[i];
×
1542
    int32_t cIndex = mndFindSuperTableTagIndex(pStb, pSchema->name);
×
1543
    if (cIndex >= 0 && pSchema->bytes == pStb->pTags[cIndex].bytes) {
×
1544
      continue;
×
1545
    }
1546
    if (cIndex < 0) {
×
1547
      alterReq.alterType = TSDB_ALTER_TABLE_ADD_TAG;
×
1548
    } else if (pSchema->bytes > pStb->pTags[cIndex].bytes){
×
1549
      alterReq.alterType = TSDB_ALTER_TABLE_UPDATE_TAG_BYTES;
×
1550
    }
1551
    SField *pAlterField = taosArrayReserve(alterReq.pFields, 1);
×
1552
    pAlterField->type = pSchema->type;
×
1553
    pAlterField->bytes = pSchema->bytes;
×
1554
    tstrncpy(pAlterField->name, pSchema->name, TSDB_COL_NAME_LEN);
×
1555
    mDebug("alter tag name:%s, type:%d, bytes:%d", pAlterField->name, pAlterField->type, pAlterField->bytes);
×
1556
  }
1557
  alterReq.numOfFields = taosArrayGetSize(alterReq.pFields);
×
1558
  if (alterReq.numOfFields == 0) {
×
1559
    mError("no valid alter field found");
×
1560
    goto END;
×
1561
  }
1562

1563
  int32_t contLen = tSerializeSMAlterStbReq(NULL, 0, &alterReq);
×
1564
  if (contLen <= 0) {
×
1565
    mError("failed to get alter stb req len");
×
1566
    goto END;
×
1567
  }
1568
  void*   buf = taosMemoryMalloc(contLen);
×
1569
  if (buf == NULL) {
×
1570
    mError("failed to malloc alter stb req buf");
×
1571
    goto END;
×
1572
  }
1573
  int32_t code = tSerializeSMAlterStbReq(buf, contLen, &alterReq);
×
1574
  if (code <= TSDB_CODE_SUCCESS) {
×
1575
    mError("failed to serialize alter stb req %d", code);
×
1576
    taosMemoryFreeClear(buf);
×
1577
    goto END;
×
1578
  }
1579
  *pAlterBuf = buf;
×
1580
  *len = contLen;
×
1581
END:
×
1582
  taosArrayDestroy(alterReq.pFields);
×
1583
}
×
1584

1585
static int32_t mndProcessCreateStbReq(SRpcMsg *pReq) {
22✔
1586
  SMnode        *pMnode = pReq->info.node;
22✔
1587
  int32_t        code = -1;
22✔
1588
  SStbObj       *pStb = NULL;
22✔
1589
  SDbObj        *pDb = NULL;
22✔
1590
  SUserObj      *pOperUser = NULL;
22✔
1591
  SMCreateStbReq createReq = {0};
22✔
1592
  bool           isAlter = false;
22✔
1593
  SHashObj      *pHash = NULL;
22✔
1594
  int64_t        tss = taosGetTimestampMs();
22✔
1595

1596
  if (tDeserializeSMCreateStbReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
22✔
1597
    code = TSDB_CODE_INVALID_MSG;
×
1598
    goto _OVER;
×
1599
  }
1600

1601
  mInfo("stb:%s, start to create", createReq.name);
22✔
1602
  if (mndCheckCreateStbReq(&createReq) != 0) {
22✔
1603
    code = TSDB_CODE_INVALID_MSG;
×
1604
    goto _OVER;
×
1605
  }
1606

1607
  pStb = mndAcquireStb(pMnode, createReq.name);
22✔
1608
  if (pStb != NULL) {
22✔
1609
    if (createReq.igExists) {
2✔
1610
      if (createReq.source == TD_REQ_FROM_APP) {
×
1611
        mInfo("stb:%s, already exist, ignore exist is set", createReq.name);
×
1612
        code = 0;
×
1613
        goto _OVER;
×
1614
      } else if (pStb->uid != createReq.suid) {
×
1615
        mInfo("stb:%s, alter table does not need to be done, because table is deleted", createReq.name);
×
1616
        code = 0;
×
1617
        goto _OVER;
×
1618
      } else if (createReq.tagVer > 0 || createReq.colVer > 0) {
×
1619
        int32_t tagDelta = createReq.tagVer - pStb->tagVer;
×
1620
        int32_t colDelta = createReq.colVer - pStb->colVer;
×
1621
        mInfo("stb:%s, already exist while create, input tagVer:%d colVer:%d, exist tagVer:%d colVer:%d",
×
1622
              createReq.name, createReq.tagVer, createReq.colVer, pStb->tagVer, pStb->colVer);
1623
        if (tagDelta <= 0 && colDelta <= 0) {
×
1624
          mInfo("stb:%s, schema version is not incremented and nothing needs to be done", createReq.name);
×
1625
          code = 0;
×
1626
          goto _OVER;
×
1627
        } else if ((tagDelta == 1 && colDelta == 0) || (tagDelta == 0 && colDelta == 1) ||
×
1628
                   (pStb->colVer == 1 && createReq.colVer > 1) || (pStb->tagVer == 1 && createReq.tagVer > 1)) {
×
1629
          isAlter = true;
×
1630
          mInfo("stb:%s, schema version is only increased by 1 number, do alter operation", createReq.name);
×
1631
        } else {
1632
          mError("stb:%s, schema version increase more than 1 number, error is returned", createReq.name);
×
1633
          code = TSDB_CODE_MND_INVALID_SCHEMA_VER;
×
1634
          goto _OVER;
×
1635
        }
1636
      } else {
1637
        mError("stb:%s, already exist while create, input tagVer:%d colVer:%d is invalid, origin tagVer:%d colVer:%d",
×
1638
               createReq.name, createReq.tagVer, createReq.colVer, pStb->tagVer, pStb->colVer);
1639
        code = TSDB_CODE_MND_INVALID_SCHEMA_VER;
×
1640
        goto _OVER;
×
1641
      }
1642
    } else {
1643
      code = TSDB_CODE_MND_STB_ALREADY_EXIST;
2✔
1644
      goto _OVER;
2✔
1645
    }
1646
  } else if (terrno != TSDB_CODE_MND_STB_NOT_EXIST) {
20✔
1647
    goto _OVER;
×
1648
  } else if ((createReq.source == TD_REQ_FROM_TAOX_OLD || createReq.source == TD_REQ_FROM_TAOX || createReq.source == TD_REQ_FROM_SML) &&
20✔
1649
             (createReq.tagVer != 1 || createReq.colVer != 1)) {
×
1650
    mInfo("stb:%s, alter table does not need to be done, because table is deleted", createReq.name);
×
1651
    code = 0;
×
1652
    goto _OVER;
×
1653
  }
1654

1655
  pHash = taosHashInit(createReq.numOfColumns + createReq.numOfTags, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY),
20✔
1656
                       false, HASH_NO_LOCK);
1657
  if (pHash == NULL) {
20✔
1658
    code = TSDB_CODE_OUT_OF_MEMORY;
×
1659
    goto _OVER;
×
1660
  }
1661

1662
  for (int32_t i = 0; i < createReq.numOfColumns; ++i) {
62✔
1663
    SFieldWithOptions *pField = taosArrayGet(createReq.pColumns, i);
42✔
1664
    if ((code = taosHashPut(pHash, pField->name, strlen(pField->name), NULL, 0)) != 0) {
42✔
1665
      if (code == TSDB_CODE_DUP_KEY) {
×
1666
        code = TSDB_CODE_TSC_DUP_COL_NAMES;
×
1667
      }
1668
      goto _OVER;
×
1669
    }
1670
  }
1671

1672
  for (int32_t i = 0; i < createReq.numOfTags; ++i) {
76✔
1673
    SField *pField = taosArrayGet(createReq.pTags, i);
58✔
1674
    if ((code = taosHashPut(pHash, pField->name, strlen(pField->name), NULL, 0)) != 0) {
58✔
1675
      if (code == TSDB_CODE_DUP_KEY) {
2✔
1676
        code = TSDB_CODE_TSC_DUP_COL_NAMES;
2✔
1677
      }
1678
      goto _OVER;
2✔
1679
    }
1680
  }
1681

1682
  pDb = mndAcquireDbByStb(pMnode, createReq.name);
18✔
1683
  if (pDb == NULL) {
18✔
1684
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
1685
    goto _OVER;
×
1686
  }
1687

1688
  if ((code = mndAcquireUser(pMnode, (RPC_MSG_USER(pReq)), &pOperUser))) {
18✔
1689
    goto _OVER;
×
1690
  }
1691

1692
  if ((code = mndCheckDbPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_USE_DB, pDb))) {
18✔
1693
    goto _OVER;
×
1694
  }
1695

1696
  if ((code =
18✔
1697
           mndCheckDbPrivilegeByNameRecF(pMnode, pOperUser, pDb->cfg.isAudit ? PRIV_AUDIT_TBL_CREATE : PRIV_TBL_CREATE,
18✔
1698
                                         pDb->cfg.isAudit ? PRIV_OBJ_CLUSTER : PRIV_OBJ_DB, pDb->name, NULL))) {
18✔
1699
    goto _OVER;
×
1700
  }
1701

1702
  if (pDb->cfg.isMount) {
18✔
1703
    code = TSDB_CODE_MND_MOUNT_OBJ_NOT_SUPPORT;
×
1704
    goto _OVER;
×
1705
  }
1706

1707
  int32_t numOfStbs = -1;
18✔
1708
  if ((code = mndGetNumOfStbs(pMnode, pDb->name, &numOfStbs)) != 0) {
18✔
1709
    goto _OVER;
×
1710
  }
1711

1712
  if (pDb->cfg.numOfStables == 1 && numOfStbs != 0) {
18✔
1713
    code = TSDB_CODE_MND_SINGLE_STB_MODE_DB;
×
1714
    goto _OVER;
×
1715
  }
1716

1717
  if ((code = grantCheck(TSDB_GRANT_STABLE)) < 0) {
18✔
1718
    goto _OVER;
×
1719
  }
1720

1721
  if (isAlter) {
18✔
1722
    bool    needRsp = false;
×
1723
    SStbObj pDst = {0};
×
1724
    if ((code = mndBuildStbFromAlter(pStb, &pDst, &createReq)) != 0) {
×
1725
      taosMemoryFreeClear(pDst.pTags);
×
1726
      taosMemoryFreeClear(pDst.pColumns);
×
1727
      taosMemoryFreeClear(pDst.pCmpr);
×
1728
      taosMemoryFreeClear(pDst.pExtSchemas);
×
1729
      goto _OVER;
×
1730
    }
1731
    void* buf = NULL;
×
1732
    int32_t contLen = 0;
×
1733
    buildAlterMsg(pStb, &pDst, &buf, &contLen);
×
1734
    code = mndAlterStbImp(pMnode, pReq, pDb, &pDst, needRsp, buf, contLen);
×
1735
    taosMemoryFree(buf);
×
1736
    taosMemoryFreeClear(pDst.pTags);
×
1737
    taosMemoryFreeClear(pDst.pColumns);
×
1738
    taosMemoryFreeClear(pDst.pCmpr);
×
1739
    taosMemoryFreeClear(pDst.pExtSchemas);
×
1740
  } else {
1741
    code = mndCreateStb(pMnode, pReq, &createReq, pDb, pOperUser);
18✔
1742
  }
1743
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
18✔
1744

1745
  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {
18✔
1746
    SName name = {0};
18✔
1747
    TAOS_CHECK_RETURN(tNameFromString(&name, createReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
18✔
1748

1749
    int64_t tse = taosGetTimestampMs();
18✔
1750
    double  duration = (double)(tse - tss);
18✔
1751
    duration = duration / 1000;
18✔
1752
    if (createReq.sql == NULL && createReq.sqlLen == 0) {
36✔
1753
      char detail[1000] = {0};
18✔
1754

1755
      (void)tsnprintf(detail, sizeof(detail), "dbname:%s, stable name:%s", name.dbname, name.tname);
18✔
1756

1757
      auditRecord(pReq, pMnode->clusterId, "createStb", name.dbname, name.tname, detail, strlen(detail), duration, 0);
18✔
1758
    } else {
1759
      auditRecord(pReq, pMnode->clusterId, "createStb", name.dbname, name.tname, createReq.sql, createReq.sqlLen,
×
1760
                  duration, 0);
1761
    }
1762
  }
1763
_OVER:
×
1764
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
22✔
1765
    mError("stb:%s, failed to create since %s", createReq.name, tstrerror(code));
4✔
1766
  }
1767

1768
  mndReleaseStb(pMnode, pStb);
22✔
1769
  mndReleaseDb(pMnode, pDb);
22✔
1770
  mndReleaseUser(pMnode, pOperUser);
22✔
1771
  tFreeSMCreateStbReq(&createReq);
22✔
1772

1773
  if (pHash != NULL) {
22✔
1774
    taosHashCleanup(pHash);
20✔
1775
  }
1776

1777
  TAOS_RETURN(code);
22✔
1778
}
1779

1780
static int32_t mndCheckAlterStbReq(SMAlterStbReq *pAlter) {
66✔
1781
  int32_t code = 0;
66✔
1782
  if (pAlter->commentLen >= 0) return 0;
66✔
1783
  if (pAlter->ttl != 0) return 0;
×
1784
  if (pAlter->keep != -1) return 0;
×
1785

1786
  if (pAlter->numOfFields < 1 || pAlter->numOfFields != (int32_t)taosArrayGetSize(pAlter->pFields)) {
×
1787
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
1788
    TAOS_RETURN(code);
×
1789
  }
1790

1791
  for (int32_t i = 0; i < pAlter->numOfFields; ++i) {
×
1792
    SField *pField = taosArrayGet(pAlter->pFields, i);
×
1793
    if (pField->name[0] == 0) {
×
1794
      code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
1795
      TAOS_RETURN(code);
×
1796
    }
1797
  }
1798

1799
  TAOS_RETURN(code);
×
1800
}
1801

1802
int32_t mndAllocStbSchemas(const SStbObj *pOld, SStbObj *pNew) {
32✔
1803
  pNew->pTags = taosMemoryCalloc(pNew->numOfTags, sizeof(SSchema));
32✔
1804
  pNew->pColumns = taosMemoryCalloc(pNew->numOfColumns, sizeof(SSchema));
32✔
1805
  pNew->pCmpr = taosMemoryCalloc(pNew->numOfColumns, sizeof(SColCmpr));
32✔
1806
  if (pNew->pTags == NULL || pNew->pColumns == NULL || pNew->pCmpr == NULL) {
32✔
1807
    TAOS_RETURN(terrno);
×
1808
  }
1809

1810
  memcpy(pNew->pColumns, pOld->pColumns, sizeof(SSchema) * pOld->numOfColumns);
32✔
1811
  memcpy(pNew->pTags, pOld->pTags, sizeof(SSchema) * pOld->numOfTags);
32✔
1812
  memcpy(pNew->pCmpr, pOld->pCmpr, sizeof(SColCmpr) * pOld->numOfColumns);
32✔
1813
  if (pOld->pExtSchemas) {
32✔
1814
    pNew->pExtSchemas = taosMemoryCalloc(pNew->numOfColumns, sizeof(SExtSchema));
×
1815
    if (pNew->pExtSchemas == NULL) {
×
1816
      TAOS_RETURN(terrno);
×
1817
    }
1818
    memcpy(pNew->pExtSchemas, pOld->pExtSchemas, sizeof(SExtSchema) * pOld->numOfColumns);
×
1819
  }
1820

1821
  TAOS_RETURN(0);
32✔
1822
}
1823

1824
static int32_t mndUpdateTableOptions(const SStbObj *pOld, SStbObj *pNew, char *pComment, int32_t commentLen,
×
1825
                                     int32_t ttl, int64_t keep) {
1826
  int32_t code = 0;
×
1827
  if (commentLen > 0) {
×
1828
    pNew->commentLen = commentLen;
×
1829
    pNew->comment = taosMemoryCalloc(1, commentLen + 1);
×
1830
    if (pNew->comment == NULL) {
×
1831
      terrno = TSDB_CODE_OUT_OF_MEMORY;
×
1832
      return -1;
×
1833
    }
1834
    memcpy(pNew->comment, pComment, commentLen + 1);
×
1835
  } else if (commentLen == 0) {
×
1836
    pNew->commentLen = 0;
×
1837
  } else {
1838
  }
1839

1840
  if (ttl >= 0) {
×
1841
    pNew->ttl = ttl;
×
1842
  }
1843

1844
  if (keep > 0) {
×
1845
    pNew->keep = keep;
×
1846
  }
1847

1848
  if ((code = mndAllocStbSchemas(pOld, pNew)) != 0) {
×
1849
    TAOS_RETURN(code);
×
1850
  }
1851
  TAOS_RETURN(code);
×
1852
}
1853

1854
static int32_t mndAddSuperTableTag(const SStbObj *pOld, SStbObj *pNew, SArray *pFields, int32_t ntags) {
6✔
1855
  int32_t code = 0;
6✔
1856
  if (pOld->numOfTags + ntags > TSDB_MAX_TAGS) {
6✔
1857
    code = TSDB_CODE_MND_TOO_MANY_TAGS;
×
1858
    TAOS_RETURN(code);
×
1859
  }
1860

1861
  int32_t maxColumns = pOld->virtualStb ? TSDB_MAX_COLUMNS : TSDB_MAX_COLUMNS_NON_VIRTUAL;
6✔
1862
  if (pOld->numOfColumns + ntags + pOld->numOfTags > maxColumns) {
6✔
1863
    code = TSDB_CODE_MND_TOO_MANY_COLUMNS;
×
1864
    TAOS_RETURN(code);
×
1865
  }
1866

1867
  if (!mndValidateSchema(pOld->pTags, pOld->numOfTags, pFields, TSDB_MAX_TAGS_LEN)) {
6✔
1868
    code = TSDB_CODE_PAR_INVALID_TAGS_LENGTH;
×
1869
    TAOS_RETURN(code);
×
1870
  }
1871

1872
  pNew->numOfTags = pNew->numOfTags + ntags;
6✔
1873
  if ((code = mndAllocStbSchemas(pOld, pNew)) != 0) {
6✔
1874
    TAOS_RETURN(code);
×
1875
  }
1876

1877
  if (pNew->nextColId < 0 || pNew->nextColId >= 0x7fff - ntags) {
6✔
1878
    code = TSDB_CODE_OUT_OF_RANGE;
×
1879
    TAOS_RETURN(code);
×
1880
  }
1881

1882
  for (int32_t i = 0; i < ntags; i++) {
8✔
1883
    SField *pField = taosArrayGet(pFields, i);
6✔
1884
    if (mndFindSuperTableColumnIndex(pOld, pField->name) >= 0) {
6✔
1885
      code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
2✔
1886
      TAOS_RETURN(code);
2✔
1887
    }
1888

1889
    if (mndFindSuperTableTagIndex(pOld, pField->name) >= 0) {
4✔
1890
      code = TSDB_CODE_MND_TAG_ALREADY_EXIST;
2✔
1891
      TAOS_RETURN(code);
2✔
1892
    }
1893

1894
    SSchema *pSchema = &pNew->pTags[pOld->numOfTags + i];
2✔
1895
    pSchema->bytes = pField->bytes;
2✔
1896
    pSchema->type = pField->type;
2✔
1897
    memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
2✔
1898
    if (pNew->nextColId > INT16_MAX) {
2✔
1899
      code = TSDB_CODE_MND_EXCEED_MAX_COL_ID;
×
1900
      TAOS_RETURN(code);
×
1901
    }
1902
    pSchema->colId = pNew->nextColId;
2✔
1903
    pNew->nextColId++;
2✔
1904

1905
    mInfo("stb:%s, start to add tag %s", pNew->name, pSchema->name);
2✔
1906
  }
1907

1908
  pNew->tagVer++;
2✔
1909
  TAOS_RETURN(code);
2✔
1910
}
1911

1912
static int32_t mndCheckAlterColForTSma(SMnode *pMnode, const char *stbFullName, int64_t suid, col_id_t colId, bool isTag) {
24✔
1913
  int32_t code = 0;
24✔
1914
  SSdb   *pSdb = pMnode->pSdb;
24✔
1915
  void   *pIter = NULL;
24✔
1916
  while (1) {
×
1917
    SSmaObj *pSma = NULL;
24✔
1918
    pIter = sdbFetch(pSdb, SDB_SMA, pIter, (void **)&pSma);
24✔
1919
    if (pIter == NULL) break;
24✔
1920

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

1924
    if (isTag) {
×
1925
      code = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TSMA;
×
1926
      mError("tsma:%s, check tag:%d conflicted", pSma->name, colId);
×
1927
      sdbRelease(pSdb, pSma);
×
1928
      sdbCancelFetch(pSdb, pIter);
×
1929
      TAOS_RETURN(code);
×
1930
    }
1931

1932
    SNode *pAst = NULL;
×
1933
    if (nodesStringToNode(pSma->ast, &pAst) != 0) {
×
1934
      code = TSDB_CODE_SDB_INVALID_DATA_CONTENT;
×
1935
      mError("tsma:%s, check tag and column modifiable, stb:%s suid:%" PRId64 " colId:%d failed since parse AST err",
×
1936
             pSma->name, stbFullName, suid, colId);
1937
      sdbCancelFetch(pSdb, pIter);
×
1938
      TAOS_RETURN(code);
×
1939
    }
1940

1941
    SNodeList *pNodeList = NULL;
×
1942
    if ((code = nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList)) !=
×
1943
        0) {
1944
      sdbCancelFetch(pSdb, pIter);
×
1945
      TAOS_RETURN(code);
×
1946
    }
1947
    SNode *pNode = NULL;
×
1948
    FOREACH(pNode, pNodeList) {
×
1949
      SColumnNode *pCol = (SColumnNode *)pNode;
×
1950
      mInfo("tsma:%s, check colId:%d tableId:%" PRId64, pSma->name, pCol->colId, pCol->tableId);
×
1951

1952
      if ((pCol->tableId != suid) && (pSma->stbUid != suid)) {
×
1953
        mInfo("tsma:%s, check colId:%d passed", pSma->name, pCol->colId);
×
1954
        goto NEXT;
×
1955
      }
1956
      if ((pCol->colId) > 0 && (pCol->colId == colId)) {
×
1957
        code = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TSMA;
×
1958
        mError("tsma:%s, check colId:%d conflicted", pSma->name, pCol->colId);
×
1959
        nodesDestroyNode(pAst);
×
1960
        nodesDestroyList(pNodeList);
×
1961
        sdbRelease(pSdb, pSma);
×
1962
        sdbCancelFetch(pSdb, pIter);
×
1963
        TAOS_RETURN(code);
×
1964
      }
1965
      mInfo("tsma:%s, check colId:%d passed", pSma->name, pCol->colId);
×
1966
    }
1967

1968
  NEXT:
×
1969
    sdbRelease(pSdb, pSma);
×
1970
    nodesDestroyNode(pAst);
×
1971
    nodesDestroyList(pNodeList);
×
1972
  }
1973
  TAOS_RETURN(code);
24✔
1974
}
1975

1976
static int32_t mndDropSuperTableTag(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const char *tagName) {
4✔
1977
  int32_t code = 0;
4✔
1978
  int32_t tag = mndFindSuperTableTagIndex(pOld, tagName);
4✔
1979
  if (tag < 0) {
4✔
1980
    code = TSDB_CODE_MND_TAG_NOT_EXIST;
2✔
1981
    TAOS_RETURN(code);
2✔
1982
  }
1983

1984
  col_id_t colId = pOld->pTags[tag].colId;
2✔
1985
  TAOS_CHECK_RETURN(mndCheckAlterColForTSma(pMnode, pOld->name, pOld->uid, colId, true));
2✔
1986

1987
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
2✔
1988

1989
  memmove(pNew->pTags + tag, pNew->pTags + tag + 1, sizeof(SSchema) * (pNew->numOfTags - tag - 1));
2✔
1990
  pNew->numOfTags--;
2✔
1991

1992
  pNew->tagVer++;
2✔
1993

1994
  // if (mndDropIndexByTag(pMnode, pOld, tagName) != 0) {
1995
  //   return -1;
1996
  // }
1997
  mInfo("stb:%s, start to drop tag %s", pNew->name, tagName);
2✔
1998
  TAOS_RETURN(code);
2✔
1999
}
2000

2001
static int32_t mndAlterStbTagName(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, SArray *pFields) {
12✔
2002
  int32_t code = 0;
12✔
2003
  if ((int32_t)taosArrayGetSize(pFields) != 2) {
12✔
2004
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
×
2005
    TAOS_RETURN(code);
×
2006
  }
2007

2008
  SField *pField0 = taosArrayGet(pFields, 0);
12✔
2009
  SField *pField1 = taosArrayGet(pFields, 1);
12✔
2010

2011
  const char *oldTagName = pField0->name;
12✔
2012
  const char *newTagName = pField1->name;
12✔
2013

2014
  int32_t tag = mndFindSuperTableTagIndex(pOld, oldTagName);
12✔
2015
  if (tag < 0) {
12✔
2016
    code = TSDB_CODE_MND_TAG_NOT_EXIST;
4✔
2017
    TAOS_RETURN(code);
4✔
2018
  }
2019

2020
  col_id_t colId = pOld->pTags[tag].colId;
8✔
2021
  TAOS_CHECK_RETURN(mndCheckAlterColForTSma(pMnode, pOld->name, pOld->uid, colId, true));
8✔
2022

2023
  if (mndFindSuperTableTagIndex(pOld, newTagName) >= 0) {
8✔
2024
    code = TSDB_CODE_MND_TAG_ALREADY_EXIST;
4✔
2025
    TAOS_RETURN(code);
4✔
2026
  }
2027

2028
  if (mndFindSuperTableColumnIndex(pOld, newTagName) >= 0) {
4✔
2029
    code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
2✔
2030
    TAOS_RETURN(code);
2✔
2031
  }
2032

2033
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
2✔
2034

2035
  SSchema *pSchema = (SSchema *)(pNew->pTags + tag);
2✔
2036
  memcpy(pSchema->name, newTagName, TSDB_COL_NAME_LEN);
2✔
2037

2038
  pNew->tagVer++;
2✔
2039
  mInfo("stb:%s, start to modify tag %s to %s", pNew->name, oldTagName, newTagName);
2✔
2040
  TAOS_RETURN(code);
2✔
2041
}
2042

2043
static int32_t mndAlterStbTagBytes(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const SField *pField) {
8✔
2044
  int32_t code = 0;
8✔
2045
  int32_t tag = mndFindSuperTableTagIndex(pOld, pField->name);
8✔
2046
  if (tag < 0) {
8✔
2047
    code = TSDB_CODE_MND_TAG_NOT_EXIST;
2✔
2048
    TAOS_RETURN(code);
2✔
2049
  }
2050

2051
  col_id_t colId = pOld->pTags[tag].colId;
6✔
2052
  TAOS_CHECK_RETURN(mndCheckAlterColForTSma(pMnode, pOld->name, pOld->uid, colId, true));
6✔
2053

2054
  uint32_t nLen = 0;
6✔
2055
  for (int32_t i = 0; i < pOld->numOfTags; ++i) {
24✔
2056
    nLen += (pOld->pTags[i].colId == colId) ? pField->bytes : pOld->pTags[i].bytes;
18✔
2057
  }
2058

2059
  if (nLen > TSDB_MAX_TAGS_LEN) {
6✔
2060
    code = TSDB_CODE_PAR_INVALID_TAGS_LENGTH;
×
2061
    TAOS_RETURN(code);
×
2062
  }
2063

2064
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
6✔
2065

2066
  SSchema *pTag = pNew->pTags + tag;
6✔
2067

2068
  if (!(pTag->type == TSDB_DATA_TYPE_BINARY || pTag->type == TSDB_DATA_TYPE_VARBINARY ||
6✔
2069
        pTag->type == TSDB_DATA_TYPE_NCHAR || pTag->type == TSDB_DATA_TYPE_GEOMETRY)) {
2✔
2070
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
2✔
2071
    TAOS_RETURN(code);
2✔
2072
  }
2073

2074
  if (pField->bytes <= pTag->bytes) {
4✔
2075
    code = TSDB_CODE_MND_INVALID_ROW_BYTES;
2✔
2076
    TAOS_RETURN(code);
2✔
2077
  }
2078

2079
  pTag->bytes = pField->bytes;
2✔
2080
  pNew->tagVer++;
2✔
2081

2082
  mInfo("stb:%s, start to modify tag len %s to %d", pNew->name, pField->name, pField->bytes);
2✔
2083
  TAOS_RETURN(code);
2✔
2084
}
2085

2086
static int32_t mndUpdateSuperTableColumnCompress(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, SArray *pField,
×
2087
                                                 int32_t nCols) {
2088
  // if (pColCmpr == NULL || colName == NULL) return -1;
2089

2090
  if (taosArrayGetSize(pField) != nCols) return TSDB_CODE_FAILED;
×
2091
  TAOS_FIELD *p = taosArrayGet(pField, 0);
×
2092

2093
  int32_t code = 0;
×
2094
  int32_t idx = mndFindSuperTableColumnIndex(pOld, p->name);
×
2095
  if (idx == -1) {
×
2096
    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;
×
2097
    TAOS_RETURN(code);
×
2098
  }
2099
  SSchema *pTarget = &pOld->pColumns[idx];
×
2100
  col_id_t colId = pTarget->colId;
×
2101

2102
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
×
2103
  code = validColCmprByType(pTarget->type, p->bytes);
×
2104
  if (code != TSDB_CODE_SUCCESS) {
×
2105
    TAOS_RETURN(code);
×
2106
  }
2107

2108
  int8_t updated = 0;
×
2109
  for (int i = 0; i < pNew->numOfColumns; i++) {
×
2110
    SColCmpr *pCmpr = &pNew->pCmpr[i];
×
2111
    if (pCmpr->id == colId) {
×
2112
      uint32_t dst = 0;
×
2113
      updated = tUpdateCompress(pCmpr->alg, p->bytes, TSDB_COLVAL_COMPRESS_DISABLED, TSDB_COLVAL_LEVEL_DISABLED,
×
2114
                                TSDB_COLVAL_LEVEL_MEDIUM, &dst);
2115
      if (updated > 0) pCmpr->alg = dst;
×
2116
      break;
×
2117
    }
2118
  }
2119

2120
  if (updated == 0) {
×
2121
    code = TSDB_CODE_MND_COLUMN_COMPRESS_ALREADY_EXIST;
×
2122
    TAOS_RETURN(code);
×
2123
  } else if (updated == -1) {
×
2124
    code = TSDB_CODE_TSC_COMPRESS_LEVEL_ERROR;
×
2125
    TAOS_RETURN(code);
×
2126
  }
2127

2128
  pNew->colVer++;
×
2129

2130
  TAOS_RETURN(code);
×
2131
}
2132
static int32_t mndAddSuperTableColumn(const SStbObj *pOld, SStbObj *pNew, const SMAlterStbReq* pReq, int32_t ncols,
8✔
2133
                                      int8_t withCompress) {
2134
  int32_t code = 0;
8✔
2135
  int32_t maxColumns = pOld->virtualStb ? TSDB_MAX_COLUMNS : TSDB_MAX_COLUMNS_NON_VIRTUAL;
8✔
2136
  int32_t maxBytesPerRow = pOld->virtualStb ? TSDB_MAX_BYTES_PER_ROW_VIRTUAL : TSDB_MAX_BYTES_PER_ROW;
8✔
2137
  if (pOld->numOfColumns + ncols + pOld->numOfTags > maxColumns) {
8✔
2138
    code = TSDB_CODE_MND_TOO_MANY_COLUMNS;
×
2139
    TAOS_RETURN(code);
×
2140
  }
2141

2142
  if ((code = grantCheck(TSDB_GRANT_TIMESERIES)) != 0) {
8✔
2143
    TAOS_RETURN(code);
×
2144
  }
2145

2146
  if (!mndValidateSchema(pOld->pColumns, pOld->numOfColumns, pReq->pFields, maxBytesPerRow)) {
8✔
2147
    code = TSDB_CODE_PAR_INVALID_ROW_LENGTH;
×
2148
    TAOS_RETURN(code);
×
2149
  }
2150

2151
  pNew->numOfColumns = pNew->numOfColumns + ncols;
8✔
2152

2153
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
8✔
2154

2155
  if (pNew->nextColId < 0 || pNew->nextColId >= 0x7fff - ncols) {
8✔
2156
    code = TSDB_CODE_OUT_OF_RANGE;
×
2157
    TAOS_RETURN(code);
×
2158
  }
2159

2160
  for (int32_t i = 0; i < ncols; i++) {
12✔
2161
    if (withCompress) {
8✔
2162
      SFieldWithOptions *pField = taosArrayGet(pReq->pFields, i);
×
2163
      if (mndFindSuperTableColumnIndex(pOld, pField->name) >= 0) {
×
2164
        code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
×
2165
        TAOS_RETURN(code);
×
2166
      }
2167

2168
      if (mndFindSuperTableTagIndex(pOld, pField->name) >= 0) {
×
2169
        code = TSDB_CODE_MND_TAG_ALREADY_EXIST;
×
2170
        TAOS_RETURN(code);
×
2171
      }
2172

2173
      SSchema *pSchema = &pNew->pColumns[pOld->numOfColumns + i];
×
2174
      pSchema->bytes = pField->bytes;
×
2175
      pSchema->type = pField->type;
×
2176
      memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
×
2177
      if (pNew->nextColId > INT16_MAX) {
×
2178
        code = TSDB_CODE_MND_EXCEED_MAX_COL_ID;
×
2179
        TAOS_RETURN(code);
×
2180
      }
2181
      pSchema->colId = pNew->nextColId;
×
2182
      pNew->nextColId++;
×
2183

2184
      SColCmpr *pCmpr = &pNew->pCmpr[pOld->numOfColumns + i];
×
2185
      pCmpr->id = pSchema->colId;
×
2186
      pCmpr->alg = pField->compress;
×
2187
      mInfo("stb:%s, start to add column %s", pNew->name, pSchema->name);
×
2188
    } else {
2189
      SField *pField = taosArrayGet(pReq->pFields, i);
8✔
2190
      if (mndFindSuperTableColumnIndex(pOld, pField->name) >= 0) {
8✔
2191
        code = TSDB_CODE_MND_COLUMN_ALREADY_EXIST;
2✔
2192
        TAOS_RETURN(code);
2✔
2193
      }
2194

2195
      if (mndFindSuperTableTagIndex(pOld, pField->name) >= 0) {
6✔
2196
        code = TSDB_CODE_MND_TAG_ALREADY_EXIST;
2✔
2197
        TAOS_RETURN(code);
2✔
2198
      }
2199

2200
      SSchema *pSchema = &pNew->pColumns[pOld->numOfColumns + i];
4✔
2201
      pSchema->bytes = pField->bytes;
4✔
2202
      pSchema->type = pField->type;
4✔
2203
      memcpy(pSchema->name, pField->name, TSDB_COL_NAME_LEN);
4✔
2204
      if (pNew->nextColId > INT16_MAX) {
4✔
2205
        code = TSDB_CODE_MND_EXCEED_MAX_COL_ID;
×
2206
        TAOS_RETURN(code);
×
2207
      }
2208
      pSchema->colId = pNew->nextColId;
4✔
2209
      pNew->nextColId++;
4✔
2210

2211
      SColCmpr *pCmpr = &pNew->pCmpr[pOld->numOfColumns + i];
4✔
2212
      pCmpr->id = pSchema->colId;
4✔
2213
      pCmpr->alg = createDefaultColCmprByType(pSchema->type);
4✔
2214
      mInfo("stb:%s, start to add column %s", pNew->name, pSchema->name);
4✔
2215
    }
2216
  }
2217
  // 1. old schema already has extschemas
2218
  // 2. new schema has extschemas
2219
  if (pReq->pTypeMods || pOld->pExtSchemas) {
4✔
2220
    if (!pNew->pExtSchemas) {
×
2221
      // all ext schemas reset to zero
2222
      pNew->pExtSchemas = taosMemoryCalloc(pNew->numOfColumns, sizeof(SExtSchema));
×
2223
      if (!pNew->pExtSchemas) TAOS_RETURN(terrno);
×
2224
    }
2225
    if (pOld->pExtSchemas) {
×
2226
      memcpy(pNew->pExtSchemas, pOld->pExtSchemas, pOld->numOfColumns * sizeof(SExtSchema));
×
2227
    }
2228
    if (taosArrayGetSize(pReq->pTypeMods) > 0) {
×
2229
      // copy added column ext schema
2230
      for (int32_t i = 0; i < ncols; ++i) {
×
2231
        pNew->pColumns[pOld->numOfColumns + i].flags |= COL_HAS_TYPE_MOD;
×
2232
        pNew->pExtSchemas[pOld->numOfColumns + i].typeMod = *(STypeMod *)taosArrayGet(pReq->pTypeMods, i);
×
2233
      }
2234
    }
2235
  }
2236

2237
  pNew->colVer++;
4✔
2238
  TAOS_RETURN(code);
4✔
2239
}
2240

2241
static int32_t mndDropSuperTableColumn(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const char *colName) {
8✔
2242
  int32_t code = 0;
8✔
2243
  int32_t col = mndFindSuperTableColumnIndex(pOld, colName);
8✔
2244
  if (col < 0) {
8✔
2245
    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;
2✔
2246
    TAOS_RETURN(code);
2✔
2247
  }
2248

2249
  if (col == 0) {
6✔
2250
    code = TSDB_CODE_MND_INVALID_STB_ALTER_OPTION;
2✔
2251
    TAOS_RETURN(code);
2✔
2252
  }
2253

2254
  if (pOld->numOfColumns == 2) {
4✔
2255
    code = TSDB_CODE_PAR_INVALID_DROP_COL;
2✔
2256
    TAOS_RETURN(code);
2✔
2257
  }
2258

2259
  col_id_t colId = pOld->pColumns[col].colId;
2✔
2260
  TAOS_CHECK_RETURN(mndCheckAlterColForTSma(pMnode, pOld->name, pOld->uid, colId, false));
2✔
2261

2262
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
2✔
2263

2264
  int32_t sz = pNew->numOfColumns - col - 1;
2✔
2265
  memmove(pNew->pColumns + col, pNew->pColumns + col + 1, sizeof(SSchema) * sz);
2✔
2266
  memmove(pNew->pCmpr + col, pNew->pCmpr + col + 1, sizeof(SColCmpr) * sz);
2✔
2267
  if (pOld->pExtSchemas) {
2✔
2268
    memmove(pNew->pExtSchemas + col, pNew->pExtSchemas + col + 1, sizeof(SExtSchema) * sz);
×
2269
  }
2270
  pNew->numOfColumns--;
2✔
2271

2272
  pNew->colVer++;
2✔
2273
  mInfo("stb:%s, start to drop col %s", pNew->name, colName);
2✔
2274
  TAOS_RETURN(code);
2✔
2275
}
2276

2277
static int32_t mndAlterStbColumnBytes(SMnode *pMnode, const SStbObj *pOld, SStbObj *pNew, const SField *pField) {
12✔
2278
  int32_t code = 0;
12✔
2279
  int32_t col = mndFindSuperTableColumnIndex(pOld, pField->name);
12✔
2280
  if (col < 0) {
12✔
2281
    code = TSDB_CODE_MND_COLUMN_NOT_EXIST;
4✔
2282
    TAOS_RETURN(code);
4✔
2283
  }
2284

2285
  col_id_t colId = pOld->pColumns[col].colId;
8✔
2286

2287
  uint32_t nLen = 0;
8✔
2288
  int32_t  maxBytesPerRow = pOld->virtualStb ? TSDB_MAX_BYTES_PER_ROW_VIRTUAL : TSDB_MAX_BYTES_PER_ROW;
8✔
2289
  for (int32_t i = 0; i < pOld->numOfColumns; ++i) {
24✔
2290
    nLen += (pOld->pColumns[i].colId == colId) ? pField->bytes : pOld->pColumns[i].bytes;
16✔
2291
  }
2292

2293
  if (nLen > maxBytesPerRow) {
8✔
2294
    code = TSDB_CODE_MND_INVALID_ROW_BYTES;
2✔
2295
    TAOS_RETURN(code);
2✔
2296
  }
2297

2298
  TAOS_CHECK_RETURN(mndCheckAlterColForTSma(pMnode, pOld->name, pOld->uid, colId, false));
6✔
2299

2300
  TAOS_CHECK_RETURN(mndAllocStbSchemas(pOld, pNew));
6✔
2301

2302
  SSchema *pCol = pNew->pColumns + col;
6✔
2303
  if (!(pCol->type == TSDB_DATA_TYPE_BINARY || pCol->type == TSDB_DATA_TYPE_VARBINARY ||
6✔
2304
        pCol->type == TSDB_DATA_TYPE_NCHAR || pCol->type == TSDB_DATA_TYPE_GEOMETRY)) {
2✔
2305
    code = TSDB_CODE_MND_INVALID_STB_OPTION;
2✔
2306
    TAOS_RETURN(code);
2✔
2307
  }
2308

2309
  if (pField->bytes <= pCol->bytes) {
4✔
2310
    code = TSDB_CODE_MND_INVALID_ROW_BYTES;
2✔
2311
    TAOS_RETURN(code);
2✔
2312
  }
2313

2314
  pCol->bytes = pField->bytes;
2✔
2315
  pNew->colVer++;
2✔
2316

2317
  mInfo("stb:%s, start to modify col len %s to %d", pNew->name, pField->name, pField->bytes);
2✔
2318
  TAOS_RETURN(code);
2✔
2319
}
2320

2321
static int32_t mndSetAlterStbPrepareLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
16✔
2322
  int32_t  code = 0;
16✔
2323
  SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);
16✔
2324
  if (pRedoRaw == NULL) {
16✔
2325
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2326
    if (terrno != 0) code = terrno;
×
2327
    TAOS_RETURN(code);
×
2328
  }
2329
  if ((code = mndTransAppendPrepareLog(pTrans, pRedoRaw)) != 0) {
16✔
2330
    sdbFreeRaw(pRedoRaw);
×
2331
    TAOS_RETURN(code);
×
2332
  }
2333
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY));
16✔
2334

2335
  TAOS_RETURN(code);
16✔
2336
}
2337

2338
static int32_t mndSetAlterStbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
16✔
2339
  int32_t  code = 0;
16✔
2340
  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);
16✔
2341
  if (pCommitRaw == NULL) {
16✔
2342
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2343
    if (terrno != 0) code = terrno;
×
2344
    TAOS_RETURN(code);
×
2345
  }
2346
  if ((code = mndTransAppendCommitlog(pTrans, pCommitRaw)) != 0) {
16✔
2347
    sdbFreeRaw(pCommitRaw);
×
2348
    TAOS_RETURN(code);
×
2349
  }
2350
  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));
16✔
2351

2352
  TAOS_RETURN(code);
16✔
2353
}
2354

2355
static int32_t mndSetAlterStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb, void *alterOriData,
16✔
2356
                                         int32_t alterOriDataLen) {
2357
  int32_t code = 0;
16✔
2358
  SSdb   *pSdb = pMnode->pSdb;
16✔
2359
  SVgObj *pVgroup = NULL;
16✔
2360
  void   *pIter = NULL;
16✔
2361
  int32_t contLen;
2362

2363
  while (1) {
32✔
2364
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
48✔
2365
    if (pIter == NULL) break;
48✔
2366
    if (!mndVgroupInDb(pVgroup, pDb->uid)) {
32✔
2367
      sdbRelease(pSdb, pVgroup);
×
2368
      continue;
×
2369
    }
2370

2371
    void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, alterOriData, alterOriDataLen);
32✔
2372
    if (pReq == NULL) {
32✔
2373
      sdbCancelFetch(pSdb, pIter);
×
2374
      sdbRelease(pSdb, pVgroup);
×
2375
      code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2376
      if (terrno != 0) code = terrno;
×
2377
      TAOS_RETURN(code);
×
2378
    }
2379
    STransAction action = {0};
32✔
2380
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
32✔
2381
    action.pCont = pReq;
32✔
2382
    action.contLen = contLen;
32✔
2383
    action.msgType = TDMT_VND_ALTER_STB;
32✔
2384
    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
32✔
2385
      taosMemoryFree(pReq);
×
2386
      sdbCancelFetch(pSdb, pIter);
×
2387
      sdbRelease(pSdb, pVgroup);
×
2388
      TAOS_RETURN(code);
×
2389
    }
2390
    sdbRelease(pSdb, pVgroup);
32✔
2391
  }
2392

2393
  TAOS_RETURN(code);
16✔
2394
}
2395

2396
static int32_t mndSetAlterStbRedoActions2(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb,
×
2397
                                          void *alterOriData, int32_t alterOriDataLen) {
2398
  int32_t code = 0;
×
2399
  SSdb   *pSdb = pMnode->pSdb;
×
2400
  SVgObj *pVgroup = NULL;
×
2401
  void   *pIter = NULL;
×
2402
  int32_t contLen;
2403

2404
  while (1) {
×
2405
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
×
2406
    if (pIter == NULL) break;
×
2407
    if (!mndVgroupInDb(pVgroup, pDb->uid)) {
×
2408
      sdbRelease(pSdb, pVgroup);
×
2409
      continue;
×
2410
    }
2411

2412
    void *pReq = mndBuildVCreateStbReq(pMnode, pVgroup, pStb, &contLen, alterOriData, alterOriDataLen);
×
2413
    if (pReq == NULL) {
×
2414
      sdbCancelFetch(pSdb, pIter);
×
2415
      sdbRelease(pSdb, pVgroup);
×
2416
      code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2417
      if (terrno != 0) code = terrno;
×
2418
      TAOS_RETURN(code);
×
2419
    }
2420
    STransAction action = {0};
×
2421
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
×
2422
    action.pCont = pReq;
×
2423
    action.contLen = contLen;
×
2424
    action.msgType = TDMT_VND_CREATE_INDEX;
×
2425
    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
×
2426
      taosMemoryFree(pReq);
×
2427
      sdbCancelFetch(pSdb, pIter);
×
2428
      sdbRelease(pSdb, pVgroup);
×
2429
      TAOS_RETURN(code);
×
2430
    }
2431
    sdbRelease(pSdb, pVgroup);
×
2432
  }
2433

2434
  TAOS_RETURN(code);
×
2435
}
2436

2437
static int32_t mndBuildStbSchemaImp(SMnode *pMnode, SDbObj *pDb, SStbObj *pStb, const char *tbName, STableMetaRsp *pRsp, bool refByStm) {
36✔
2438
  int32_t code = 0;
36✔
2439
  taosRLockLatch(&pStb->lock);
36✔
2440

2441
  int32_t totalCols = pStb->numOfColumns + pStb->numOfTags;
36✔
2442
  pRsp->pSchemas = taosMemoryCalloc(totalCols, sizeof(SSchema));
36✔
2443
  if (pRsp->pSchemas == NULL) {
36✔
2444
    taosRUnLockLatch(&pStb->lock);
×
2445
    code = terrno;
×
2446
    TAOS_RETURN(code);
×
2447
  }
2448
  pRsp->pSchemaExt = taosMemoryCalloc(pStb->numOfColumns, sizeof(SSchemaExt));
36✔
2449
  if (pRsp->pSchemaExt == NULL) {
36✔
2450
    taosRUnLockLatch(&pStb->lock);
×
2451
    code = terrno;
×
2452
    TAOS_RETURN(code);
×
2453
  }
2454
  pRsp->numOfColRefs = 0;
36✔
2455
  pRsp->pColRefs = NULL;
36✔
2456
  tstrncpy(pRsp->dbFName, pStb->db, sizeof(pRsp->dbFName));
36✔
2457
  tstrncpy(pRsp->tbName, tbName, sizeof(pRsp->tbName));
36✔
2458
  tstrncpy(pRsp->stbName, tbName, sizeof(pRsp->stbName));
36✔
2459
  pRsp->dbId = pDb->uid;
36✔
2460
  pRsp->numOfTags = pStb->numOfTags;
36✔
2461
  pRsp->numOfColumns = pStb->numOfColumns;
36✔
2462
  pRsp->precision = pDb->cfg.precision;
36✔
2463
  pRsp->tableType = TSDB_SUPER_TABLE;
36✔
2464
  pRsp->sversion = pStb->colVer;
36✔
2465
  pRsp->tversion = pStb->tagVer;
36✔
2466
  pRsp->suid = pStb->uid;
36✔
2467
  pRsp->tuid = pStb->uid;
36✔
2468
  pRsp->virtualStb = pStb->virtualStb;
36✔
2469
  pRsp->ownerId = pStb->ownerId;
36✔
2470
  pRsp->isAudit = pDb->cfg.isAudit ? 1 : 0;
36✔
2471

2472
  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
114✔
2473
    SSchema *pSchema = &pRsp->pSchemas[i];
78✔
2474
    SSchema *pSrcSchema = &pStb->pColumns[i];
78✔
2475
    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
78✔
2476
    pSchema->type = pSrcSchema->type;
78✔
2477
    pSchema->flags = pSrcSchema->flags;
78✔
2478
    pSchema->colId = pSrcSchema->colId;
78✔
2479
    pSchema->bytes = pSrcSchema->bytes;
78✔
2480
  }
2481
  
2482
  for (int32_t i = 0; i < pStb->numOfTags; ++i) {
140✔
2483
    SSchema *pSchema = &pRsp->pSchemas[i + pStb->numOfColumns];
104✔
2484
    SSchema *pSrcSchema = &pStb->pTags[i];
104✔
2485
    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
104✔
2486
    pSchema->type = pSrcSchema->type;
104✔
2487
    pSchema->flags = pSrcSchema->flags;
104✔
2488
    pSchema->colId = pSrcSchema->colId;
104✔
2489
    pSchema->bytes = pSrcSchema->bytes;
104✔
2490
  }
2491

2492
  if (refByStm) {
36✔
2493
    mndStreamUpdateTagsRefFlag(pMnode, pStb->uid, &pRsp->pSchemas[pStb->numOfColumns], pStb->numOfTags);
2✔
2494
  }
2495

2496
  for (int32_t i = 0; i < pStb->numOfColumns; i++) {
114✔
2497
    SColCmpr   *pCmpr = &pStb->pCmpr[i];
78✔
2498
    SSchemaExt *pSchEx = &pRsp->pSchemaExt[i];
78✔
2499
    pSchEx->colId = pCmpr->id;
78✔
2500
    pSchEx->compress = pCmpr->alg;
78✔
2501
    if (pStb->pExtSchemas) {
78✔
2502
      pSchEx->typeMod = pStb->pExtSchemas[i].typeMod;
×
2503
    }
2504
  }
2505

2506
  taosRUnLockLatch(&pStb->lock);
36✔
2507
  TAOS_RETURN(code);
36✔
2508
}
2509

2510
static int32_t mndBuildStbCfgImp(SDbObj *pDb, SStbObj *pStb, const char *tbName, STableCfgRsp *pRsp) {
×
2511
  int32_t code = 0;
×
2512
  taosRLockLatch(&pStb->lock);
×
2513

2514
  int32_t totalCols = pStb->numOfColumns + pStb->numOfTags;
×
2515
  pRsp->pSchemas = taosMemoryCalloc(totalCols, sizeof(SSchema));
×
2516
  if (pRsp->pSchemas == NULL) {
×
2517
    taosRUnLockLatch(&pStb->lock);
×
2518
    code = terrno;
×
2519
    TAOS_RETURN(code);
×
2520
  }
2521

2522
  tstrncpy(pRsp->dbFName, pStb->db, sizeof(pRsp->dbFName));
×
2523
  tstrncpy(pRsp->tbName, tbName, sizeof(pRsp->tbName));
×
2524
  tstrncpy(pRsp->stbName, tbName, sizeof(pRsp->stbName));
×
2525
  pRsp->numOfTags = pStb->numOfTags;
×
2526
  pRsp->numOfColumns = pStb->numOfColumns;
×
2527
  pRsp->tableType = TSDB_SUPER_TABLE;
×
2528
  pRsp->delay1 = pStb->maxdelay[0];
×
2529
  pRsp->delay2 = pStb->maxdelay[1];
×
2530
  pRsp->watermark1 = pStb->watermark[0];
×
2531
  pRsp->watermark2 = pStb->watermark[1];
×
2532
  pRsp->ttl = pStb->ttl;
×
2533
  pRsp->keep = pStb->keep;
×
2534
  pRsp->commentLen = pStb->commentLen;
×
2535
  if (pStb->commentLen > 0) {
×
2536
    pRsp->pComment = taosStrdup(pStb->comment);
×
2537
  }
2538

2539
  for (int32_t i = 0; i < pStb->numOfColumns; ++i) {
×
2540
    SSchema *pSchema = &pRsp->pSchemas[i];
×
2541
    SSchema *pSrcSchema = &pStb->pColumns[i];
×
2542
    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
×
2543
    pSchema->type = pSrcSchema->type;
×
2544
    pSchema->flags = pSrcSchema->flags;
×
2545
    pSchema->colId = pSrcSchema->colId;
×
2546
    pSchema->bytes = pSrcSchema->bytes;
×
2547
  }
2548

2549
  for (int32_t i = 0; i < pStb->numOfTags; ++i) {
×
2550
    SSchema *pSchema = &pRsp->pSchemas[i + pStb->numOfColumns];
×
2551
    SSchema *pSrcSchema = &pStb->pTags[i];
×
2552
    memcpy(pSchema->name, pSrcSchema->name, TSDB_COL_NAME_LEN);
×
2553
    pSchema->type = pSrcSchema->type;
×
2554
    pSchema->flags = pSrcSchema->flags;
×
2555
    pSchema->colId = pSrcSchema->colId;
×
2556
    pSchema->bytes = pSrcSchema->bytes;
×
2557
  }
2558

2559
  if (pStb->numOfFuncs > 0) {
×
2560
    pRsp->pFuncs = taosArrayDup(pStb->pFuncs, NULL);
×
2561
  }
2562

2563
  pRsp->pSchemaExt = taosMemoryCalloc(pStb->numOfColumns, sizeof(SSchemaExt));
×
2564
  for (int32_t i = 0; i < pStb->numOfColumns; i++) {
×
2565
    SColCmpr *pCmpr = &pStb->pCmpr[i];
×
2566

2567
    SSchemaExt *pSchExt = &pRsp->pSchemaExt[i];
×
2568
    pSchExt->colId = pCmpr->id;
×
2569
    pSchExt->compress = pCmpr->alg;
×
2570
    if (pStb->pExtSchemas) {
×
2571
      pSchExt->typeMod = pStb->pExtSchemas[i].typeMod;
×
2572
    }
2573
  }
2574
  pRsp->virtualStb = pStb->virtualStb;
×
2575
  pRsp->pColRefs = NULL;
×
2576

2577
  taosRUnLockLatch(&pStb->lock);
×
2578
  TAOS_RETURN(code);
×
2579
}
2580

2581
static int32_t mndValidateStbVersion(SMnode *pMnode, SSTableVersion *pStbVer, bool *schema, bool *sma) {
×
2582
  int32_t code = 0;
×
2583
  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};
×
2584
  snprintf(tbFName, sizeof(tbFName), "%s.%s", pStbVer->dbFName, pStbVer->stbName);
×
2585

2586
  SDbObj *pDb = mndAcquireDb(pMnode, pStbVer->dbFName);
×
2587
  if (pDb == NULL) {
×
2588
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
2589
    TAOS_RETURN(code);
×
2590
  }
2591

2592
  if (pDb->uid != pStbVer->dbId) {
×
2593
    mndReleaseDb(pMnode, pDb);
×
2594
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
2595
    TAOS_RETURN(code);
×
2596
  }
2597

2598
  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);
×
2599
  if (pStb == NULL) {
×
2600
    mndReleaseDb(pMnode, pDb);
×
2601
    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
×
2602
    TAOS_RETURN(code);
×
2603
  }
2604

2605
  taosRLockLatch(&pStb->lock);
×
2606

2607
  if (pStbVer->sversion != pStb->colVer || pStbVer->tversion != pStb->tagVer) {
×
2608
    *schema = true;
×
2609
  } else {
2610
    *schema = false;
×
2611
  }
2612

2613
  if (pStbVer->smaVer && pStbVer->smaVer != pStb->smaVer) {
×
2614
    *sma = true;
×
2615
  } else {
2616
    *sma = false;
×
2617
  }
2618

2619
  taosRUnLockLatch(&pStb->lock);
×
2620

2621
  mndReleaseDb(pMnode, pDb);
×
2622
  mndReleaseStb(pMnode, pStb);
×
2623
  return TSDB_CODE_SUCCESS;
×
2624
}
2625

2626
static int32_t mndBuildStbSchema(SMnode *pMnode, const char *dbFName, const char *tbName, STableMetaRsp *pRsp, bool refByStm) {
2✔
2627
  int32_t code = 0;
2✔
2628
  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};
2✔
2629
  snprintf(tbFName, sizeof(tbFName), "%s.%s", dbFName, tbName);
2✔
2630

2631
  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);
2✔
2632
  if (pDb == NULL) {
2✔
2633
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
2634
    TAOS_RETURN(code);
×
2635
  }
2636

2637
  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);
2✔
2638
  if (pStb == NULL) {
2✔
2639
    mndReleaseDb(pMnode, pDb);
×
2640
    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
×
2641
    TAOS_RETURN(code);
×
2642
  }
2643

2644
  code = mndBuildStbSchemaImp(pMnode, pDb, pStb, tbName, pRsp, refByStm);
2✔
2645
  mndReleaseDb(pMnode, pDb);
2✔
2646
  mndReleaseStb(pMnode, pStb);
2✔
2647
  TAOS_RETURN(code);
2✔
2648
}
2649

2650
static int32_t mndBuildStbCfg(SMnode *pMnode, const char *dbFName, const char *tbName, STableCfgRsp *pRsp) {
×
2651
  int32_t code = 0;
×
2652
  char    tbFName[TSDB_TABLE_FNAME_LEN] = {0};
×
2653
  snprintf(tbFName, sizeof(tbFName), "%s.%s", dbFName, tbName);
×
2654

2655
  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);
×
2656
  if (pDb == NULL) {
×
2657
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
2658
    TAOS_RETURN(code);
×
2659
  }
2660

2661
  SStbObj *pStb = mndAcquireStb(pMnode, tbFName);
×
2662
  if (pStb == NULL) {
×
2663
    mndReleaseDb(pMnode, pDb);
×
2664
    code = TSDB_CODE_PAR_TABLE_NOT_EXIST;
×
2665
    TAOS_RETURN(code);
×
2666
  }
2667

2668
  code = mndBuildStbCfgImp(pDb, pStb, tbName, pRsp);
×
2669

2670
  mndReleaseDb(pMnode, pDb);
×
2671
  mndReleaseStb(pMnode, pStb);
×
2672
  TAOS_RETURN(code);
×
2673
}
2674

2675
static int32_t mndBuildSMAlterStbRsp(SDbObj *pDb, SStbObj *pObj, void **pCont, int32_t *pLen) {
16✔
2676
  int32_t       code = 0;
16✔
2677
  SEncoder      ec = {0};
16✔
2678
  uint32_t      contLen = 0;
16✔
2679
  SMAlterStbRsp alterRsp = {0};
16✔
2680
  SName         name = {0};
16✔
2681
  TAOS_CHECK_RETURN(tNameFromString(&name, pObj->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
16✔
2682

2683
  alterRsp.pMeta = taosMemoryCalloc(1, sizeof(STableMetaRsp));
16✔
2684
  if (NULL == alterRsp.pMeta) {
16✔
2685
    code = terrno;
×
2686
    TAOS_RETURN(code);
×
2687
  }
2688

2689
  code = mndBuildStbSchemaImp(NULL, pDb, pObj, name.tname, alterRsp.pMeta, false);
16✔
2690
  if (code) {
16✔
2691
    tFreeSMAlterStbRsp(&alterRsp);
×
2692
    return code;
×
2693
  }
2694

2695
  tEncodeSize(tEncodeSMAlterStbRsp, &alterRsp, contLen, code);
16✔
2696
  if (code) {
16✔
2697
    tFreeSMAlterStbRsp(&alterRsp);
×
2698
    return code;
×
2699
  }
2700

2701
  void *cont = taosMemoryMalloc(contLen);
16✔
2702
  if (NULL == cont) {
16✔
2703
    code = terrno;
×
2704
    tFreeSMAlterStbRsp(&alterRsp);
×
2705
    TAOS_RETURN(code);
×
2706
  }
2707
  tEncoderInit(&ec, cont, contLen);
16✔
2708
  code = tEncodeSMAlterStbRsp(&ec, &alterRsp);
16✔
2709
  tEncoderClear(&ec);
16✔
2710

2711
  tFreeSMAlterStbRsp(&alterRsp);
16✔
2712

2713
  if (code < 0) TAOS_RETURN(code);
16✔
2714

2715
  *pCont = cont;
16✔
2716
  *pLen = contLen;
16✔
2717

2718
  TAOS_RETURN(code);
16✔
2719
}
2720

2721
int32_t mndBuildSMCreateStbRsp(SMnode *pMnode, char *dbFName, char *stbFName, void **pCont, int32_t *pLen) {
18✔
2722
  int32_t code = -1;
18✔
2723
  SDbObj *pDb = mndAcquireDb(pMnode, dbFName);
18✔
2724
  if (NULL == pDb) {
18✔
2725
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2726
    if (terrno != 0) code = terrno;
×
2727
    TAOS_RETURN(code);
×
2728
  }
2729

2730
  SStbObj *pObj = mndAcquireStb(pMnode, stbFName);
18✔
2731
  if (NULL == pObj) {
18✔
2732
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2733
    if (terrno != 0) code = terrno;
×
2734
    goto _OVER;
×
2735
  }
2736

2737
  SEncoder       ec = {0};
18✔
2738
  uint32_t       contLen = 0;
18✔
2739
  SMCreateStbRsp stbRsp = {0};
18✔
2740
  SName          name = {0};
18✔
2741
  TAOS_CHECK_GOTO(tNameFromString(&name, pObj->name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE), NULL, _OVER);
18✔
2742

2743
  stbRsp.pMeta = taosMemoryCalloc(1, sizeof(STableMetaRsp));
18✔
2744
  if (NULL == stbRsp.pMeta) {
18✔
2745
    code = terrno;
×
2746
    goto _OVER;
×
2747
  }
2748

2749
  code = mndBuildStbSchemaImp(NULL, pDb, pObj, name.tname, stbRsp.pMeta, false);
18✔
2750
  if (code) {
18✔
2751
    tFreeSMCreateStbRsp(&stbRsp);
×
2752
    goto _OVER;
×
2753
  }
2754

2755
  tEncodeSize(tEncodeSMCreateStbRsp, &stbRsp, contLen, code);
18✔
2756
  if (code) {
18✔
2757
    tFreeSMCreateStbRsp(&stbRsp);
×
2758
    goto _OVER;
×
2759
  }
2760

2761
  void *cont = taosMemoryMalloc(contLen);
18✔
2762
  if (NULL == cont) {
18✔
2763
    code = terrno;
×
2764
    tFreeSMCreateStbRsp(&stbRsp);
×
2765
    goto _OVER;
×
2766
  }
2767
  tEncoderInit(&ec, cont, contLen);
18✔
2768
  TAOS_CHECK_GOTO(tEncodeSMCreateStbRsp(&ec, &stbRsp), NULL, _OVER);
18✔
2769
  tEncoderClear(&ec);
18✔
2770

2771
  tFreeSMCreateStbRsp(&stbRsp);
18✔
2772

2773
  *pCont = cont;
18✔
2774
  *pLen = contLen;
18✔
2775

2776
  code = 0;
18✔
2777

2778
_OVER:
18✔
2779
  if (pObj) {
18✔
2780
    mndReleaseStb(pMnode, pObj);
18✔
2781
  }
2782

2783
  if (pDb) {
18✔
2784
    mndReleaseDb(pMnode, pDb);
18✔
2785
  }
2786

2787
  TAOS_RETURN(code);
18✔
2788
}
2789

2790
static int32_t mndAlterStbImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp,
12✔
2791
                              void *alterOriData, int32_t alterOriDataLen) {
2792
  int32_t code = -1;
12✔
2793
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "alter-stb");
12✔
2794
  if (pTrans == NULL) {
12✔
2795
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2796
    if (terrno != 0) code = terrno;
×
2797
    goto _OVER;
×
2798
  }
2799

2800
  mInfo("trans:%d, used to alter stb:%s, alterOriDataLen:%d", pTrans->id, pStb->name, alterOriDataLen);
12✔
2801
  mndTransSetDbName(pTrans, pDb->name, pStb->name);
12✔
2802
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
12✔
2803

2804
  if (needRsp) {
12✔
2805
    void   *pCont = NULL;
12✔
2806
    int32_t contLen = 0;
12✔
2807
    TAOS_CHECK_GOTO(mndBuildSMAlterStbRsp(pDb, pStb, &pCont, &contLen), NULL, _OVER);
12✔
2808
    mndTransSetRpcRsp(pTrans, pCont, contLen);
12✔
2809
  }
2810

2811
  TAOS_CHECK_GOTO(mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
12✔
2812
  TAOS_CHECK_GOTO(mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
12✔
2813
  TAOS_CHECK_GOTO(mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb, alterOriData, alterOriDataLen), NULL, _OVER);
12✔
2814
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
12✔
2815

2816
  code = 0;
12✔
2817

2818
_OVER:
12✔
2819
  mndTransDrop(pTrans);
12✔
2820
  TAOS_RETURN(code);
12✔
2821
}
2822

2823
static int32_t mndAlterStbAndUpdateTagIdxImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp,
4✔
2824
                                             void *alterOriData, int32_t alterOriDataLen, const SMAlterStbReq *pAlter) {
2825
  int32_t code = -1;
4✔
2826
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "alter-stb");
4✔
2827
  if (pTrans == NULL) {
4✔
2828
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
2829
    if (terrno != 0) code = terrno;
×
2830
    goto _OVER;
×
2831
  }
2832

2833
  mInfo("trans:%d, used to alter stb:%s", pTrans->id, pStb->name);
4✔
2834
  mndTransSetDbName(pTrans, pDb->name, pStb->name);
4✔
2835

2836
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
4✔
2837

2838
  if (needRsp) {
4✔
2839
    void   *pCont = NULL;
4✔
2840
    int32_t contLen = 0;
4✔
2841
    TAOS_CHECK_GOTO(mndBuildSMAlterStbRsp(pDb, pStb, &pCont, &contLen), NULL, _OVER);
4✔
2842
    mndTransSetRpcRsp(pTrans, pCont, contLen);
4✔
2843
  }
2844

2845
  if (pAlter->alterType == TSDB_ALTER_TABLE_DROP_TAG) {
4✔
2846
    SIdxObj idxObj = {0};
2✔
2847
    SField *pField0 = taosArrayGet(pAlter->pFields, 0);
2✔
2848
    bool    exist = false;
2✔
2849
    if (mndGetIdxsByTagName(pMnode, pStb, pField0->name, &idxObj) == 0) {
2✔
2850
      exist = true;
×
2851
    }
2852
    TAOS_CHECK_GOTO(mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
2✔
2853
    TAOS_CHECK_GOTO(mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
2✔
2854

2855
    if (exist == true) {
2✔
2856
      TAOS_CHECK_GOTO(mndSetDropIdxPrepareLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
×
2857
      TAOS_CHECK_GOTO(mndSetDropIdxCommitLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
×
2858
    }
2859

2860
    TAOS_CHECK_GOTO(mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb, alterOriData, alterOriDataLen), NULL, _OVER);
2✔
2861
    TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
2✔
2862

2863
  } else if (pAlter->alterType == TSDB_ALTER_TABLE_UPDATE_TAG_NAME) {
2✔
2864
    SIdxObj     idxObj = {0};
2✔
2865
    SField     *pField0 = taosArrayGet(pAlter->pFields, 0);
2✔
2866
    SField     *pField1 = taosArrayGet(pAlter->pFields, 1);
2✔
2867
    const char *oTagName = pField0->name;
2✔
2868
    const char *nTagName = pField1->name;
2✔
2869
    bool        exist = false;
2✔
2870

2871
    if (mndGetIdxsByTagName(pMnode, pStb, pField0->name, &idxObj) == 0) {
2✔
2872
      exist = true;
×
2873
    }
2874

2875
    TAOS_CHECK_GOTO(mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
2✔
2876
    TAOS_CHECK_GOTO(mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb), NULL, _OVER);
2✔
2877

2878
    if (exist == true) {
2✔
2879
      memcpy(idxObj.colName, nTagName, strlen(nTagName));
×
2880
      idxObj.colName[strlen(nTagName)] = 0;
×
2881
      TAOS_CHECK_GOTO(mndSetAlterIdxPrepareLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
×
2882
      TAOS_CHECK_GOTO(mndSetAlterIdxCommitLogs(pMnode, pTrans, &idxObj), NULL, _OVER);
×
2883
    }
2884

2885
    TAOS_CHECK_GOTO(mndSetAlterStbRedoActions(pMnode, pTrans, pDb, pStb, alterOriData, alterOriDataLen), NULL, _OVER);
2✔
2886
    TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
2✔
2887
  }
2888
  code = 0;
4✔
2889

2890
_OVER:
4✔
2891
  mndTransDrop(pTrans);
4✔
2892
  TAOS_RETURN(code);
4✔
2893
}
2894

2895
static int32_t mndAlterStb(SMnode *pMnode, SRpcMsg *pReq, const SMAlterStbReq *pAlter, SDbObj *pDb, SStbObj *pOld) {
58✔
2896
  bool    needRsp = true;
58✔
2897
  int32_t code = -1;
58✔
2898
  SField *pField0 = NULL;
58✔
2899

2900
  SStbObj stbObj = {0};
58✔
2901
  taosRLockLatch(&pOld->lock);
58✔
2902
  memcpy(&stbObj, pOld, sizeof(SStbObj));
58✔
2903
  taosRUnLockLatch(&pOld->lock);
58✔
2904
  stbObj.pColumns = NULL;
58✔
2905
  stbObj.pTags = NULL;
58✔
2906
  stbObj.pFuncs = NULL;
58✔
2907
  stbObj.pCmpr = NULL;
58✔
2908
  stbObj.pExtSchemas = NULL;
58✔
2909
  stbObj.updateTime = taosGetTimestampMs();
58✔
2910
  stbObj.lock = 0;
58✔
2911
  stbObj.virtualStb = pOld->virtualStb;
58✔
2912
  bool updateTagIndex = false;
58✔
2913
  switch (pAlter->alterType) {
58✔
2914
    case TSDB_ALTER_TABLE_ADD_TAG:
6✔
2915
      code = mndAddSuperTableTag(pOld, &stbObj, pAlter->pFields, pAlter->numOfFields);
6✔
2916
      break;
6✔
2917
    case TSDB_ALTER_TABLE_DROP_TAG:
4✔
2918
      pField0 = taosArrayGet(pAlter->pFields, 0);
4✔
2919
      code = mndDropSuperTableTag(pMnode, pOld, &stbObj, pField0->name);
4✔
2920
      updateTagIndex = true;
4✔
2921
      break;
4✔
2922
    case TSDB_ALTER_TABLE_UPDATE_TAG_NAME:
12✔
2923
      code = mndAlterStbTagName(pMnode, pOld, &stbObj, pAlter->pFields);
12✔
2924
      updateTagIndex = true;
12✔
2925
      break;
12✔
2926
    case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES:
8✔
2927
      pField0 = taosArrayGet(pAlter->pFields, 0);
8✔
2928
      code = mndAlterStbTagBytes(pMnode, pOld, &stbObj, pField0);
8✔
2929
      break;
8✔
2930
    case TSDB_ALTER_TABLE_ADD_COLUMN:
8✔
2931
      code = mndAddSuperTableColumn(pOld, &stbObj, pAlter, pAlter->numOfFields, 0);
8✔
2932
      break;
8✔
2933
    case TSDB_ALTER_TABLE_DROP_COLUMN:
8✔
2934
      pField0 = taosArrayGet(pAlter->pFields, 0);
8✔
2935
      code = mndDropSuperTableColumn(pMnode, pOld, &stbObj, pField0->name);
8✔
2936
      break;
8✔
2937
    case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES:
12✔
2938
      pField0 = taosArrayGet(pAlter->pFields, 0);
12✔
2939
      code = mndAlterStbColumnBytes(pMnode, pOld, &stbObj, pField0);
12✔
2940
      break;
12✔
2941
    case TSDB_ALTER_TABLE_UPDATE_OPTIONS:
×
2942
      needRsp = false;
×
2943
      code = mndUpdateTableOptions(pOld, &stbObj, pAlter->comment, pAlter->commentLen, pAlter->ttl, pAlter->keep);
×
2944
      break;
×
2945
    case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS:
×
2946
      code = mndUpdateSuperTableColumnCompress(pMnode, pOld, &stbObj, pAlter->pFields, pAlter->numOfFields);
×
2947
      break;
×
2948
    case TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COMPRESS_OPTION:
×
2949
      code = mndAddSuperTableColumn(pOld, &stbObj, pAlter, pAlter->numOfFields, 1);
×
2950
      break;
×
2951
    default:
×
2952
      needRsp = false;
×
2953
      terrno = TSDB_CODE_OPS_NOT_SUPPORT;
×
2954
      break;
×
2955
  }
2956

2957
  if (code != 0) goto _OVER;
58✔
2958
  if (updateTagIndex == false) {
16✔
2959
    code = mndAlterStbImp(pMnode, pReq, pDb, &stbObj, needRsp, pReq->pCont, pReq->contLen);
12✔
2960
  } else {
2961
    code = mndAlterStbAndUpdateTagIdxImp(pMnode, pReq, pDb, &stbObj, needRsp, pReq->pCont, pReq->contLen, pAlter);
4✔
2962
  }
2963

2964
_OVER:
58✔
2965
  taosMemoryFreeClear(stbObj.pTags);
58✔
2966
  taosMemoryFreeClear(stbObj.pColumns);
58✔
2967
  taosMemoryFreeClear(stbObj.pCmpr);
58✔
2968
  if (pAlter->commentLen > 0) {
58✔
2969
    taosMemoryFreeClear(stbObj.comment);
×
2970
  }
2971
  taosMemoryFreeClear(stbObj.pExtSchemas);
58✔
2972
  TAOS_RETURN(code);
58✔
2973
}
2974

2975
static int32_t mndProcessAlterStbReq(SRpcMsg *pReq) {
66✔
2976
  SMnode       *pMnode = pReq->info.node;
66✔
2977
  int32_t       code = -1;
66✔
2978
  SDbObj       *pDb = NULL;
66✔
2979
  SStbObj      *pStb = NULL;
66✔
2980
  SUserObj     *pOperUser = NULL;
66✔
2981
  SMAlterStbReq alterReq = {0};
66✔
2982
  int64_t       tss = taosGetTimestampMs();
66✔
2983

2984
  if (tDeserializeSMAlterStbReq(pReq->pCont, pReq->contLen, &alterReq) != 0) {
66✔
2985
    code = TSDB_CODE_INVALID_MSG;
×
2986
    goto _OVER;
×
2987
  }
2988

2989
  mInfo("stb:%s, start to alter", alterReq.name);
66✔
2990
  if (mndCheckAlterStbReq(&alterReq) != 0) goto _OVER;
66✔
2991

2992
  pDb = mndAcquireDbByStb(pMnode, alterReq.name);
66✔
2993
  if (pDb == NULL) {
66✔
2994
    code = TSDB_CODE_MND_DB_NOT_EXIST;
4✔
2995
    goto _OVER;
4✔
2996
  }
2997
  if (pDb->cfg.isMount) {
62✔
2998
    code = TSDB_CODE_MND_MOUNT_OBJ_NOT_SUPPORT;
×
2999
    goto _OVER;
×
3000
  }
3001

3002
  pStb = mndAcquireStb(pMnode, alterReq.name);
62✔
3003
  if (pStb == NULL) {
62✔
3004
    code = TSDB_CODE_MND_STB_NOT_EXIST;
4✔
3005
    goto _OVER;
4✔
3006
  }
3007

3008
  SName   name = {0};
58✔
3009
  int32_t ret = 0;
58✔
3010
  if ((ret = tNameFromString(&name, alterReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0)
58✔
3011
    mError("stb:%s, failed to tNameFromString since %s", alterReq.name, tstrerror(ret));
×
3012

3013
  // if ((code = mndCheckDbPrivilege(pMnode, RPC_MSG_USER(pReq), MND_OPER_WRITE_DB, pDb)) != 0) {
3014
  //   goto _OVER;
3015
  // }
3016
  TAOS_CHECK_GOTO(mndAcquireUser(pMnode, RPC_MSG_USER(pReq), &pOperUser), NULL, _OVER);
58✔
3017
  TAOS_CHECK_GOTO(mndCheckDbPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_USE_DB, pDb), NULL,
58✔
3018
                  _OVER);
3019
  TAOS_CHECK_GOTO(mndCheckDbPrivilegeByNameRecF(pMnode, pOperUser, PRIV_CM_ALTER, PRIV_OBJ_TBL, pDb->name, name.tname),
58✔
3020
                  NULL, _OVER);
3021

3022
  code = mndAlterStb(pMnode, pReq, &alterReq, pDb, pStb);
58✔
3023
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
58✔
3024

3025
  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {
58✔
3026
    int64_t tse = taosGetTimestampMs();
58✔
3027
    double  duration = (double)(tse - tss);
58✔
3028
    duration = duration / 1000;
58✔
3029
    auditRecord(pReq, pMnode->clusterId, "alterStb", name.dbname, name.tname, alterReq.sql, alterReq.sqlLen, duration,
58✔
3030
                0);
3031
  }
3032
_OVER:
×
3033
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
66✔
3034
    mError("stb:%s, failed to alter since %s", alterReq.name, tstrerror(code));
50✔
3035
  }
3036

3037
  mndReleaseStb(pMnode, pStb);
66✔
3038
  mndReleaseDb(pMnode, pDb);
66✔
3039
  mndReleaseUser(pMnode, pOperUser);
66✔
3040
  tFreeSMAltertbReq(&alterReq);
66✔
3041

3042
  TAOS_RETURN(code);
66✔
3043
}
3044

3045
static int32_t mndSetDropStbPrepareLogs(SMnode *pMnode, STrans *pTrans, SStbObj *pStb) {
4✔
3046
  int32_t  code = 0;
4✔
3047
  SSdbRaw *pRedoRaw = mndStbActionEncode(pStb);
4✔
3048
  if (pRedoRaw == NULL) {
4✔
3049
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
3050
    if (terrno != 0) code = terrno;
×
3051
    TAOS_RETURN(code);
×
3052
  }
3053
  if ((code = mndTransAppendPrepareLog(pTrans, pRedoRaw)) != 0) {
4✔
3054
    sdbFreeRaw(pRedoRaw);
×
3055
    TAOS_RETURN(code);
×
3056
  }
3057
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING));
4✔
3058

3059
  TAOS_RETURN(code);
4✔
3060
}
3061

3062
static int32_t mndSetDropStbCommitLogs(SMnode *pMnode, STrans *pTrans, SStbObj *pStb) {
4✔
3063
  int32_t  code = 0;
4✔
3064
  SSdbRaw *pCommitRaw = mndStbActionEncode(pStb);
4✔
3065
  if (pCommitRaw == NULL) {
4✔
3066
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
3067
    if (terrno != 0) code = terrno;
×
3068
    TAOS_RETURN(code);
×
3069
  }
3070
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
4✔
3071
    sdbFreeRaw(pCommitRaw);
×
3072
    TAOS_RETURN(code);
×
3073
  }
3074
  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));
4✔
3075

3076
  TAOS_RETURN(code);
4✔
3077
}
3078

3079
static int32_t mndSetDropStbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SStbObj *pStb) {
4✔
3080
  int32_t code = 0;
4✔
3081
  SSdb   *pSdb = pMnode->pSdb;
4✔
3082
  SVgObj *pVgroup = NULL;
4✔
3083
  void   *pIter = NULL;
4✔
3084

3085
  while (1) {
8✔
3086
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
12✔
3087
    if (pIter == NULL) break;
12✔
3088
    if (!mndVgroupInDb(pVgroup, pDb->uid)) {
8✔
3089
      sdbRelease(pSdb, pVgroup);
×
3090
      continue;
×
3091
    }
3092

3093
    int32_t contLen = 0;
8✔
3094
    void   *pReq = mndBuildVDropStbReq(pMnode, pVgroup, pStb, &contLen);
8✔
3095
    if (pReq == NULL) {
8✔
3096
      sdbCancelFetch(pSdb, pIter);
×
3097
      sdbRelease(pSdb, pVgroup);
×
3098
      code = TSDB_CODE_OUT_OF_MEMORY;
×
3099
      TAOS_RETURN(code);
×
3100
    }
3101

3102
    STransAction action = {0};
8✔
3103
    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
8✔
3104
    action.pCont = pReq;
8✔
3105
    action.contLen = contLen;
8✔
3106
    action.msgType = TDMT_VND_DROP_STB;
8✔
3107
    action.acceptableCode = TSDB_CODE_TDB_STB_NOT_EXIST;
8✔
3108
    if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
8✔
3109
      taosMemoryFree(pReq);
×
3110
      sdbCancelFetch(pSdb, pIter);
×
3111
      sdbRelease(pSdb, pVgroup);
×
3112
      TAOS_RETURN(code);
×
3113
    }
3114
    sdbRelease(pSdb, pVgroup);
8✔
3115
  }
3116

3117
  TAOS_RETURN(code);
4✔
3118
}
3119

3120
static int32_t mndDropStb(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb) {
4✔
3121
  int32_t code = -1;
4✔
3122
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "drop-stb");
4✔
3123
  if (pTrans == NULL) {
4✔
3124
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
3125
    if (terrno != 0) code = terrno;
×
3126
    goto _OVER;
×
3127
  }
3128

3129
  mInfo("trans:%d, used to drop stb:%s", pTrans->id, pStb->name);
4✔
3130
  mndTransSetDbName(pTrans, pDb->name, pStb->name);
4✔
3131
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
4✔
3132

3133
  TAOS_CHECK_GOTO(mndSetDropStbPrepareLogs(pMnode, pTrans, pStb), NULL, _OVER);
4✔
3134
  TAOS_CHECK_GOTO(mndSetDropStbCommitLogs(pMnode, pTrans, pStb), NULL, _OVER);
4✔
3135
  TAOS_CHECK_GOTO(mndSetDropStbRedoActions(pMnode, pTrans, pDb, pStb), NULL, _OVER);
4✔
3136
  TAOS_CHECK_GOTO(mndDropIdxsByStb(pMnode, pTrans, pDb, pStb), NULL, _OVER);
4✔
3137
  TAOS_CHECK_GOTO(mndDropRsmaByStb(pMnode, pTrans, pDb, pStb), NULL, _OVER);
4✔
3138
  TAOS_CHECK_GOTO(mndUserRemoveStb(pMnode, pTrans, pStb->name), NULL, _OVER);
4✔
3139
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
4✔
3140
  code = 0;
4✔
3141

3142
_OVER:
4✔
3143
  mndTransDrop(pTrans);
4✔
3144
  TAOS_RETURN(code);
4✔
3145
}
3146

3147
static int32_t mndCheckDropStbForTopic(SMnode *pMnode, const char *stbFullName, int64_t suid) {
4✔
3148
  int32_t code = 0;
4✔
3149
  SSdb   *pSdb = pMnode->pSdb;
4✔
3150
  void   *pIter = NULL;
4✔
3151
  while (1) {
×
3152
    SMqTopicObj *pTopic = NULL;
4✔
3153
    pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);
4✔
3154
    if (pIter == NULL) break;
4✔
3155

3156
    if (pTopic->stbUid == suid) {
×
3157
      sdbRelease(pSdb, pTopic);
×
3158
      sdbCancelFetch(pSdb, pIter);
×
3159
      TAOS_RETURN(TSDB_CODE_MND_TOPIC_MUST_BE_DELETED);
×
3160
    }
3161
    sdbRelease(pSdb, pTopic);
×
3162
  }
3163
  TAOS_RETURN(code);
4✔
3164
}
3165

3166
static int32_t mndCheckDropStbForStream(SMnode *pMnode, const char *stbFullName, int64_t suid) {
×
3167
  int32_t code = 0;
×
3168
  SSdb   *pSdb = pMnode->pSdb;
×
3169
  void   *pIter = NULL;
×
3170
  while (1) {
×
3171
    SStreamObj *pStream = NULL;
×
3172
    pIter = sdbFetch(pSdb, SDB_STREAM, pIter, (void **)&pStream);
×
3173
    if (pIter == NULL) break;
×
3174

3175
    if (pStream->pCreate->outStbUid == suid) {
×
3176
      sdbCancelFetch(pSdb, pIter);
×
3177
      sdbRelease(pSdb, pStream);
×
3178
      TAOS_RETURN(-1);
×
3179
    }
3180

3181
    sdbRelease(pSdb, pStream);
×
3182
  }
3183
  TAOS_RETURN(code);
×
3184
}
3185

3186
static int32_t mndProcessDropTtltbRsp(SRpcMsg *pRsp) { return 0; }
×
3187
static int32_t mndProcessTrimDbRsp(SRpcMsg *pRsp) { return 0; }
×
3188
static int32_t mndProcessTrimDbWalRsp(SRpcMsg *pRsp) { return 0; }
×
3189
static int32_t mndProcessS3MigrateDbRsp(SRpcMsg *pRsp) { return 0; }
×
3190

3191
static int32_t mndProcessDropStbReq(SRpcMsg *pReq) {
6✔
3192
  SMnode      *pMnode = pReq->info.node;
6✔
3193
  int32_t      code = -1;
6✔
3194
  SDbObj      *pDb = NULL;
6✔
3195
  SStbObj     *pStb = NULL;
6✔
3196
  SUserObj    *pOperUser = NULL;
6✔
3197
  SMDropStbReq dropReq = {0};
6✔
3198
  int64_t      tss = taosGetTimestampMs();
6✔
3199

3200
  TAOS_CHECK_GOTO(tDeserializeSMDropStbReq(pReq->pCont, pReq->contLen, &dropReq), NULL, _OVER);
6✔
3201

3202
  mInfo("stb:%s, start to drop", dropReq.name);
6✔
3203

3204
  pStb = mndAcquireStb(pMnode, dropReq.name);
6✔
3205
  if (pStb == NULL) {
6✔
3206
    if (dropReq.igNotExists) {
2✔
3207
      mInfo("stb:%s, not exist, ignore not exist is set", dropReq.name);
×
3208
      code = 0;
×
3209
      goto _OVER;
×
3210
    } else {
3211
      code = TSDB_CODE_MND_STB_NOT_EXIST;
2✔
3212
      goto _OVER;
2✔
3213
    }
3214
  }
3215

3216
  if ((dropReq.source == TD_REQ_FROM_TAOX_OLD || dropReq.source == TD_REQ_FROM_TAOX) && pStb->uid != dropReq.suid) {
4✔
3217
    code = 0;
×
3218
    goto _OVER;
×
3219
  }
3220

3221
  pDb = mndAcquireDbByStb(pMnode, dropReq.name);
4✔
3222
  if (pDb == NULL) {
4✔
3223
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
3224
    goto _OVER;
×
3225
  }
3226

3227
  SName   name = {0};
4✔
3228
  int32_t ret = 0;
4✔
3229
  if ((ret = tNameFromString(&name, dropReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE)) != 0)
4✔
3230
    mError("stb:%s, failed to tNameFromString since %s", dropReq.name, tstrerror(ret));
×
3231

3232
  // if ((code = mndCheckDbPrivilege(pMnode, RPC_MSG_USER(pReq), MND_OPER_WRITE_DB, pDb)) != 0) {
3233
  //   goto _OVER;
3234
  // }
3235
  TAOS_CHECK_GOTO(mndAcquireUser(pMnode, RPC_MSG_USER(pReq), &pOperUser), NULL, _OVER);
4✔
3236
  TAOS_CHECK_GOTO(mndCheckDbPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_USE_DB, pDb), NULL,
4✔
3237
                  _OVER);
3238
  TAOS_CHECK_GOTO(
4✔
3239
      mndCheckObjPrivilegeRecF(pMnode, pOperUser, PRIV_CM_DROP, PRIV_OBJ_TBL, pStb->ownerId, pDb->name, name.tname),
3240
      NULL, _OVER);
3241
  if ((code = mndCheckDropStbForTopic(pMnode, dropReq.name, pStb->uid)) != 0) {
4✔
3242
    goto _OVER;
×
3243
  }
3244

3245
  if (pDb->cfg.isMount) {
4✔
3246
    code = TSDB_CODE_MND_MOUNT_OBJ_NOT_SUPPORT;
×
3247
    goto _OVER;
×
3248
  }
3249

3250
  code = mndDropStb(pMnode, pReq, pDb, pStb);
4✔
3251
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
4✔
3252

3253
  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {
4✔
3254
    int64_t tse = taosGetTimestampMs();
4✔
3255
    double  duration = (double)(tse - tss);
4✔
3256
    duration = duration / 1000;
4✔
3257
    auditRecord(pReq, pMnode->clusterId, "dropStb", name.dbname, name.tname, dropReq.sql, dropReq.sqlLen, duration, 0);
4✔
3258
  }
3259
_OVER:
×
3260
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
6✔
3261
    mError("stb:%s, failed to drop since %s", dropReq.name, tstrerror(code));
2✔
3262
  }
3263

3264
  mndReleaseDb(pMnode, pDb);
6✔
3265
  mndReleaseStb(pMnode, pStb);
6✔
3266
  mndReleaseUser(pMnode, pOperUser);
6✔
3267
  tFreeSMDropStbReq(&dropReq);
6✔
3268
  TAOS_RETURN(code);
6✔
3269
}
3270

3271
static int32_t mndProcessTableMetaReq(SRpcMsg *pReq) {
2✔
3272
  SMnode       *pMnode = pReq->info.node;
2✔
3273
  int32_t       code = -1;
2✔
3274
  STableInfoReq infoReq = {0};
2✔
3275
  STableMetaRsp metaRsp = {0};
2✔
3276
  SUserObj     *pUser = NULL;
2✔
3277

3278
  code = mndAcquireUser(pMnode, RPC_MSG_USER(pReq), &pUser);
2✔
3279
  if (pUser == NULL) return 0;
2✔
3280
  bool sysinfo = pUser->sysInfo;
2✔
3281

3282
  TAOS_CHECK_GOTO(tDeserializeSTableInfoReq(pReq->pCont, pReq->contLen, &infoReq), NULL, _OVER);
2✔
3283

3284
  if (0 == strcmp(infoReq.dbFName, TSDB_INFORMATION_SCHEMA_DB)) {
2✔
3285
    mInfo("information_schema table:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);
×
3286
    TAOS_CHECK_GOTO(mndBuildInsTableSchema(pMnode, infoReq.dbFName, infoReq.tbName, sysinfo, &metaRsp), NULL, _OVER);
×
3287
  } else if (0 == strcmp(infoReq.dbFName, TSDB_PERFORMANCE_SCHEMA_DB)) {
2✔
3288
    mInfo("performance_schema table:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);
×
3289
    TAOS_CHECK_GOTO(mndBuildPerfsTableSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp), NULL, _OVER);
×
3290
  } else {
3291
    mInfo("stb:%s.%s, start to retrieve meta", infoReq.dbFName, infoReq.tbName);
2✔
3292
    TAOS_CHECK_GOTO(mndBuildStbSchema(pMnode, infoReq.dbFName, infoReq.tbName, &metaRsp, true), NULL, _OVER);
2✔
3293
  }
3294

3295
  int32_t rspLen = tSerializeSTableMetaRsp(NULL, 0, &metaRsp);
2✔
3296
  if (rspLen < 0) {
2✔
3297
    code = TSDB_CODE_INVALID_MSG;
×
3298
    goto _OVER;
×
3299
  }
3300

3301
  void *pRsp = rpcMallocCont(rspLen);
2✔
3302
  if (pRsp == NULL) {
2✔
3303
    code = terrno;
×
3304
    goto _OVER;
×
3305
  }
3306

3307
  if ((rspLen = tSerializeSTableMetaRsp(pRsp, rspLen, &metaRsp)) < 0) {
2✔
3308
    code = rspLen;
×
3309
    goto _OVER;
×
3310
  }
3311
  pReq->info.rsp = pRsp;
2✔
3312
  pReq->info.rspLen = rspLen;
2✔
3313
  code = 0;
2✔
3314

3315
  mTrace("%s.%s, meta is retrieved", infoReq.dbFName, infoReq.tbName);
2✔
3316

3317
_OVER:
×
3318
  if (code != 0) {
2✔
3319
    mError("stb:%s.%s, failed to retrieve meta since %s", infoReq.dbFName, infoReq.tbName, tstrerror(code));
×
3320
  }
3321

3322
  mndReleaseUser(pMnode, pUser);
2✔
3323
  tFreeSTableMetaRsp(&metaRsp);
2✔
3324
  // TODO change to TAOS_RETURN
3325
  return code;
2✔
3326
}
3327

3328
static int32_t mndProcessTableCfgReq(SRpcMsg *pReq) {
×
3329
  SMnode      *pMnode = pReq->info.node;
×
3330
  int32_t      code = -1;
×
3331
  STableCfgReq cfgReq = {0};
×
3332
  STableCfgRsp cfgRsp = {0};
×
3333

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

3336
  char dbName[TSDB_DB_NAME_LEN] = {0};
×
3337
  TAOS_CHECK_GOTO(mndExtractShortDbNameFromDbFullName(cfgReq.dbFName, dbName), NULL, _OVER);
×
3338
  if (0 == strcmp(dbName, TSDB_INFORMATION_SCHEMA_DB)) {
×
3339
    mInfo("information_schema table:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
×
3340
    TAOS_CHECK_GOTO(mndBuildInsTableCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp), NULL, _OVER);
×
3341
  } else if (0 == strcmp(dbName, TSDB_PERFORMANCE_SCHEMA_DB)) {
×
3342
    mInfo("performance_schema table:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
×
3343
    TAOS_CHECK_GOTO(mndBuildPerfsTableCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp), NULL, _OVER);
×
3344
  } else {
3345
    mInfo("stb:%s.%s, start to retrieve cfg", cfgReq.dbFName, cfgReq.tbName);
×
3346
    TAOS_CHECK_GOTO(mndBuildStbCfg(pMnode, cfgReq.dbFName, cfgReq.tbName, &cfgRsp), NULL, _OVER);
×
3347
  }
3348

3349
  int32_t rspLen = tSerializeSTableCfgRsp(NULL, 0, &cfgRsp);
×
3350
  if (rspLen < 0) {
×
3351
    code = TSDB_CODE_INVALID_MSG;
×
3352
    goto _OVER;
×
3353
  }
3354

3355
  void *pRsp = rpcMallocCont(rspLen);
×
3356
  if (pRsp == NULL) {
×
3357
    code = terrno;
×
3358
    goto _OVER;
×
3359
  }
3360

3361
  if ((rspLen = tSerializeSTableCfgRsp(pRsp, rspLen, &cfgRsp)) < 0) {
×
3362
    code = rspLen;
×
3363
    goto _OVER;
×
3364
  }
3365
  pReq->info.rsp = pRsp;
×
3366
  pReq->info.rspLen = rspLen;
×
3367
  code = 0;
×
3368

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

3371
_OVER:
×
3372
  if (code != 0) {
×
3373
    mError("stb:%s.%s, failed to retrieve cfg since %s", cfgReq.dbFName, cfgReq.tbName, tstrerror(code));
×
3374
  }
3375

3376
  tFreeSTableCfgRsp(&cfgRsp);
×
3377
  TAOS_RETURN(code);
×
3378
}
3379

3380
int32_t mndValidateStbInfo(SMnode *pMnode, SSTableVersion *pStbVersions, int32_t numOfStbs, void **ppRsp,
×
3381
                           int32_t *pRspLen) {
3382
  int32_t   code = 0;
×
3383
  SSTbHbRsp hbRsp = {0};
×
3384
  hbRsp.pMetaRsp = taosArrayInit(numOfStbs, sizeof(STableMetaRsp));
×
3385
  if (hbRsp.pMetaRsp == NULL) {
×
3386
    code = terrno;
×
3387
    TAOS_RETURN(code);
×
3388
  }
3389

3390
  hbRsp.pIndexRsp = taosArrayInit(numOfStbs, sizeof(STableIndexRsp));
×
3391
  if (NULL == hbRsp.pIndexRsp) {
×
3392
    taosArrayDestroy(hbRsp.pMetaRsp);
×
3393
    code = terrno;
×
3394
    TAOS_RETURN(code);
×
3395
  }
3396

3397
  for (int32_t i = 0; i < numOfStbs; ++i) {
×
3398
    SSTableVersion *pStbVersion = &pStbVersions[i];
×
3399
    pStbVersion->suid = be64toh(pStbVersion->suid);
×
3400
    pStbVersion->sversion = ntohl(pStbVersion->sversion);
×
3401
    pStbVersion->tversion = ntohl(pStbVersion->tversion);
×
3402
    pStbVersion->smaVer = ntohl(pStbVersion->smaVer);
×
3403

3404
    bool    schema = false;
×
3405
    bool    sma = false;
×
3406
    int32_t code = mndValidateStbVersion(pMnode, pStbVersion, &schema, &sma);
×
3407
    if (TSDB_CODE_SUCCESS != code) {
×
3408
      STableMetaRsp metaRsp = {0};
×
3409
      metaRsp.numOfColumns = -1;
×
3410
      metaRsp.suid = pStbVersion->suid;
×
3411
      tstrncpy(metaRsp.dbFName, pStbVersion->dbFName, sizeof(metaRsp.dbFName));
×
3412
      tstrncpy(metaRsp.tbName, pStbVersion->stbName, sizeof(metaRsp.tbName));
×
3413
      tstrncpy(metaRsp.stbName, pStbVersion->stbName, sizeof(metaRsp.stbName));
×
3414
      if (taosArrayPush(hbRsp.pMetaRsp, &metaRsp) == NULL) {
×
3415
        code = terrno;
×
3416
        return code;
×
3417
      }
3418
      continue;
×
3419
    }
3420

3421
    if (schema) {
×
3422
      STableMetaRsp metaRsp = {0};
×
3423
      mInfo("stb:%s.%s, start to retrieve meta", pStbVersion->dbFName, pStbVersion->stbName);
×
3424
      if (mndBuildStbSchema(pMnode, pStbVersion->dbFName, pStbVersion->stbName, &metaRsp, false) != 0) {
×
3425
        metaRsp.numOfColumns = -1;
×
3426
        metaRsp.suid = pStbVersion->suid;
×
3427
        tstrncpy(metaRsp.dbFName, pStbVersion->dbFName, sizeof(metaRsp.dbFName));
×
3428
        tstrncpy(metaRsp.tbName, pStbVersion->stbName, sizeof(metaRsp.tbName));
×
3429
        tstrncpy(metaRsp.stbName, pStbVersion->stbName, sizeof(metaRsp.stbName));
×
3430
        if (taosArrayPush(hbRsp.pMetaRsp, &metaRsp) == NULL) {
×
3431
          code = terrno;
×
3432
          return code;
×
3433
        }
3434
        continue;
×
3435
      }
3436

3437
      if (taosArrayPush(hbRsp.pMetaRsp, &metaRsp) == NULL) {
×
3438
        code = terrno;
×
3439
        return code;
×
3440
      }
3441
    }
3442

3443
    if (sma) {
×
3444
      bool           exist = false;
×
3445
      char           tbFName[TSDB_TABLE_FNAME_LEN];
3446
      STableIndexRsp indexRsp = {0};
×
3447
      indexRsp.pIndex = taosArrayInit(10, sizeof(STableIndexInfo));
×
3448
      if (NULL == indexRsp.pIndex) {
×
3449
        code = terrno;
×
3450
        TAOS_RETURN(code);
×
3451
      }
3452

3453
      (void)tsnprintf(tbFName, sizeof(tbFName), "%s.%s", pStbVersion->dbFName, pStbVersion->stbName);
×
3454
      tstrncpy(indexRsp.dbFName, pStbVersion->dbFName, sizeof(indexRsp.dbFName));
×
3455
      tstrncpy(indexRsp.tbName, pStbVersion->stbName, sizeof(indexRsp.tbName));
×
3456

3457
      if (taosArrayPush(hbRsp.pIndexRsp, &indexRsp) == NULL) {
×
3458
        code = terrno;
×
3459
        return code;
×
3460
      }
3461
    }
3462
  }
3463

3464
  int32_t rspLen = tSerializeSSTbHbRsp(NULL, 0, &hbRsp);
×
3465
  if (rspLen < 0) {
×
3466
    tFreeSSTbHbRsp(&hbRsp);
×
3467
    code = TSDB_CODE_INVALID_MSG;
×
3468
    TAOS_RETURN(code);
×
3469
  }
3470

3471
  void *pRsp = taosMemoryMalloc(rspLen);
×
3472
  if (pRsp == NULL) {
×
3473
    tFreeSSTbHbRsp(&hbRsp);
×
3474
    code = terrno;
×
3475
    TAOS_RETURN(code);
×
3476
  }
3477

3478
  rspLen = tSerializeSSTbHbRsp(pRsp, rspLen, &hbRsp);
×
3479
  tFreeSSTbHbRsp(&hbRsp);
×
3480
  if (rspLen < 0) return rspLen;
×
3481
  *ppRsp = pRsp;
×
3482
  *pRspLen = rspLen;
×
3483
  TAOS_RETURN(code);
×
3484
}
3485

3486
int32_t mndGetNumOfStbs(SMnode *pMnode, char *dbName, int32_t *pNumOfStbs) {
18✔
3487
  int32_t code = 0;
18✔
3488
  SSdb   *pSdb = pMnode->pSdb;
18✔
3489
  SDbObj *pDb = mndAcquireDb(pMnode, dbName);
18✔
3490
  if (pDb == NULL) {
18✔
3491
    code = TSDB_CODE_MND_DB_NOT_SELECTED;
×
3492
    TAOS_RETURN(code);
×
3493
  }
3494

3495
  int32_t numOfStbs = 0;
18✔
3496
  void   *pIter = NULL;
18✔
3497
  while (1) {
×
3498
    SStbObj *pStb = NULL;
18✔
3499
    pIter = sdbFetch(pSdb, SDB_STB, pIter, (void **)&pStb);
18✔
3500
    if (pIter == NULL) break;
18✔
3501

3502
    if (pStb->dbUid == pDb->uid) {
×
3503
      numOfStbs++;
×
3504
    }
3505

3506
    sdbRelease(pSdb, pStb);
×
3507
  }
3508

3509
  *pNumOfStbs = numOfStbs;
18✔
3510
  mndReleaseDb(pMnode, pDb);
18✔
3511
  TAOS_RETURN(code);
18✔
3512
}
3513

3514
int32_t mndExtractDbNameFromStbFullName(const char *stbFullName, char *dst) {
×
3515
  SName name = {0};
×
3516
  TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
×
3517

3518
  TAOS_CHECK_RETURN(tNameGetFullDbName(&name, dst));
×
3519

3520
  return 0;
×
3521
}
3522

3523
int32_t mndExtractShortDbNameFromStbFullName(const char *stbFullName, char *dst) {
×
3524
  SName name = {0};
×
3525
  TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));
×
3526

3527
  TAOS_CHECK_RETURN(tNameGetDbName(&name, dst));
×
3528

3529
  return 0;
×
3530
}
3531

3532
int32_t mndExtractShortDbNameFromDbFullName(const char *stbFullName, char *dst) {
×
3533
  SName name = {0};
×
3534
  TAOS_CHECK_RETURN(tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB));
×
3535

3536
  TAOS_CHECK_RETURN(tNameGetDbName(&name, dst));
×
3537

3538
  return 0;
×
3539
}
3540

3541
void mndExtractTbNameFromStbFullName(const char *stbFullName, char *dst, int32_t dstSize) {
22✔
3542
  int32_t pos = -1;
22✔
3543
  int32_t num = 0;
22✔
3544
  for (pos = 0; stbFullName[pos] != 0; ++pos) {
110✔
3545
    if (stbFullName[pos] == TS_PATH_DELIMITER[0]) num++;
110✔
3546
    if (num == 2) break;
110✔
3547
  }
3548

3549
  if (num == 2) {
22✔
3550
    tstrncpy(dst, stbFullName + pos + 1, dstSize);
22✔
3551
  }
3552
}
22✔
3553

3554
static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
26✔
3555
  SMnode    *pMnode = pReq->info.node;
26✔
3556
  SSdb      *pSdb = pMnode->pSdb;
26✔
3557
  int32_t    numOfRows = 0;
26✔
3558
  SStbObj   *pStb = NULL;
26✔
3559
  SUserObj  *pOperUser = NULL;
26✔
3560
  SSHashObj *pUidNames = NULL;
26✔
3561
  int32_t    cols = 0;
26✔
3562
  int32_t    lino = 0;
26✔
3563
  int32_t    code = 0;
26✔
3564
  char       objFName[TSDB_OBJ_FNAME_LEN + 1] = {0};
26✔
3565
  bool       showAll = false;
26✔
3566

3567
  SDbObj *pDb = NULL;
26✔
3568
  if (strlen(pShow->db) > 0) {
26✔
3569
    pDb = mndAcquireDb(pMnode, pShow->db);
26✔
3570
    if (pDb == NULL) return terrno;
26✔
3571
  }
3572

3573
  if ((code = mndAcquireUser(pMnode, RPC_MSG_USER(pReq), &pOperUser)) != 0) {
26✔
3574
    goto _ERROR;
×
3575
  }
3576

3577
  (void)snprintf(objFName, sizeof(objFName), "%d.*", pOperUser->acctId);
26✔
3578
  showAll = (0 == mndCheckObjPrivilegeRecF(pMnode, pOperUser, PRIV_CM_SHOW, PRIV_OBJ_TBL, pDb ? pDb->ownerId : 0,
26✔
3579
                                           pDb ? pDb->name : objFName, "*"));
3580
  showAll = showAll && (0 == mndCheckDbPrivilegeByName(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_USE_DB,
26✔
3581
                                                       pDb ? pDb->name : objFName, false));
3582

3583
  while (numOfRows < rows) {
48✔
3584
    pShow->pIter = sdbFetch(pSdb, SDB_STB, pShow->pIter, (void **)&pStb);
48✔
3585
    if (pShow->pIter == NULL) break;
48✔
3586

3587
    if (pDb != NULL && pStb->dbUid != pDb->uid) {
22✔
3588
      sdbRelease(pSdb, pStb);
×
3589
      continue;
×
3590
    }
3591

3592
    if (isTsmaResSTb(pStb->name)) {
22✔
3593
      sdbRelease(pSdb, pStb);
×
3594
      continue;
×
3595
    }
3596

3597
#if 0
3598
    if ((0 == pUser->superUser) && mndCheckStbPrivilege(pMnode, pUser, RPC_MSG_TOKEN(pReq), MND_OPER_SHOW_STB, pStb) != 0) {
3599
      sdbRelease(pSdb, pStb);
3600
      terrno = 0;
3601
      continue;
3602
    }
3603
#endif
3604
    cols = 0;
22✔
3605

3606
    SName name = {0};
22✔
3607

3608
    char stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
22✔
3609
    mndExtractTbNameFromStbFullName(pStb->name, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN);
22✔
3610

3611
    if (!showAll && (mndCheckObjPrivilegeRecF(pMnode, pOperUser, PRIV_CM_SHOW, PRIV_OBJ_TBL, pStb->ownerId, pStb->db,
22✔
3612
                                              &stbName[VARSTR_HEADER_SIZE]) ||
×
3613
                     mndCheckDbPrivilegeByName(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_USE_DB,
×
3614
                                               pStb->db, false))) {
×
3615
      sdbRelease(pSdb, pStb);
×
3616
      terrno = 0;
×
3617
      continue;
×
3618
    }
3619

3620
    varDataSetLen(stbName, strlen(&stbName[VARSTR_HEADER_SIZE]));
22✔
3621
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
22✔
3622
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbName, false), pStb, &lino, _ERROR);
22✔
3623

3624
    char db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
22✔
3625
    RETRIEVE_CHECK_GOTO(tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB), pStb, &lino, _ERROR);
22✔
3626
    RETRIEVE_CHECK_GOTO(tNameGetDbName(&name, varDataVal(db)), pStb, &lino, _ERROR);
22✔
3627
    varDataSetLen(db, strlen(varDataVal(db)));
22✔
3628
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
22✔
3629
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)db, false), pStb, &lino, _ERROR);
22✔
3630

3631
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
22✔
3632
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->createdTime, false), pStb, &lino,
22✔
3633
                        _ERROR);
3634

3635
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
22✔
3636
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->numOfColumns, false), pStb, &lino,
22✔
3637
                        _ERROR);
3638

3639
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
22✔
3640
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->numOfTags, false), pStb, &lino, _ERROR);
22✔
3641

3642
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
22✔
3643
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->updateTime, false), pStb, &lino,
22✔
3644
                        _ERROR);  // number of tables
3645

3646
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
22✔
3647
    if (pStb->commentLen > 0) {
22✔
3648
      char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0};
×
3649
      STR_TO_VARSTR(comment, pStb->comment);
×
3650
      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, comment, false), pStb, &lino, _ERROR);
×
3651
    } else if (pStb->commentLen == 0) {
22✔
3652
      char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};
22✔
3653
      STR_TO_VARSTR(comment, "");
22✔
3654
      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, comment, false), pStb, &lino, _ERROR);
22✔
3655
    } else {
3656
      colDataSetNULL(pColInfo, numOfRows);
×
3657
    }
3658

3659
    char watermark[64 + VARSTR_HEADER_SIZE] = {0};
22✔
3660
    (void)tsnprintf(varDataVal(watermark), sizeof(watermark) - VARSTR_HEADER_SIZE, "%" PRId64 "a,%" PRId64 "a",
22✔
3661
              pStb->watermark[0], pStb->watermark[1]);
22✔
3662
    varDataSetLen(watermark, strlen(varDataVal(watermark)));
22✔
3663

3664
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
22✔
3665
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)watermark, false), pStb, &lino, _ERROR);
22✔
3666

3667
    char maxDelay[64 + VARSTR_HEADER_SIZE] = {0};
22✔
3668
    (void)tsnprintf(varDataVal(maxDelay), sizeof(maxDelay) - VARSTR_HEADER_SIZE, "%" PRId64 "a,%" PRId64 "a",
22✔
3669
              pStb->maxdelay[0], pStb->maxdelay[1]);
22✔
3670
    varDataSetLen(maxDelay, strlen(varDataVal(maxDelay)));
22✔
3671

3672
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
22✔
3673
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)maxDelay, false), pStb, &lino, _ERROR);
22✔
3674

3675
    char    rollup[160 + VARSTR_HEADER_SIZE] = {0};
22✔
3676
    int32_t rollupNum = (int32_t)taosArrayGetSize(pStb->pFuncs);
22✔
3677
    char   *sep = ", ";
22✔
3678
    int32_t sepLen = strlen(sep);
22✔
3679
    int32_t rollupLen = sizeof(rollup) - VARSTR_HEADER_SIZE - 2;
22✔
3680
    for (int32_t i = 0; i < rollupNum; ++i) {
22✔
3681
      char *funcName = taosArrayGet(pStb->pFuncs, i);
×
3682
      if (i) {
×
3683
        (void)strncat(varDataVal(rollup), sep, rollupLen);
×
3684
        rollupLen -= sepLen;
×
3685
      }
3686
      (void)strncat(varDataVal(rollup), funcName, rollupLen);
×
3687
      rollupLen -= strlen(funcName);
×
3688
    }
3689
    varDataSetLen(rollup, strlen(varDataVal(rollup)));
22✔
3690

3691
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
22✔
3692
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)rollup, false), pStb, &lino, _ERROR);
22✔
3693

3694
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
22✔
3695
    if (pColInfo) {
22✔
3696
      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)(&pStb->uid), false), pStb, &lino, _ERROR);
22✔
3697
    }
3698

3699
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
22✔
3700
    if (pColInfo) {
22✔
3701
      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)(&pStb->virtualStb), false), pStb, &lino, _ERROR);
22✔
3702
    }
3703

3704
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
22✔
3705
    if (pColInfo) {
22✔
3706
      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)(&pStb->keep), false), pStb, &lino, _ERROR);
22✔
3707
    }
3708

3709
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
22✔
3710
    if (pColInfo) {
22✔
3711
      if (!pUidNames) {
22✔
3712
        TAOS_CHECK_GOTO(mndBuildUidNamesHash(pMnode, &pUidNames), &lino, _OVER);
22✔
3713
      }
3714
      const char *ownerName = tSimpleHashGet(pUidNames, (const char *)&pStb->ownerId, sizeof(pStb->ownerId));
22✔
3715
      char        owner[TSDB_USER_LEN + VARSTR_HEADER_SIZE] = {0};
22✔
3716
      STR_WITH_MAXSIZE_TO_VARSTR(owner, ownerName ? ownerName : "[unknown]", sizeof(owner));
22✔
3717
      RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)owner, false), pStb, &lino, _ERROR);
22✔
3718
    }
3719

3720
    numOfRows++;
22✔
3721
    sdbRelease(pSdb, pStb);
22✔
3722
  }
3723

3724
  if (pDb != NULL) {
26✔
3725
    mndReleaseDb(pMnode, pDb);
26✔
3726
  }
3727
  if (pOperUser != NULL) {
26✔
3728
    mndReleaseUser(pMnode, pOperUser);
26✔
3729
  }
3730

3731
  goto _OVER;
26✔
3732

3733
_ERROR:
×
3734
  if (pDb != NULL) {
×
3735
    mndReleaseDb(pMnode, pDb);
×
3736
  }
3737
  if (pOperUser != NULL) {
×
3738
    mndReleaseUser(pMnode, pOperUser);
×
3739
  }
3740
  mError("show:0x%" PRIx64 ", failed to retrieve data at %s:%d since %s", pShow->id, __FUNCTION__, lino,
×
3741
         tstrerror(code));
3742

3743
_OVER:
×
3744
  pShow->numOfRows += numOfRows;
26✔
3745
  tSimpleHashCleanup(pUidNames);
26✔
3746
  return numOfRows;
26✔
3747
}
3748

3749
static int32_t buildDbColsInfoBlock(const SSDataBlock *p, const SSysTableMeta *pSysDbTableMeta, size_t size,
×
3750
                                    const char *dbName, const char *tbName) {
3751
  char    tName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3752
  char    dName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3753
  char    typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3754
  int32_t numOfRows = p->info.rows;
×
3755
  int32_t lino = 0;
×
3756
  int32_t code = 0;
×
3757

3758
  STR_TO_VARSTR(dName, dbName);
×
3759
  STR_TO_VARSTR(typeName, "SYSTEM_TABLE");
×
3760

3761
  for (int32_t i = 0; i < size; ++i) {
×
3762
    const SSysTableMeta *pm = &pSysDbTableMeta[i];
×
3763
    //    if (pm->sysInfo) {
3764
    //      continue;
3765
    //    }
3766
    if (tbName[0] && strncmp(tbName, pm->name, TSDB_TABLE_NAME_LEN) != 0) {
×
3767
      continue;
×
3768
    }
3769

3770
    STR_TO_VARSTR(tName, pm->name);
×
3771

3772
    for (int32_t j = 0; j < pm->colNum; j++) {
×
3773
      // table name
3774
      SColumnInfoData *pColInfoData = taosArrayGet(p->pDataBlock, 0);
×
3775
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, tName, false), &lino, _OVER);
×
3776

3777
      // database name
3778
      pColInfoData = taosArrayGet(p->pDataBlock, 1);
×
3779
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, dName, false), &lino, _OVER);
×
3780

3781
      pColInfoData = taosArrayGet(p->pDataBlock, 2);
×
3782
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, typeName, false), &lino, _OVER);
×
3783

3784
      // col name
3785
      char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3786
      STR_TO_VARSTR(colName, pm->schema[j].name);
×
3787
      pColInfoData = taosArrayGet(p->pDataBlock, 3);
×
3788
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, colName, false), &lino, _OVER);
×
3789

3790
      // col type
3791
      int8_t colType = pm->schema[j].type;
×
3792
      pColInfoData = taosArrayGet(p->pDataBlock, 4);
×
3793
      char colTypeStr[VARSTR_HEADER_SIZE + 32];
3794
      int  colTypeLen =
×
3795
          tsnprintf(varDataVal(colTypeStr), sizeof(colTypeStr) - VARSTR_HEADER_SIZE, "%s", tDataTypes[colType].name);
×
3796
      if (colType == TSDB_DATA_TYPE_VARCHAR) {
×
3797
        colTypeLen +=
×
3798
            tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE, "(%d)",
×
3799
                      (int32_t)(pm->schema[j].bytes - VARSTR_HEADER_SIZE));
×
3800
      } else if (colType == TSDB_DATA_TYPE_NCHAR) {
×
3801
        colTypeLen +=
×
3802
            tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE, "(%d)",
×
3803
                      (int32_t)((pm->schema[j].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
×
3804
      }
3805
      varDataSetLen(colTypeStr, colTypeLen);
×
3806
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, (char *)colTypeStr, false), &lino, _OVER);
×
3807

3808
      // col length
3809
      pColInfoData = taosArrayGet(p->pDataBlock, 5);
×
3810
      TAOS_CHECK_GOTO(colDataSetVal(pColInfoData, numOfRows, (const char *)&pm->schema[j].bytes, false), &lino, _OVER);
×
3811

3812
      // col precision, col scale, col nullable, col source
3813
      for (int32_t k = 6; k <= 10; ++k) {
×
3814
        pColInfoData = taosArrayGet(p->pDataBlock, k);
×
3815
        colDataSetNULL(pColInfoData, numOfRows);
×
3816
      }
3817

3818
      numOfRows += 1;
×
3819
    }
3820
  }
3821
  return numOfRows;
×
3822
_OVER:
×
3823
  mError("failed at %s:%d since %s", __FUNCTION__, lino, tstrerror(code));
×
3824
  return numOfRows;
×
3825
}
3826
#define BUILD_COL_FOR_INFO_DB 1
3827
#define BUILD_COL_FOR_PERF_DB 1 << 1
3828
#define BUILD_COL_FOR_USER_DB 1 << 2
3829
#define BUILD_COL_FOR_ALL_DB  (BUILD_COL_FOR_INFO_DB | BUILD_COL_FOR_PERF_DB | BUILD_COL_FOR_USER_DB)
3830

3831
static int32_t buildSysDbColsInfo(SSDataBlock *p, int8_t buildWhichDBs, char *tb) {
×
3832
  size_t               size = 0;
×
3833
  const SSysTableMeta *pSysDbTableMeta = NULL;
×
3834

3835
  if (buildWhichDBs & BUILD_COL_FOR_INFO_DB) {
×
3836
    getInfosDbMeta(&pSysDbTableMeta, &size);
×
3837
    p->info.rows = buildDbColsInfoBlock(p, pSysDbTableMeta, size, TSDB_INFORMATION_SCHEMA_DB, tb);
×
3838
  }
3839

3840
  if (buildWhichDBs & BUILD_COL_FOR_PERF_DB) {
×
3841
    getPerfDbMeta(&pSysDbTableMeta, &size);
×
3842
    p->info.rows = buildDbColsInfoBlock(p, pSysDbTableMeta, size, TSDB_PERFORMANCE_SCHEMA_DB, tb);
×
3843
  }
3844

3845
  return p->info.rows;
×
3846
}
3847

3848
static int8_t determineBuildColForWhichDBs(const char *db) {
×
3849
  int8_t buildWhichDBs;
3850
  if (!db[0])
×
3851
    buildWhichDBs = BUILD_COL_FOR_ALL_DB;
×
3852
  else {
3853
    char *p = strchr(db, '.');
×
3854
    if (p && strcmp(p + 1, TSDB_INFORMATION_SCHEMA_DB) == 0) {
×
3855
      buildWhichDBs = BUILD_COL_FOR_INFO_DB;
×
3856
    } else if (p && strcmp(p + 1, TSDB_PERFORMANCE_SCHEMA_DB) == 0) {
×
3857
      buildWhichDBs = BUILD_COL_FOR_PERF_DB;
×
3858
    } else {
3859
      buildWhichDBs = BUILD_COL_FOR_USER_DB;
×
3860
    }
3861
  }
3862
  return buildWhichDBs;
×
3863
}
3864

3865
static int32_t mndRetrieveStbCol(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
×
3866
  uint8_t  buildWhichDBs;
3867
  SMnode  *pMnode = pReq->info.node;
×
3868
  SSdb    *pSdb = pMnode->pSdb;
×
3869
  SStbObj *pStb = NULL;
×
3870
  int32_t  numOfRows = 0;
×
3871
  int32_t  lino = 0;
×
3872
  int32_t  code = 0;
×
3873

3874
  buildWhichDBs = determineBuildColForWhichDBs(pShow->db);
×
3875

3876
  if (!pShow->sysDbRsp) {
×
3877
    numOfRows = buildSysDbColsInfo(pBlock, buildWhichDBs, pShow->filterTb);
×
3878
    mDebug("mndRetrieveStbCol get system table cols, rows:%d, db:%s", numOfRows, pShow->db);
×
3879
    pShow->sysDbRsp = true;
×
3880
  }
3881

3882
  if (buildWhichDBs & BUILD_COL_FOR_USER_DB) {
×
3883
    SDbObj *pDb = NULL;
×
3884
    if (strlen(pShow->db) > 0) {
×
3885
      pDb = mndAcquireDb(pMnode, pShow->db);
×
3886
      if (pDb == NULL && TSDB_CODE_MND_DB_NOT_EXIST != terrno && pBlock->info.rows == 0) return terrno;
×
3887
    }
3888

3889
    char typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3890
    STR_TO_VARSTR(typeName, "SUPER_TABLE");
×
3891
    bool fetch = pShow->restore ? false : true;
×
3892
    pShow->restore = false;
×
3893
    while (numOfRows < rows) {
×
3894
      if (fetch) {
×
3895
        pShow->pIter = sdbFetch(pSdb, SDB_STB, pShow->pIter, (void **)&pStb);
×
3896
        if (pShow->pIter == NULL) break;
×
3897
      } else {
3898
        fetch = true;
×
3899
        void *pKey = taosHashGetKey(pShow->pIter, NULL);
×
3900
        pStb = sdbAcquire(pSdb, SDB_STB, pKey);
×
3901
        if (!pStb) continue;
×
3902
      }
3903

3904
      if (pDb != NULL && pStb->dbUid != pDb->uid) {
×
3905
        sdbRelease(pSdb, pStb);
×
3906
        continue;
×
3907
      }
3908

3909
      SName name = {0};
×
3910
      char  stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3911
      mndExtractTbNameFromStbFullName(pStb->name, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN);
×
3912
      if (pShow->filterTb[0] && strncmp(pShow->filterTb, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN) != 0) {
×
3913
        sdbRelease(pSdb, pStb);
×
3914
        continue;
×
3915
      }
3916

3917
      if ((numOfRows + pStb->numOfColumns) > rows) {
×
3918
        pShow->restore = true;
×
3919
        if (numOfRows == 0) {
×
3920
          mError("mndRetrieveStbCol failed to get stable cols since buf:%d less than result:%d, stable name:%s, db:%s",
×
3921
                 rows, pStb->numOfColumns, pStb->name, pStb->db);
3922
        }
3923
        sdbRelease(pSdb, pStb);
×
3924
        break;
×
3925
      }
3926

3927
      varDataSetLen(stbName, strlen(&stbName[VARSTR_HEADER_SIZE]));
×
3928

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

3931
      char db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3932
      RETRIEVE_CHECK_GOTO(tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB), pStb, &lino, _OVER);
×
3933
      RETRIEVE_CHECK_GOTO(tNameGetDbName(&name, varDataVal(db)), pStb, &lino, _OVER);
×
3934
      varDataSetLen(db, strlen(varDataVal(db)));
×
3935

3936
      for (int i = 0; i < pStb->numOfColumns; i++) {
×
3937
        int32_t          cols = 0;
×
3938
        SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
3939
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)stbName, false), pStb, &lino, _OVER);
×
3940

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

3944
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
3945
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, typeName, false), pStb, &lino, _OVER);
×
3946

3947
        // col name
3948
        char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
×
3949
        STR_TO_VARSTR(colName, pStb->pColumns[i].name);
×
3950
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
3951
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, colName, false), pStb, &lino, _OVER);
×
3952

3953
        // col type
3954
        int8_t colType = pStb->pColumns[i].type;
×
3955
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
3956
        char colTypeStr[VARSTR_HEADER_SIZE + 32];
3957
        int  colTypeLen =
×
3958
            tsnprintf(varDataVal(colTypeStr), sizeof(colTypeStr) - VARSTR_HEADER_SIZE, "%s", tDataTypes[colType].name);
×
3959
        if (colType == TSDB_DATA_TYPE_VARCHAR) {
×
3960
          colTypeLen +=
×
3961
              tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE,
×
3962
                        "(%d)", (int32_t)(pStb->pColumns[i].bytes - VARSTR_HEADER_SIZE));
×
3963
        } else if (colType == TSDB_DATA_TYPE_NCHAR) {
×
3964
          colTypeLen +=
×
3965
              tsnprintf(varDataVal(colTypeStr) + colTypeLen, sizeof(colTypeStr) - colTypeLen - VARSTR_HEADER_SIZE,
×
3966
                        "(%d)", (int32_t)((pStb->pColumns[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
×
3967
        } else if (IS_DECIMAL_TYPE(colType)) {
×
3968
          STypeMod typeMod = pStb->pExtSchemas[i].typeMod;
×
3969
          uint8_t prec = 0, scale = 0;
×
3970
          decimalFromTypeMod(typeMod, &prec, &scale);
×
3971
          colTypeLen += sprintf(varDataVal(colTypeStr) + colTypeLen, "(%d,%d)", prec, scale);
×
3972
        }
3973
        varDataSetLen(colTypeStr, colTypeLen);
×
3974
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (char *)colTypeStr, false), pStb, &lino, _OVER);
×
3975

3976
        // col length
3977
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
×
3978
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->pColumns[i].bytes, false), pStb,
×
3979
                            &lino, _OVER);
3980
        
3981
        // col precision, col scale, col nullable, col source
3982
        for (int32_t j = 6; j <= 9; ++j) {
×
3983
          pColInfo = taosArrayGet(pBlock->pDataBlock, j);
×
3984
          colDataSetNULL(pColInfo, numOfRows);
×
3985
        }
3986

3987
        // col id
3988
        pColInfo = taosArrayGet(pBlock->pDataBlock, 10);
×
3989
        RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pStb->pColumns[i].colId, false), pStb,
×
3990
                            &lino, _OVER);
3991
        numOfRows++;
×
3992
      }
3993

3994
      sdbRelease(pSdb, pStb);
×
3995
    }
3996

3997
    if (pDb != NULL) {
×
3998
      mndReleaseDb(pMnode, pDb);
×
3999
    }
4000
  }
4001

4002
  mDebug("mndRetrieveStbCol success, rows:%d, pShow->numOfRows:%d", numOfRows, pShow->numOfRows);
×
4003
  goto _OVER;
×
4004

4005
_ERROR:
4006
  mError("failed to mndRetrieveStbCol, rows:%d, pShow->numOfRows:%d, at %s:%d since %s", numOfRows, pShow->numOfRows,
4007
         __FUNCTION__, lino, tstrerror(code));
4008

4009
_OVER:
×
4010
  pShow->numOfRows += numOfRows;
×
4011
  return numOfRows;
×
4012
}
4013

4014
static void mndCancelGetNextStb(SMnode *pMnode, void *pIter) {
×
4015
  SSdb *pSdb = pMnode->pSdb;
×
4016
  sdbCancelFetchByType(pSdb, pIter, SDB_STB);
×
4017
}
×
4018

4019
const char *mndGetStbStr(const char *src) {
×
4020
  char *posDb = strstr(src, TS_PATH_DELIMITER);
×
4021
  if (posDb != NULL) ++posDb;
×
4022
  if (posDb == NULL) return src;
×
4023

4024
  char *posStb = strstr(posDb, TS_PATH_DELIMITER);
×
4025
  if (posStb != NULL) ++posStb;
×
4026
  if (posStb == NULL) return posDb;
×
4027
  return posStb;
×
4028
}
4029

4030
static int32_t mndCheckIndexReq(SCreateTagIndexReq *pReq) {
×
4031
  // impl
4032
  return TSDB_CODE_SUCCESS;
×
4033
}
4034

4035
/*int32_t mndAddIndexImpl(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp, void *sql,
4036
                        int32_t len) {
4037
  // impl later
4038
  int32_t code = 0;
4039
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB_INSIDE, pReq, "create-stb-index");
4040
  if (pTrans == NULL) goto _OVER;
4041

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

4046
  if (mndSetAlterStbPrepareLogs(pMnode, pTrans, pDb, pStb) != 0) goto _OVER;
4047
  if (mndSetAlterStbCommitLogs(pMnode, pTrans, pDb, pStb) != 0) goto _OVER;
4048
  if (mndSetAlterStbRedoActions2(pMnode, pTrans, pDb, pStb, sql, len) != 0) goto _OVER;
4049
  if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
4050

4051
  return code;
4052

4053
_OVER:
4054
  mndTransDrop(pTrans);
4055
  return code;
4056
}
4057
static int32_t mndAddIndex(SMnode *pMnode, SRpcMsg *pReq, SCreateTagIndexReq *tagIdxReq, SDbObj *pDb, SStbObj *pOld) {
4058
  bool    needRsp = true;
4059
  int32_t code = -1;
4060
  SField *pField0 = NULL;
4061

4062
  SStbObj  stbObj = {0};
4063
  SStbObj *pNew = &stbObj;
4064

4065
  taosRLockLatch(&pOld->lock);
4066
  memcpy(&stbObj, pOld, sizeof(SStbObj));
4067
  taosRUnLockLatch(&pOld->lock);
4068

4069
  stbObj.pColumns = NULL;
4070
  stbObj.pTags = NULL;
4071
  stbObj.updateTime = taosGetTimestampMs();
4072
  stbObj.lock = 0;
4073

4074
  int32_t tag = mndFindSuperTableTagIndex(pOld, tagIdxReq->colName);
4075
  if (tag < 0) {
4076
    terrno = TSDB_CODE_MND_TAG_NOT_EXIST;
4077
    return -1;
4078
  }
4079
  if (mndAllocStbSchemas(pOld, pNew) != 0) {
4080
    return -1;
4081
  }
4082

4083
  SSchema *pTag = pNew->pTags + tag;
4084
  if (IS_IDX_ON(pTag)) {
4085
    terrno = TSDB_CODE_MND_TAG_INDEX_ALREADY_EXIST;
4086
    return -1;
4087
  } else {
4088
    pTag->flags |= COL_IDX_ON;
4089
  }
4090
  pNew->tagVer++;
4091

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

4094
  return code;
4095
}
4096
static int32_t mndProcessCreateIndexReq(SRpcMsg *pReq) {
4097
  SMnode            *pMnode = pReq->info.node;
4098
  int32_t            code = -1;
4099
  SDbObj            *pDb = NULL;
4100
  SStbObj           *pStb = NULL;
4101
  SCreateTagIndexReq tagIdxReq = {0};
4102

4103
  if (tDeserializeSCreateTagIdxReq(pReq->pCont, pReq->contLen, &tagIdxReq) != 0) {
4104
    terrno = TSDB_CODE_INVALID_MSG;
4105
    goto _OVER;
4106
  }
4107

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

4110
  if (mndCheckIndexReq(&tagIdxReq) != TSDB_CODE_SUCCESS) {
4111
    goto _OVER;
4112
  }
4113

4114
  pDb = mndAcquireDbByStb(pMnode, tagIdxReq.dbFName);
4115
  if (pDb == NULL) {
4116
    terrno = TSDB_CODE_MND_DB_NOT_EXIST;
4117
    goto _OVER;
4118
  }
4119

4120
  pStb = mndAcquireStb(pMnode, tagIdxReq.stbName);
4121
  if (pStb == NULL) {
4122
    terrno = TSDB_CODE_MND_STB_NOT_EXIST;
4123
    goto _OVER;
4124
  }
4125
  if (mndCheckDbPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_WRITE_DB, pDb) != 0) {
4126
    goto _OVER;
4127
  }
4128

4129
  code = mndAddIndex(pMnode, pReq, &tagIdxReq, pDb, pStb);
4130
  if (terrno == TSDB_CODE_MND_TAG_INDEX_ALREADY_EXIST || terrno == TSDB_CODE_MND_TAG_NOT_EXIST) {
4131
    return terrno;
4132
  } else {
4133
    if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
4134
  }
4135
_OVER:
4136
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
4137
    mError("stb:%s, failed to create index since %s", tagIdxReq.stbName, terrstr());
4138
  }
4139
  mndReleaseStb(pMnode, pStb);
4140
  mndReleaseDb(pMnode, pDb);
4141
  return code;
4142
}
4143
static int32_t mndProcessDropIndexReq(SRpcMsg *pReq) {
4144
  SMnode          *pMnode = pReq->info.node;
4145
  int32_t          code = -1;
4146
  SDbObj          *pDb = NULL;
4147
  SStbObj         *pStb = NULL;
4148
  SDropTagIndexReq dropReq = {0};
4149
  if (tDeserializeSDropTagIdxReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
4150
    terrno = TSDB_CODE_INVALID_MSG;
4151
    goto _OVER;
4152
  }
4153
  //
4154
  return TSDB_CODE_SUCCESS;
4155
_OVER:
4156
  return code;
4157
}*/
4158

4159
static int32_t mndProcessDropStbReqFromMNode(SRpcMsg *pReq) {
×
4160
  int32_t code = mndProcessDropStbReq(pReq);
×
4161
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
×
4162
    pReq->info.rsp = rpcMallocCont(1);
×
4163
    pReq->info.rspLen = 1;
×
4164
    pReq->info.noResp = false;
×
4165
    pReq->code = code;
×
4166
  }
4167
  return code;
×
4168
}
4169

4170
typedef struct SVDropTbVgReqs {
4171
  SArray     *pBatchReqs;
4172
  SVgroupInfo info;
4173
} SVDropTbVgReqs;
4174

4175
typedef struct SMDropTbDbInfo {
4176
  SArray *dbVgInfos;
4177
  int32_t hashPrefix;
4178
  int32_t hashSuffix;
4179
  int32_t hashMethod;
4180
} SMDropTbDbInfo;
4181

4182
typedef struct SMDropTbTsmaInfo {
4183
  char           tsmaResTbDbFName[TSDB_DB_FNAME_LEN];
4184
  char           tsmaResTbNamePrefix[TSDB_TABLE_FNAME_LEN];
4185
  int32_t        suid;
4186
  SMDropTbDbInfo dbInfo;  // reference to DbInfo in pDbMap
4187
} SMDropTbTsmaInfo;
4188

4189
typedef struct SMDropTbTsmaInfos {
4190
  SArray *pTsmaInfos;  // SMDropTbTsmaInfo
4191
} SMDropTbTsmaInfos;
4192

4193
typedef struct SMndDropTbsWithTsmaCtx {
4194
  SHashObj *pVgMap;  // <vgId, SVDropTbVgReqs>
4195
} SMndDropTbsWithTsmaCtx;
4196

4197
static int32_t mndDropTbForSingleVg(SMnode *pMnode, SMndDropTbsWithTsmaCtx *pCtx, SArray *pTbs, int32_t vgId);
4198

4199
static void destroySVDropTbBatchReqs(void *p);
4200
static void mndDestroyDropTbsWithTsmaCtx(SMndDropTbsWithTsmaCtx *p) {
×
4201
  if (!p) return;
×
4202

4203
  if (p->pVgMap) {
×
4204
    void *pIter = taosHashIterate(p->pVgMap, NULL);
×
4205
    while (pIter) {
×
4206
      SVDropTbVgReqs *pReqs = pIter;
×
4207
      taosArrayDestroyEx(pReqs->pBatchReqs, destroySVDropTbBatchReqs);
×
4208
      pIter = taosHashIterate(p->pVgMap, pIter);
×
4209
    }
4210
    taosHashCleanup(p->pVgMap);
×
4211
  }
4212
  taosMemoryFree(p);
×
4213
}
4214

4215
static int32_t mndInitDropTbsWithTsmaCtx(SMndDropTbsWithTsmaCtx **ppCtx) {
×
4216
  int32_t                 code = 0;
×
4217
  SMndDropTbsWithTsmaCtx *pCtx = taosMemoryCalloc(1, sizeof(SMndDropTbsWithTsmaCtx));
×
4218
  if (!pCtx) return terrno;
×
4219

4220
  pCtx->pVgMap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
×
4221
  if (!pCtx->pVgMap) {
×
4222
    code = terrno;
×
4223
    goto _end;
×
4224
  }
4225

4226
  *ppCtx = pCtx;
×
4227
_end:
×
4228
  if (code) mndDestroyDropTbsWithTsmaCtx(pCtx);
×
4229
  return code;
×
4230
}
4231

4232
static void *mndBuildVDropTbsReq(SMnode *pMnode, const SVgroupInfo *pVgInfo, const SVDropTbBatchReq *pReq,
×
4233
                                 int32_t *len) {
4234
  int32_t   contLen = 0;
×
4235
  int32_t   ret = 0;
×
4236
  SMsgHead *pHead = NULL;
×
4237
  SEncoder  encoder = {0};
×
4238

4239
  tEncodeSize(tEncodeSVDropTbBatchReq, pReq, contLen, ret);
×
4240
  if (ret < 0) return NULL;
×
4241

4242
  contLen += sizeof(SMsgHead);
×
4243
  pHead = taosMemoryMalloc(contLen);
×
4244
  if (pHead == NULL) {
×
4245
    terrno = TSDB_CODE_OUT_OF_MEMORY;
×
4246
    return NULL;
×
4247
  }
4248

4249
  pHead->contLen = htonl(contLen);
×
4250
  pHead->vgId = htonl(pVgInfo->vgId);
×
4251

4252
  void *pBuf = POINTER_SHIFT(pHead, sizeof(SMsgHead));
×
4253

4254
  tEncoderInit(&encoder, pBuf, contLen - sizeof(SMsgHead));
×
4255
  int32_t code = tEncodeSVDropTbBatchReq(&encoder, pReq);
×
4256
  tEncoderClear(&encoder);
×
4257
  if (code != 0) return NULL;
×
4258

4259
  *len = contLen;
×
4260
  return pHead;
×
4261
}
4262

4263
static int32_t mndSetDropTbsRedoActions(SMnode *pMnode, STrans *pTrans, const SVDropTbVgReqs *pVgReqs, void *pCont,
×
4264
                                        int32_t contLen, tmsg_t msgType) {
4265
  STransAction action = {0};
×
4266
  action.epSet = pVgReqs->info.epSet;
×
4267
  action.pCont = pCont;
×
4268
  action.contLen = contLen;
×
4269
  action.msgType = msgType;
×
4270
  action.acceptableCode = TSDB_CODE_TDB_TABLE_NOT_EXIST;
×
4271
  return mndTransAppendRedoAction(pTrans, &action);
×
4272
}
4273

4274
static int32_t mndBuildDropTbRedoActions(SMnode *pMnode, STrans *pTrans, SHashObj *pVgMap, tmsg_t msgType) {
×
4275
  int32_t code = 0;
×
4276
  void   *pIter = taosHashIterate(pVgMap, NULL);
×
4277
  while (pIter) {
×
4278
    const SVDropTbVgReqs *pVgReqs = pIter;
×
4279
    int32_t               len = 0;
×
4280
    for (int32_t i = 0; i < taosArrayGetSize(pVgReqs->pBatchReqs) && code == TSDB_CODE_SUCCESS; ++i) {
×
4281
      SVDropTbBatchReq *pBatchReq = taosArrayGet(pVgReqs->pBatchReqs, i);
×
4282
      void             *p = mndBuildVDropTbsReq(pMnode, &pVgReqs->info, pBatchReq, &len);
×
4283
      if (!p) {
×
4284
        code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
4285
        if (terrno != 0) code = terrno;
×
4286
        break;
×
4287
      }
4288
      if ((code = mndSetDropTbsRedoActions(pMnode, pTrans, pVgReqs, p, len, msgType)) != 0) {
×
4289
        break;
×
4290
      }
4291
    }
4292
    if (TSDB_CODE_SUCCESS != code) {
×
4293
      taosHashCancelIterate(pVgMap, pIter);
×
4294
      break;
×
4295
    }
4296
    pIter = taosHashIterate(pVgMap, pIter);
×
4297
  }
4298
  return code;
×
4299
}
4300

4301
static int32_t mndCreateDropTbsTxnPrepare(SRpcMsg *pRsp, SMndDropTbsWithTsmaCtx *pCtx) {
×
4302
  int32_t code = 0;
×
4303
  SMnode *pMnode = pRsp->info.node;
×
4304
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, pRsp, "drop-tbs");
×
4305
  mndTransSetChangeless(pTrans);
×
4306
  mndTransSetSerial(pTrans);
×
4307
  if (pTrans == NULL) {
×
4308
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
4309
    if (terrno != 0) code = terrno;
×
4310
    goto _OVER;
×
4311
  }
4312

4313
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
×
4314

4315
  if ((code = mndBuildDropTbRedoActions(pMnode, pTrans, pCtx->pVgMap, TDMT_VND_DROP_TABLE)) != 0) goto _OVER;
×
4316
  if ((code = mndTransPrepare(pMnode, pTrans)) != 0) goto _OVER;
×
4317

4318
_OVER:
×
4319
  mndTransDrop(pTrans);
×
4320
  TAOS_RETURN(code);
×
4321
}
4322

4323
static int32_t mndProcessDropTbWithTsma(SRpcMsg *pReq) {
×
4324
  int32_t      code = -1;
×
4325
  SMnode      *pMnode = pReq->info.node;
×
4326
  SDbObj      *pDb = NULL;
×
4327
  SStbObj     *pStb = NULL;
×
4328
  SMDropTbsReq dropReq = {0};
×
4329
  bool         locked = false;
×
4330
  if (tDeserializeSMDropTbsReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
×
4331
    code = TSDB_CODE_INVALID_MSG;
×
4332
    goto _OVER;
×
4333
  }
4334

4335
  SMndDropTbsWithTsmaCtx *pCtx = NULL;
×
4336
  code = mndInitDropTbsWithTsmaCtx(&pCtx);
×
4337
  if (code) goto _OVER;
×
4338
  for (int32_t i = 0; i < dropReq.pVgReqs->size; ++i) {
×
4339
    SMDropTbReqsOnSingleVg *pReq = taosArrayGet(dropReq.pVgReqs, i);
×
4340
    code = mndDropTbForSingleVg(pMnode, pCtx, pReq->pTbs, pReq->vgInfo.vgId);
×
4341
    if (code) goto _OVER;
×
4342
  }
4343
  code = mndCreateDropTbsTxnPrepare(pReq, pCtx);
×
4344
  if (code == 0) {
×
4345
    code = TSDB_CODE_ACTION_IN_PROGRESS;
×
4346
  }
4347
_OVER:
×
4348
  tFreeSMDropTbsReq(&dropReq);
×
4349
  if (pCtx) mndDestroyDropTbsWithTsmaCtx(pCtx);
×
4350
  TAOS_RETURN(code);
×
4351
}
4352

4353
static int32_t createDropTbBatchReq(const SVDropTbReq *pReq, SVDropTbBatchReq *pBatchReq) {
×
4354
  pBatchReq->nReqs = 1;
×
4355
  pBatchReq->pArray = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SVDropTbReq));
×
4356
  if (!pBatchReq->pArray) return terrno;
×
4357
  if (taosArrayPush(pBatchReq->pArray, pReq) == NULL) {
×
4358
    taosArrayDestroy(pBatchReq->pArray);
×
4359
    pBatchReq->pArray = NULL;
×
4360
    return terrno;
×
4361
  }
4362
  return TSDB_CODE_SUCCESS;
×
4363
}
4364

4365
static void destroySVDropTbBatchReqs(void *p) {
×
4366
  SVDropTbBatchReq *pReq = p;
×
4367
  taosArrayDestroy(pReq->pArray);
×
4368
  pReq->pArray = NULL;
×
4369
}
×
4370

4371
static int32_t mndDropTbAdd(SMnode *pMnode, SHashObj *pVgHashMap, const SVgroupInfo *pVgInfo, char *name, tb_uid_t suid,
×
4372
                            bool ignoreNotExists) {
4373
  SVDropTbReq req = {.name = name, .suid = suid, .igNotExists = ignoreNotExists, .uid = 0};
×
4374

4375
  SVDropTbVgReqs *pVgReqs = taosHashGet(pVgHashMap, &pVgInfo->vgId, sizeof(pVgInfo->vgId));
×
4376
  SVDropTbVgReqs  vgReqs = {0};
×
4377
  if (pVgReqs == NULL) {
×
4378
    vgReqs.info = *pVgInfo;
×
4379
    vgReqs.pBatchReqs = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SVDropTbBatchReq));
×
4380
    if (!vgReqs.pBatchReqs) return terrno;
×
4381
    SVDropTbBatchReq batchReq = {0};
×
4382
    int32_t          code = createDropTbBatchReq(&req, &batchReq);
×
4383
    if (TSDB_CODE_SUCCESS != code) return code;
×
4384
    if (taosArrayPush(vgReqs.pBatchReqs, &batchReq) == NULL) {
×
4385
      taosArrayDestroy(batchReq.pArray);
×
4386
      return terrno;
×
4387
    }
4388
    if (taosHashPut(pVgHashMap, &pVgInfo->vgId, sizeof(pVgInfo->vgId), &vgReqs, sizeof(vgReqs)) != 0) {
×
4389
      taosArrayDestroyEx(vgReqs.pBatchReqs, destroySVDropTbBatchReqs);
×
4390
      return terrno;
×
4391
    }
4392
  } else {
4393
    SVDropTbBatchReq batchReq = {0};
×
4394
    int32_t          code = createDropTbBatchReq(&req, &batchReq);
×
4395
    if (TSDB_CODE_SUCCESS != code) return code;
×
4396
    if (taosArrayPush(pVgReqs->pBatchReqs, &batchReq) == NULL) {
×
4397
      taosArrayDestroy(batchReq.pArray);
×
4398
      return terrno;
×
4399
    }
4400
  }
4401
  return 0;
×
4402
}
4403

4404
static int32_t mndDropTbForSingleVg(SMnode *pMnode, SMndDropTbsWithTsmaCtx *pCtx, SArray *pTbs, int32_t vgId) {
×
4405
  int32_t code = 0;
×
4406

4407
  SVgObj *pVgObj = mndAcquireVgroup(pMnode, vgId);
×
4408
  if (!pVgObj) {
×
4409
    code = 0;
×
4410
    goto _end;
×
4411
  }
4412
  SVgroupInfo vgInfo = {.hashBegin = pVgObj->hashBegin,
×
4413
                        .hashEnd = pVgObj->hashEnd,
×
4414
                        .numOfTable = pVgObj->numOfTables,
×
4415
                        .vgId = pVgObj->vgId};
×
4416
  vgInfo.epSet = mndGetVgroupEpset(pMnode, pVgObj);
×
4417
  mndReleaseVgroup(pMnode, pVgObj);
×
4418

4419
  for (int32_t i = 0; i < pTbs->size; ++i) {
×
4420
    SVDropTbReq *pTb = taosArrayGet(pTbs, i);
×
4421
    TAOS_CHECK_GOTO(mndDropTbAdd(pMnode, pCtx->pVgMap, &vgInfo, pTb->name, pTb->suid, pTb->igNotExists), NULL, _end);
×
4422
  }
4423
_end:
×
4424
  return code;
×
4425
}
4426

4427
static int32_t mndProcessFetchTtlExpiredTbs(SRpcMsg *pRsp) {
2✔
4428
  int32_t                 code = -1;
2✔
4429
  SDecoder                decoder = {0};
2✔
4430
  SMnode                 *pMnode = pRsp->info.node;
2✔
4431
  SVFetchTtlExpiredTbsRsp rsp = {0};
2✔
4432
  SMndDropTbsWithTsmaCtx *pCtx = NULL;
2✔
4433
  if (pRsp->code != TSDB_CODE_SUCCESS) {
2✔
4434
    code = pRsp->code;
1✔
4435
    goto _end;
1✔
4436
  }
4437
  if (pRsp->contLen == 0) {
1✔
4438
    code = 0;
1✔
4439
    goto _end;
1✔
4440
  }
4441

4442
  tDecoderInit(&decoder, pRsp->pCont, pRsp->contLen);
×
4443
  code = tDecodeVFetchTtlExpiredTbsRsp(&decoder, &rsp);
×
4444
  if (code) goto _end;
×
4445

4446
  code = mndInitDropTbsWithTsmaCtx(&pCtx);
×
4447
  if (code) goto _end;
×
4448

4449
  code = mndDropTbForSingleVg(pMnode, pCtx, rsp.pExpiredTbs, rsp.vgId);
×
4450
  if (code) goto _end;
×
4451
  code = mndCreateDropTbsTxnPrepare(pRsp, pCtx);
×
4452
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
×
4453
_end:
×
4454
  if (pCtx) mndDestroyDropTbsWithTsmaCtx(pCtx);
2✔
4455
  tDecoderClear(&decoder);
2✔
4456
  tFreeFetchTtlExpiredTbsRsp(&rsp);
2✔
4457
  TAOS_RETURN(code);
2✔
4458
}
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