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

H
Hongze Cheng 已提交
16
#include "vnd.h"
H
Hongze Cheng 已提交
17

H
Hongze Cheng 已提交
18
static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp);
H
Hongze Cheng 已提交
19
static int vnodeProcessAlterStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
H
Hongze Cheng 已提交
20
static int vnodeProcessDropStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
H
Hongze Cheng 已提交
21
static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp);
H
Hongze Cheng 已提交
22
static int vnodeProcessAlterTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
H
Hongze Cheng 已提交
23
static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
H
Hongze Cheng 已提交
24
static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
25
static int vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
H
Hongze Cheng 已提交
26

H
Hongze Cheng 已提交
27
int32_t vnodePreprocessReq(SVnode *pVnode, SRpcMsg *pMsg) {
S
Shengliang Guan 已提交
28
  int32_t  code = 0;
H
Hongze Cheng 已提交
29
  SDecoder dc = {0};
H
Hongze Cheng 已提交
30

H
Hongze Cheng 已提交
31 32 33 34 35 36 37 38 39 40 41
  switch (pMsg->msgType) {
    case TDMT_VND_CREATE_TABLE: {
      int64_t ctime = taosGetTimestampMs();
      int32_t nReqs;

      tDecoderInit(&dc, (uint8_t *)pMsg->pCont + sizeof(SMsgHead), pMsg->contLen - sizeof(SMsgHead));
      tStartDecode(&dc);

      tDecodeI32v(&dc, &nReqs);
      for (int32_t iReq = 0; iReq < nReqs; iReq++) {
        tb_uid_t uid = tGenIdPI64();
H
Hongze Cheng 已提交
42
        char    *name = NULL;
H
Hongze Cheng 已提交
43 44 45
        tStartDecode(&dc);

        tDecodeI32v(&dc, NULL);
H
Hongze Cheng 已提交
46
        tDecodeCStr(&dc, &name);
H
Hongze Cheng 已提交
47 48 49 50 51 52 53 54 55 56 57 58 59 60
        *(int64_t *)(dc.data + dc.pos) = uid;
        *(int64_t *)(dc.data + dc.pos + 8) = ctime;

        tEndDecode(&dc);
      }

      tEndDecode(&dc);
      tDecoderClear(&dc);
    } break;
    case TDMT_VND_SUBMIT: {
      SSubmitMsgIter msgIter = {0};
      SSubmitReq    *pSubmitReq = (SSubmitReq *)pMsg->pCont;
      SSubmitBlk    *pBlock = NULL;
      int64_t        ctime = taosGetTimestampMs();
H
Hongze Cheng 已提交
61
      tb_uid_t       uid;
H
Hongze Cheng 已提交
62 63 64 65 66 67 68 69

      tInitSubmitMsgIter(pSubmitReq, &msgIter);

      for (;;) {
        tGetSubmitMsgNext(&msgIter, &pBlock);
        if (pBlock == NULL) break;

        if (msgIter.schemaLen > 0) {
H
Hongze Cheng 已提交
70
          char *name = NULL;
H
Hongze Cheng 已提交
71

H
Hongze Cheng 已提交
72 73 74 75
          tDecoderInit(&dc, pBlock->data, msgIter.schemaLen);
          tStartDecode(&dc);

          tDecodeI32v(&dc, NULL);
H
Hongze Cheng 已提交
76 77 78 79 80 81
          tDecodeCStr(&dc, &name);

          uid = metaGetTableEntryUidByName(pVnode->pMeta, name);
          if (uid == 0) {
            uid = tGenIdPI64();
          }
H
Hongze Cheng 已提交
82
          *(int64_t *)(dc.data + dc.pos) = uid;
H
Hongze Cheng 已提交
83
          *(int64_t *)(dc.data + dc.pos + 8) = ctime;
H
Hongze Cheng 已提交
84
          pBlock->uid = htobe64(uid);
H
Hongze Cheng 已提交
85 86 87 88 89 90 91

          tEndDecode(&dc);
          tDecoderClear(&dc);
        }
      }

    } break;
92
    case TDMT_VND_ALTER_REPLICA: {
S
Shengliang Guan 已提交
93
      code = vnodeSyncAlter(pVnode, pMsg);
94
    } break;
H
Hongze Cheng 已提交
95 96 97
    default:
      break;
  }
H
Hongze Cheng 已提交
98

S
Shengliang Guan 已提交
99
  return code;
H
Hongze Cheng 已提交
100 101
}

H
Hongze Cheng 已提交
102
int vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
103
  void *ptr = NULL;
H
Hongze Cheng 已提交
104 105
  void *pReq;
  int   len;
H
Hongze Cheng 已提交
106
  int   ret;
H
Hongze Cheng 已提交
107

S
Shengliang Guan 已提交
108
  vTrace("vgId:%d, start to process write request %s, version %" PRId64, TD_VID(pVnode), TMSG_INFO(pMsg->msgType),
H
Hongze Cheng 已提交
109
         version);
H
Hongze Cheng 已提交
110

H
Hongze Cheng 已提交
111 112
  pVnode->state.applied = version;

H
Hongze Cheng 已提交
113 114 115
  // skip header
  pReq = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
  len = pMsg->contLen - sizeof(SMsgHead);
H
Hongze Cheng 已提交
116 117

  switch (pMsg->msgType) {
H
Hongze Cheng 已提交
118
    /* META */
H
Hongze Cheng 已提交
119
    case TDMT_VND_CREATE_STB:
H
Hongze Cheng 已提交
120
      if (vnodeProcessCreateStbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
121
      break;
H
Hongze Cheng 已提交
122
    case TDMT_VND_ALTER_STB:
H
Hongze Cheng 已提交
123
      if (vnodeProcessAlterStbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
124
      break;
H
Hongze Cheng 已提交
125
    case TDMT_VND_DROP_STB:
H
Hongze Cheng 已提交
126
      if (vnodeProcessDropStbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
127
      break;
H
Hongze Cheng 已提交
128
    case TDMT_VND_CREATE_TABLE:
H
Hongze Cheng 已提交
129
      if (vnodeProcessCreateTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
130 131
      break;
    case TDMT_VND_ALTER_TABLE:
H
Hongze Cheng 已提交
132
      if (vnodeProcessAlterTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
133
      break;
H
Hongze Cheng 已提交
134
    case TDMT_VND_DROP_TABLE:
H
Hongze Cheng 已提交
135
      if (vnodeProcessDropTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
136
      break;
137 138
    case TDMT_VND_CREATE_SMA: {
      if (vnodeProcessCreateTSmaReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
139 140
    } break;
    /* TSDB */
H
Hongze Cheng 已提交
141
    case TDMT_VND_SUBMIT:
H
Hongze Cheng 已提交
142
      if (vnodeProcessSubmitReq(pVnode, version, pMsg->pCont, pMsg->contLen, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
143
      break;
H
Hongze Cheng 已提交
144
    /* TQ */
L
Liu Jicong 已提交
145 146 147 148 149 150
    case TDMT_VND_MQ_VG_CHANGE:
      if (tqProcessVgChangeReq(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
                               pMsg->contLen - sizeof(SMsgHead)) < 0) {
        // TODO: handle error
      }
      break;
L
Liu Jicong 已提交
151 152 153 154 155
    case TDMT_VND_MQ_VG_DELETE:
      if (tqProcessVgDeleteReq(pVnode->pTq, pMsg->pCont, pMsg->contLen) < 0) {
        // TODO: handle error
      }
      break;
H
Hongze Cheng 已提交
156 157 158 159 160
    case TDMT_VND_TASK_DEPLOY: {
      if (tqProcessTaskDeploy(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
                              pMsg->contLen - sizeof(SMsgHead)) < 0) {
      }
    } break;
161
    case TDMT_VND_ALTER_CONFIG:
S
Shengliang Guan 已提交
162
      break;
H
Hongze Cheng 已提交
163 164 165 166 167
    default:
      ASSERT(0);
      break;
  }

S
Shengliang Guan 已提交
168
  vTrace("vgId:%d, process %s request success, version: %" PRId64, TD_VID(pVnode), TMSG_INFO(pMsg->msgType), version);
H
Hongze Cheng 已提交
169

170
  if (tqPushMsg(pVnode->pTq, pMsg->pCont, pMsg->contLen, pMsg->msgType, version) < 0) {
S
Shengliang Guan 已提交
171
    vError("vgId:%d, failed to push msg to TQ since %s", TD_VID(pVnode), tstrerror(terrno));
172 173 174
    return -1;
  }

H
Hongze Cheng 已提交
175
  // commit if need
H
Hongze Cheng 已提交
176
  if (vnodeShouldCommit(pVnode)) {
S
Shengliang Guan 已提交
177
    vInfo("vgId:%d, commit at version %" PRId64, TD_VID(pVnode), version);
H
Hongze Cheng 已提交
178 179 180 181 182
    // commit current change
    vnodeCommit(pVnode);

    // start a new one
    vnodeBegin(pVnode);
H
Hongze Cheng 已提交
183 184 185
  }

  return 0;
H
Hongze Cheng 已提交
186 187

_err:
S
Shengliang Guan 已提交
188
  vError("vgId:%d, process %s request failed since %s, version: %" PRId64, TD_VID(pVnode), TMSG_INFO(pMsg->msgType),
H
Hongze Cheng 已提交
189 190
         tstrerror(terrno), version);
  return -1;
H
Hongze Cheng 已提交
191 192 193
}

int vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
D
dapan 已提交
194
  vTrace("message in vnode query queue is processing");
195
  SReadHandle handle = {.meta = pVnode->pMeta, .config = &pVnode->config, .vnode = pVnode, .pMsgCb = &pVnode->msgCb};
H
Hongze Cheng 已提交
196 197
  switch (pMsg->msgType) {
    case TDMT_VND_QUERY:
D
dapan1121 已提交
198
      return qWorkerProcessQueryMsg(&handle, pVnode->pQuery, pMsg, 0);
H
Hongze Cheng 已提交
199
    case TDMT_VND_QUERY_CONTINUE:
D
dapan1121 已提交
200
      return qWorkerProcessCQueryMsg(&handle, pVnode->pQuery, pMsg, 0);
H
Hongze Cheng 已提交
201 202 203 204 205 206 207 208 209 210 211 212
    default:
      vError("unknown msg type:%d in query queue", pMsg->msgType);
      return TSDB_CODE_VND_APP_ERROR;
  }
}

int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
  vTrace("message in fetch queue is processing");
  char   *msgstr = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
  int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);
  switch (pMsg->msgType) {
    case TDMT_VND_FETCH:
D
dapan1121 已提交
213
      return qWorkerProcessFetchMsg(pVnode, pVnode->pQuery, pMsg, 0);
H
Hongze Cheng 已提交
214
    case TDMT_VND_FETCH_RSP:
D
dapan1121 已提交
215
      return qWorkerProcessFetchRsp(pVnode, pVnode->pQuery, pMsg, 0);
H
Hongze Cheng 已提交
216
    case TDMT_VND_CANCEL_TASK:
D
dapan1121 已提交
217
      return qWorkerProcessCancelMsg(pVnode, pVnode->pQuery, pMsg, 0);
H
Hongze Cheng 已提交
218
    case TDMT_VND_DROP_TASK:
D
dapan1121 已提交
219 220 221 222
      return qWorkerProcessDropMsg(pVnode, pVnode->pQuery, pMsg, 0);
    case TDMT_VND_QUERY_HEARTBEAT:
      return qWorkerProcessHbMsg(pVnode, pVnode->pQuery, pMsg, 0);

H
Hongze Cheng 已提交
223 224 225 226
    case TDMT_VND_TABLE_META:
      return vnodeGetTableMeta(pVnode, pMsg);
    case TDMT_VND_CONSUME:
      return tqProcessPollReq(pVnode->pTq, pMsg, pInfo->workerId);
L
Liu Jicong 已提交
227

L
Liu Jicong 已提交
228 229 230 231 232
    case TDMT_VND_TASK_RUN: {
      int32_t code = tqProcessTaskRunReq(pVnode->pTq, pMsg);
      pMsg->pCont = NULL;
      return code;
    }
L
Liu Jicong 已提交
233 234 235 236 237 238 239 240
    case TDMT_VND_TASK_DISPATCH:
      return tqProcessTaskDispatchReq(pVnode->pTq, pMsg);
    case TDMT_VND_TASK_RECOVER:
      return tqProcessTaskRecoverReq(pVnode->pTq, pMsg);
    case TDMT_VND_TASK_DISPATCH_RSP:
      return tqProcessTaskDispatchRsp(pVnode->pTq, pMsg);
    case TDMT_VND_TASK_RECOVER_RSP:
      return tqProcessTaskRecoverRsp(pVnode->pTq, pMsg);
H
Hongze Cheng 已提交
241 242 243 244 245 246 247 248 249 250 251
    default:
      vError("unknown msg type:%d in fetch queue", pMsg->msgType);
      return TSDB_CODE_VND_APP_ERROR;
  }
}

// TODO: remove the function
void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data) {
  // TODO

  // blockDebugShowData(data);
252
  tdProcessTSmaInsert(((SVnode *)pVnode)->pSma, smaId, (const char *)data);
H
Hongze Cheng 已提交
253 254
}

D
dapan1121 已提交
255 256 257 258 259 260 261
void vnodeUpdateMetaRsp(SVnode *pVnode, STableMetaRsp *pMetaRsp) {
  strcpy(pMetaRsp->dbFName, pVnode->config.dbname);
  pMetaRsp->dbId = pVnode->config.dbId;
  pMetaRsp->vgId = TD_VID(pVnode);
  pMetaRsp->precision = pVnode->config.tsdbCfg.precision;
}

H
Hongze Cheng 已提交
262
int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
263
  int32_t ret = TAOS_SYNC_PROPOSE_OTHER_ERROR;
H
Hongze Cheng 已提交
264

265 266 267
  if (syncEnvIsStart()) {
    SSyncNode *pSyncNode = syncNodeAcquire(pVnode->sync);
    assert(pSyncNode != NULL);
M
Minghao Li 已提交
268

269 270
    ESyncState state = syncGetMyRole(pVnode->sync);
    SyncTerm   currentTerm = syncGetMyTerm(pVnode->sync);
M
Minghao Li 已提交
271

272
    SMsgHead *pHead = pMsg->pCont;
M
Minghao Li 已提交
273

274
    char  logBuf[512] = {0};
275 276 277 278
    char *syncNodeStr = sync2SimpleStr(pVnode->sync);
    snprintf(logBuf, sizeof(logBuf), "==vnodeProcessSyncReq== msgType:%d, syncNode: %s", pMsg->msgType, syncNodeStr);
    syncRpcMsgLog2(logBuf, pMsg);
    taosMemoryFree(syncNodeStr);
M
Minghao Li 已提交
279

280
    SRpcMsg *pRpcMsg = pMsg;
M
Minghao Li 已提交
281

282 283 284
    if (pRpcMsg->msgType == TDMT_VND_SYNC_TIMEOUT) {
      SyncTimeout *pSyncMsg = syncTimeoutFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);
M
Minghao Li 已提交
285

286
      ret = syncNodeOnTimeoutCb(pSyncNode, pSyncMsg);
287
      syncTimeoutDestroy(pSyncMsg);
M
Minghao Li 已提交
288

289 290 291
    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING) {
      SyncPing *pSyncMsg = syncPingFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);
M
Minghao Li 已提交
292

293
      ret = syncNodeOnPingCb(pSyncNode, pSyncMsg);
294
      syncPingDestroy(pSyncMsg);
M
Minghao Li 已提交
295

296 297 298
    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING_REPLY) {
      SyncPingReply *pSyncMsg = syncPingReplyFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);
M
Minghao Li 已提交
299

300
      ret = syncNodeOnPingReplyCb(pSyncNode, pSyncMsg);
301
      syncPingReplyDestroy(pSyncMsg);
M
Minghao Li 已提交
302

303 304 305
    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_CLIENT_REQUEST) {
      SyncClientRequest *pSyncMsg = syncClientRequestFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);
M
Minghao Li 已提交
306

307
      ret = syncNodeOnClientRequestCb(pSyncNode, pSyncMsg);
308
      syncClientRequestDestroy(pSyncMsg);
M
Minghao Li 已提交
309

310 311 312
    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE) {
      SyncRequestVote *pSyncMsg = syncRequestVoteFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);
M
Minghao Li 已提交
313

314
      ret = syncNodeOnRequestVoteCb(pSyncNode, pSyncMsg);
315
      syncRequestVoteDestroy(pSyncMsg);
M
Minghao Li 已提交
316

317 318 319
    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE_REPLY) {
      SyncRequestVoteReply *pSyncMsg = syncRequestVoteReplyFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);
M
Minghao Li 已提交
320

321
      ret = syncNodeOnRequestVoteReplyCb(pSyncNode, pSyncMsg);
322
      syncRequestVoteReplyDestroy(pSyncMsg);
M
Minghao Li 已提交
323

324 325 326
    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES) {
      SyncAppendEntries *pSyncMsg = syncAppendEntriesFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);
M
Minghao Li 已提交
327

328
      ret = syncNodeOnAppendEntriesCb(pSyncNode, pSyncMsg);
329
      syncAppendEntriesDestroy(pSyncMsg);
M
Minghao Li 已提交
330

331 332 333 334
    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES_REPLY) {
      SyncAppendEntriesReply *pSyncMsg = syncAppendEntriesReplyFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);

335
      ret = syncNodeOnAppendEntriesReplyCb(pSyncNode, pSyncMsg);
336
      syncAppendEntriesReplyDestroy(pSyncMsg);
M
Minghao Li 已提交
337

338 339
    } else {
      vError("==vnodeProcessSyncReq== error msg type:%d", pRpcMsg->msgType);
340
      ret = TAOS_SYNC_PROPOSE_OTHER_ERROR;
341
    }
M
Minghao Li 已提交
342

343 344 345
    syncNodeRelease(pSyncNode);
  } else {
    vError("==vnodeProcessSyncReq== error syncEnv stop");
346
    ret = TAOS_SYNC_PROPOSE_OTHER_ERROR;
347
  }
348 349

  return ret;
H
Hongze Cheng 已提交
350 351
}

H
Hongze Cheng 已提交
352
static int vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
353
  SVCreateStbReq req = {0};
H
Hongze Cheng 已提交
354
  SDecoder       coder;
H
Hongze Cheng 已提交
355

H
Hongze Cheng 已提交
356 357 358 359 360 361
  pRsp->msgType = TDMT_VND_CREATE_STB_RSP;
  pRsp->code = TSDB_CODE_SUCCESS;
  pRsp->pCont = NULL;
  pRsp->contLen = 0;

  // decode and process req
H
Hongze Cheng 已提交
362
  tDecoderInit(&coder, pReq, len);
H
Hongze Cheng 已提交
363 364

  if (tDecodeSVCreateStbReq(&coder, &req) < 0) {
H
Hongze Cheng 已提交
365 366
    pRsp->code = terrno;
    goto _err;
H
Hongze Cheng 已提交
367 368
  }

H
Hongze Cheng 已提交
369
  if (metaCreateSTable(pVnode->pMeta, version, &req) < 0) {
H
Hongze Cheng 已提交
370 371
    pRsp->code = terrno;
    goto _err;
H
Hongze Cheng 已提交
372 373
  }

5
54liuyao 已提交
374
  tdProcessRSmaCreate(pVnode, &req);
C
Cary Xu 已提交
375

H
Hongze Cheng 已提交
376
  tDecoderClear(&coder);
H
Hongze Cheng 已提交
377
  return 0;
H
Hongze Cheng 已提交
378 379

_err:
H
Hongze Cheng 已提交
380
  tDecoderClear(&coder);
H
Hongze Cheng 已提交
381
  return -1;
H
Hongze Cheng 已提交
382 383
}

H
Hongze Cheng 已提交
384
static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
385
  SDecoder           decoder = {0};
H
Hongze Cheng 已提交
386 387 388
  int                rcode = 0;
  SVCreateTbBatchReq req = {0};
  SVCreateTbReq     *pCreateReq;
H
Hongze Cheng 已提交
389 390 391
  SVCreateTbBatchRsp rsp = {0};
  SVCreateTbRsp      cRsp = {0};
  char               tbName[TSDB_TABLE_FNAME_LEN];
C
Cary Xu 已提交
392
  STbUidStore       *pStore = NULL;
393
  SArray            *tbUids = NULL;
H
Hongze Cheng 已提交
394 395

  pRsp->msgType = TDMT_VND_CREATE_TABLE_RSP;
H
Hongze Cheng 已提交
396 397 398
  pRsp->code = TSDB_CODE_SUCCESS;
  pRsp->pCont = NULL;
  pRsp->contLen = 0;
H
Hongze Cheng 已提交
399

H
Hongze Cheng 已提交
400
  // decode
H
Hongze Cheng 已提交
401 402
  tDecoderInit(&decoder, pReq, len);
  if (tDecodeSVCreateTbBatchReq(&decoder, &req) < 0) {
H
Hongze Cheng 已提交
403 404 405 406
    rcode = -1;
    terrno = TSDB_CODE_INVALID_MSG;
    goto _exit;
  }
H
Hongze Cheng 已提交
407

H
Hongze Cheng 已提交
408
  rsp.pArray = taosArrayInit(req.nReqs, sizeof(cRsp));
409 410
  tbUids = taosArrayInit(req.nReqs, sizeof(int64_t));
  if (rsp.pArray == NULL || tbUids == NULL) {
H
Hongze Cheng 已提交
411 412 413 414 415
    rcode = -1;
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

H
Hongze Cheng 已提交
416 417 418
  // loop to create table
  for (int iReq = 0; iReq < req.nReqs; iReq++) {
    pCreateReq = req.pReqs + iReq;
H
Hongze Cheng 已提交
419 420 421 422 423 424 425 426 427 428

    // validate hash
    sprintf(tbName, "%s.%s", pVnode->config.dbname, pCreateReq->name);
    if (vnodeValidateTableHash(pVnode, tbName) < 0) {
      cRsp.code = TSDB_CODE_VND_HASH_MISMATCH;
      taosArrayPush(rsp.pArray, &cRsp);
      continue;
    }

    // do create table
H
Hongze Cheng 已提交
429
    if (metaCreateTable(pVnode->pMeta, version, pCreateReq) < 0) {
H
Hongze Cheng 已提交
430 431 432 433 434
      if (pCreateReq->flags & TD_CREATE_IF_NOT_EXISTS && terrno == TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
        cRsp.code = TSDB_CODE_SUCCESS;
      } else {
        cRsp.code = terrno;
      }
H
Hongze Cheng 已提交
435
    } else {
H
Hongze Cheng 已提交
436
      cRsp.code = TSDB_CODE_SUCCESS;
437
      tdFetchTbUidList(pVnode->pSma, &pStore, pCreateReq->ctb.suid, pCreateReq->uid);
438
      taosArrayPush(tbUids, &pCreateReq->uid);
H
Hongze Cheng 已提交
439
    }
H
Hongze Cheng 已提交
440 441

    taosArrayPush(rsp.pArray, &cRsp);
H
Hongze Cheng 已提交
442 443
  }

H
Hongze Cheng 已提交
444
  tDecoderClear(&decoder);
H
Hongze Cheng 已提交
445

446
  tqUpdateTbUidList(pVnode->pTq, tbUids, true);
447 448
  tdUpdateTbUidList(pVnode->pSma, pStore);
  tdUidStoreFree(pStore);
C
Cary Xu 已提交
449

H
Hongze Cheng 已提交
450
  // prepare rsp
H
Hongze Cheng 已提交
451 452
  SEncoder encoder = {0};
  int32_t  ret = 0;
wafwerar's avatar
wafwerar 已提交
453
  tEncodeSize(tEncodeSVCreateTbBatchRsp, &rsp, pRsp->contLen, ret);
H
Hongze Cheng 已提交
454 455 456 457 458 459
  pRsp->pCont = rpcMallocCont(pRsp->contLen);
  if (pRsp->pCont == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    rcode = -1;
    goto _exit;
  }
H
Hongze Cheng 已提交
460 461 462
  tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
  tEncodeSVCreateTbBatchRsp(&encoder, &rsp);
  tEncoderClear(&encoder);
H
Hongze Cheng 已提交
463

H
Hongze Cheng 已提交
464
_exit:
H
Hongze Cheng 已提交
465
  taosArrayDestroy(rsp.pArray);
466
  taosArrayDestroy(tbUids);
H
Hongze Cheng 已提交
467 468
  tDecoderClear(&decoder);
  tEncoderClear(&encoder);
H
Hongze Cheng 已提交
469
  return rcode;
H
Hongze Cheng 已提交
470 471
}

H
Hongze Cheng 已提交
472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487
static int vnodeProcessAlterStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
  SVCreateStbReq req = {0};
  SDecoder       dc = {0};

  pRsp->msgType = TDMT_VND_ALTER_STB_RSP;
  pRsp->code = TSDB_CODE_SUCCESS;
  pRsp->pCont = NULL;
  pRsp->contLen = 0;

  tDecoderInit(&dc, pReq, len);

  // decode req
  if (tDecodeSVCreateStbReq(&dc, &req) < 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    tDecoderClear(&dc);
    return -1;
H
Hongze Cheng 已提交
488
  }
H
Hongze Cheng 已提交
489 490 491 492 493 494 495 496 497

  if (metaAlterSTable(pVnode->pMeta, version, &req) < 0) {
    pRsp->code = terrno;
    tDecoderClear(&dc);
    return -1;
  }

  tDecoderClear(&dc);

H
Hongze Cheng 已提交
498 499 500
  return 0;
}

H
Hongze Cheng 已提交
501
static int vnodeProcessDropStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
502 503
  SVDropStbReq req = {0};
  int          rcode = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
504
  SDecoder     decoder = {0};
H
Hongze Cheng 已提交
505 506 507 508 509 510

  pRsp->msgType = TDMT_VND_CREATE_STB_RSP;
  pRsp->pCont = NULL;
  pRsp->contLen = 0;

  // decode request
H
Hongze Cheng 已提交
511 512
  tDecoderInit(&decoder, pReq, len);
  if (tDecodeSVDropStbReq(&decoder, &req) < 0) {
H
Hongze Cheng 已提交
513 514 515 516 517
    rcode = TSDB_CODE_INVALID_MSG;
    goto _exit;
  }

  // process request
H
Hongze Cheng 已提交
518 519 520 521
  // if (metaDropSTable(pVnode->pMeta, version, &req) < 0) {
  //   rcode = terrno;
  //   goto _exit;
  // }
H
Hongze Cheng 已提交
522 523 524 525

  // return rsp
_exit:
  pRsp->code = rcode;
H
Hongze Cheng 已提交
526
  tDecoderClear(&decoder);
H
Hongze Cheng 已提交
527 528 529
  return 0;
}

H
Hongze Cheng 已提交
530
static int vnodeProcessAlterTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
D
dapan1121 已提交
531 532 533 534 535 536 537
  SVAlterTbReq  vAlterTbReq = {0};
  SVAlterTbRsp  vAlterTbRsp = {0};
  SDecoder      dc = {0};
  int           rcode = 0;
  int           ret;
  SEncoder      ec = {0};
  STableMetaRsp vMetaRsp = {0};
H
Hongze Cheng 已提交
538 539 540 541 542 543 544 545 546 547

  pRsp->msgType = TDMT_VND_ALTER_TABLE_RSP;
  pRsp->pCont = NULL;
  pRsp->contLen = 0;
  pRsp->code = TSDB_CODE_SUCCESS;

  tDecoderInit(&dc, pReq, len);

  // decode
  if (tDecodeSVAlterTbReq(&dc, &vAlterTbReq) < 0) {
H
Hongze Cheng 已提交
548
    vAlterTbRsp.code = TSDB_CODE_INVALID_MSG;
H
Hongze Cheng 已提交
549
    tDecoderClear(&dc);
H
Hongze Cheng 已提交
550 551
    rcode = -1;
    goto _exit;
H
Hongze Cheng 已提交
552 553 554
  }

  // process
D
dapan1121 已提交
555
  if (metaAlterTable(pVnode->pMeta, version, &vAlterTbReq, &vMetaRsp) < 0) {
H
Hongze Cheng 已提交
556
    vAlterTbRsp.code = TSDB_CODE_INVALID_MSG;
H
Hongze Cheng 已提交
557
    tDecoderClear(&dc);
H
Hongze Cheng 已提交
558 559
    rcode = -1;
    goto _exit;
H
Hongze Cheng 已提交
560 561
  }
  tDecoderClear(&dc);
H
Hongze Cheng 已提交
562

D
dapan1121 已提交
563 564 565 566 567
  if (NULL != vMetaRsp.pSchemas) {
    vnodeUpdateMetaRsp(pVnode, &vMetaRsp);
    vAlterTbRsp.pMeta = &vMetaRsp;
  }

H
Hongze Cheng 已提交
568 569 570 571 572 573
_exit:
  tEncodeSize(tEncodeSVAlterTbRsp, &vAlterTbRsp, pRsp->contLen, ret);
  pRsp->pCont = rpcMallocCont(pRsp->contLen);
  tEncoderInit(&ec, pRsp->pCont, pRsp->contLen);
  tEncodeSVAlterTbRsp(&ec, &vAlterTbRsp);
  tEncoderClear(&ec);
H
Hongze Cheng 已提交
574 575 576
  return 0;
}

H
Hongze Cheng 已提交
577
static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
578 579
  SVDropTbBatchReq req = {0};
  SVDropTbBatchRsp rsp = {0};
H
Hongze Cheng 已提交
580
  SDecoder         decoder = {0};
H
Hongze Cheng 已提交
581
  SEncoder         encoder = {0};
H
Hongze Cheng 已提交
582
  int              ret;
583
  SArray          *tbUids = NULL;
H
Hongze Cheng 已提交
584

H
Hongze Cheng 已提交
585
  pRsp->msgType = TDMT_VND_DROP_TABLE_RSP;
H
Hongze Cheng 已提交
586 587 588
  pRsp->pCont = NULL;
  pRsp->contLen = 0;
  pRsp->code = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
589 590

  // decode req
H
Hongze Cheng 已提交
591 592
  tDecoderInit(&decoder, pReq, len);
  ret = tDecodeSVDropTbBatchReq(&decoder, &req);
H
Hongze Cheng 已提交
593 594 595 596 597
  if (ret < 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    pRsp->code = terrno;
    goto _exit;
  }
H
Hongze Cheng 已提交
598 599

  // process req
600
  tbUids = taosArrayInit(req.nReqs, sizeof(int64_t));
H
Hongze Cheng 已提交
601
  rsp.pArray = taosArrayInit(req.nReqs, sizeof(SVDropTbRsp));
602 603
  if (tbUids == NULL || rsp.pArray == NULL) goto _exit;

H
Hongze Cheng 已提交
604 605 606
  for (int iReq = 0; iReq < req.nReqs; iReq++) {
    SVDropTbReq *pDropTbReq = req.pReqs + iReq;
    SVDropTbRsp  dropTbRsp = {0};
H
Hongze Cheng 已提交
607

H
Hongze Cheng 已提交
608
    /* code */
609
    ret = metaDropTable(pVnode->pMeta, version, pDropTbReq, tbUids);
H
Hongze Cheng 已提交
610
    if (ret < 0) {
H
Hongze Cheng 已提交
611 612 613 614 615
      if (pDropTbReq->igNotExists && terrno == TSDB_CODE_VND_TABLE_NOT_EXIST) {
        dropTbRsp.code = TSDB_CODE_SUCCESS;
      } else {
        dropTbRsp.code = terrno;
      }
H
Hongze Cheng 已提交
616
    } else {
H
Hongze Cheng 已提交
617
      dropTbRsp.code = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
618 619 620 621 622
    }

    taosArrayPush(rsp.pArray, &dropTbRsp);
  }

623 624
  tqUpdateTbUidList(pVnode->pTq, tbUids, false);

H
Hongze Cheng 已提交
625
_exit:
626
  taosArrayDestroy(tbUids);
H
Hongze Cheng 已提交
627
  tDecoderClear(&decoder);
H
Hongze Cheng 已提交
628 629 630 631 632
  tEncodeSize(tEncodeSVDropTbBatchRsp, &rsp, pRsp->contLen, ret);
  pRsp->pCont = rpcMallocCont(pRsp->contLen);
  tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
  tEncodeSVDropTbBatchRsp(&encoder, &rsp);
  tEncoderClear(&encoder);
H
Hongze Cheng 已提交
633 634 635
  return 0;
}

H
Hongze Cheng 已提交
636
static int vnodeDebugPrintSingleSubmitMsg(SMeta *pMeta, SSubmitBlk *pBlock, SSubmitMsgIter *msgIter, const char *tags) {
D
dapan 已提交
637 638 639 640
  SSubmitBlkIter blkIter = {0};
  STSchema      *pSchema = NULL;
  tb_uid_t       suid = 0;
  STSRow        *row = NULL;
C
Cary Xu 已提交
641
  int32_t        rv = -1;
D
dapan 已提交
642 643 644

  tInitSubmitBlkIter(msgIter, pBlock, &blkIter);
  if (blkIter.row == NULL) return 0;
C
Cary Xu 已提交
645
  if (!pSchema || (suid != msgIter->suid) || rv != TD_ROW_SVER(blkIter.row)) {
D
dapan 已提交
646 647 648
    if (pSchema) {
      taosMemoryFreeClear(pSchema);
    }
C
Cary Xu 已提交
649
    pSchema = metaGetTbTSchema(pMeta, msgIter->suid, TD_ROW_SVER(blkIter.row));  // TODO: use the real schema
D
dapan 已提交
650 651
    if (pSchema) {
      suid = msgIter->suid;
C
Cary Xu 已提交
652
      rv = TD_ROW_SVER(blkIter.row);
D
dapan 已提交
653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669
    }
  }
  if (!pSchema) {
    printf("%s:%d no valid schema\n", tags, __LINE__);
    return -1;
  }
  char __tags[128] = {0};
  snprintf(__tags, 128, "%s: uid %" PRIi64 " ", tags, msgIter->uid);
  while ((row = tGetSubmitBlkNext(&blkIter))) {
    tdSRowPrint(row, pSchema, __tags);
  }

  taosMemoryFreeClear(pSchema);

  return TSDB_CODE_SUCCESS;
}

H
Hongze Cheng 已提交
670
static int vnodeDebugPrintSubmitMsg(SVnode *pVnode, SSubmitReq *pMsg, const char *tags) {
C
Cary Xu 已提交
671 672 673 674 675 676 677 678 679
  ASSERT(pMsg != NULL);
  SSubmitMsgIter msgIter = {0};
  SMeta         *pMeta = pVnode->pMeta;
  SSubmitBlk    *pBlock = NULL;

  if (tInitSubmitMsgIter(pMsg, &msgIter) < 0) return -1;
  while (true) {
    if (tGetSubmitMsgNext(&msgIter, &pBlock) < 0) return -1;
    if (pBlock == NULL) break;
H
Hongze Cheng 已提交
680

D
dapan 已提交
681 682
    vnodeDebugPrintSingleSubmitMsg(pMeta, pBlock, &msgIter, tags);
  }
C
Cary Xu 已提交
683 684 685 686

  return 0;
}

H
Hongze Cheng 已提交
687
static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
688
  SSubmitReq    *pSubmitReq = (SSubmitReq *)pReq;
H
Hongze Cheng 已提交
689
  SSubmitRsp     submitRsp = {0};
H
Hongze Cheng 已提交
690 691 692 693
  SSubmitMsgIter msgIter = {0};
  SSubmitBlk    *pBlock;
  SSubmitRsp     rsp = {0};
  SVCreateTbReq  createTbReq = {0};
H
Hongze Cheng 已提交
694
  SDecoder       decoder = {0};
H
Hongze Cheng 已提交
695
  int32_t        nRows;
H
Hongze Cheng 已提交
696 697
  int32_t        tsize, ret;
  SEncoder       encoder = {0};
698
  SArray        *newTbUids = NULL;
699
  terrno = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
700 701

  pRsp->code = 0;
C
Cary Xu 已提交
702

C
Cary Xu 已提交
703 704 705 706
#ifdef TD_DEBUG_PRINT_ROW
  vnodeDebugPrintSubmitMsg(pVnode, pReq, __func__);
#endif

C
Cary Xu 已提交
707 708 709 710 711
  if (tsdbScanAndConvertSubmitMsg(pVnode->pTsdb, pSubmitReq) < 0) {
    pRsp->code = terrno;
    goto _exit;
  }

H
Hongze Cheng 已提交
712
  // handle the request
H
Hongze Cheng 已提交
713 714 715
  if (tInitSubmitMsgIter(pSubmitReq, &msgIter) < 0) {
    pRsp->code = TSDB_CODE_INVALID_MSG;
    goto _exit;
H
Hongze Cheng 已提交
716 717
  }

C
Cary Xu 已提交
718 719
  submitRsp.pArray = taosArrayInit(msgIter.numOfBlocks, sizeof(SSubmitBlkRsp));
  newTbUids = taosArrayInit(msgIter.numOfBlocks, sizeof(int64_t));
720 721 722 723 724
  if (!submitRsp.pArray) {
    pRsp->code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

H
Hongze Cheng 已提交
725
  for (;;) {
H
Hongze Cheng 已提交
726 727 728
    tGetSubmitMsgNext(&msgIter, &pBlock);
    if (pBlock == NULL) break;

H
Hongze Cheng 已提交
729 730
    SSubmitBlkRsp submitBlkRsp = {0};

H
Hongze Cheng 已提交
731 732
    // create table for auto create table mode
    if (msgIter.schemaLen > 0) {
H
Hongze Cheng 已提交
733 734
      submitBlkRsp.hashMeta = 1;

H
Hongze Cheng 已提交
735 736
      tDecoderInit(&decoder, pBlock->data, msgIter.schemaLen);
      if (tDecodeSVCreateTbReq(&decoder, &createTbReq) < 0) {
H
Hongze Cheng 已提交
737
        pRsp->code = TSDB_CODE_INVALID_MSG;
H
Hongze Cheng 已提交
738
        tDecoderClear(&decoder);
H
Hongze Cheng 已提交
739 740 741 742 743
        goto _exit;
      }

      if (metaCreateTable(pVnode->pMeta, version, &createTbReq) < 0) {
        if (terrno != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
H
Hongze Cheng 已提交
744
          submitBlkRsp.code = terrno;
H
Hongze Cheng 已提交
745
          tDecoderClear(&decoder);
H
Hongze Cheng 已提交
746 747 748
          goto _exit;
        }
      }
749
      taosArrayPush(newTbUids, &createTbReq.uid);
H
Hongze Cheng 已提交
750

H
Hongze Cheng 已提交
751
      submitBlkRsp.uid = createTbReq.uid;
D
dapan 已提交
752
      submitBlkRsp.tblFName = taosMemoryMalloc(strlen(pVnode->config.dbname) + strlen(createTbReq.name) + 2);
D
dapan 已提交
753
      sprintf(submitBlkRsp.tblFName, "%s.", pVnode->config.dbname);
H
Hongze Cheng 已提交
754

H
Hongze Cheng 已提交
755 756 757 758 759 760 761
      msgIter.uid = createTbReq.uid;
      if (createTbReq.type == TSDB_CHILD_TABLE) {
        msgIter.suid = createTbReq.ctb.suid;
      } else {
        msgIter.suid = 0;
      }

D
dapan 已提交
762
      vnodeDebugPrintSingleSubmitMsg(pVnode->pMeta, pBlock, &msgIter, "real uid");
H
Hongze Cheng 已提交
763
      tDecoderClear(&decoder);
D
dapan1121 已提交
764 765 766
    } else {
      submitBlkRsp.tblFName = taosMemoryMalloc(TSDB_TABLE_FNAME_LEN);
      sprintf(submitBlkRsp.tblFName, "%s.", pVnode->config.dbname);
H
Hongze Cheng 已提交
767 768
    }

H
Hongze Cheng 已提交
769
    if (tsdbInsertTableData(pVnode->pTsdb, &msgIter, pBlock, &submitBlkRsp) < 0) {
H
Hongze Cheng 已提交
770
      submitBlkRsp.code = terrno;
H
Hongze Cheng 已提交
771 772
    }

H
Hongze Cheng 已提交
773 774 775
    submitRsp.numOfRows += submitBlkRsp.numOfRows;
    submitRsp.affectedRows += submitBlkRsp.affectedRows;
    taosArrayPush(submitRsp.pArray, &submitBlkRsp);
H
Hongze Cheng 已提交
776
  }
777
  tqUpdateTbUidList(pVnode->pTq, newTbUids, true);
778

H
Hongze Cheng 已提交
779
_exit:
780
  taosArrayDestroy(newTbUids);
H
Hongze Cheng 已提交
781 782 783 784 785 786 787 788
  tEncodeSize(tEncodeSSubmitRsp, &submitRsp, tsize, ret);
  pRsp->pCont = rpcMallocCont(tsize);
  pRsp->contLen = tsize;
  tEncoderInit(&encoder, pRsp->pCont, tsize);
  tEncodeSSubmitRsp(&encoder, &submitRsp);
  tEncoderClear(&encoder);

  for (int32_t i = 0; i < taosArrayGetSize(submitRsp.pArray); i++) {
D
dapan 已提交
789
    taosMemoryFree(((SSubmitBlkRsp *)taosArrayGet(submitRsp.pArray, i))[0].tblFName);
H
Hongze Cheng 已提交
790 791 792
  }

  taosArrayDestroy(submitRsp.pArray);
H
Hongze Cheng 已提交
793

794 795
  // TODO: the partial success scenario and the error case
  // TODO: refactor
C
Cary Xu 已提交
796
  if ((terrno == TSDB_CODE_SUCCESS) && (pRsp->code == TSDB_CODE_SUCCESS)) {
797
    tdProcessRSmaSubmit(pVnode->pSma, pReq, STREAM_DATA_TYPE_SUBMIT_BLOCK);
798
  }
C
Cary Xu 已提交
799

H
Hongze Cheng 已提交
800
  return 0;
L
Liu Jicong 已提交
801
}
802

803
static int vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
804
  SVCreateTSmaReq req = {0};
H
Hongze Cheng 已提交
805
  SDecoder        coder;
806

C
Cary Xu 已提交
807 808 809 810 811 812
  if (pRsp) {
    pRsp->msgType = TDMT_VND_CREATE_SMA_RSP;
    pRsp->code = TSDB_CODE_SUCCESS;
    pRsp->pCont = NULL;
    pRsp->contLen = 0;
  }
813 814 815 816 817

  // decode and process req
  tDecoderInit(&coder, pReq, len);

  if (tDecodeSVCreateTSmaReq(&coder, &req) < 0) {
C
Cary Xu 已提交
818 819
    terrno = TSDB_CODE_MSG_DECODE_ERROR;
    if (pRsp) pRsp->code = terrno;
820
    goto _err;
821
  }
C
Cary Xu 已提交
822

C
Cary Xu 已提交
823 824 825 826
  // record current timezone of server side
  req.timezoneInt = tsTimezone;

  if (tdProcessTSmaCreate(pVnode->pSma, version, (const char *)&req) < 0) {
C
Cary Xu 已提交
827
    if (pRsp) pRsp->code = terrno;
828
    goto _err;
829
  }
C
Cary Xu 已提交
830

831
  tDecoderClear(&coder);
S
Shengliang Guan 已提交
832
  vDebug("vgId:%d, success to create tsma %s:%" PRIi64 " version %" PRIi64 " for table %" PRIi64, TD_VID(pVnode),
C
Cary Xu 已提交
833
         req.indexName, req.indexUid, version, req.tableUid);
H
Hongze Cheng 已提交
834
  return 0;
835 836 837

_err:
  tDecoderClear(&coder);
S
Shengliang Guan 已提交
838
  vError("vgId:%d, failed to create tsma %s:%" PRIi64 " version %" PRIi64 "for table %" PRIi64 " since %s",
C
Cary Xu 已提交
839
         TD_VID(pVnode), req.indexName, req.indexUid, version, req.tableUid, terrstr(terrno));
840
  return -1;
L
Liu Jicong 已提交
841
}
C
Cary Xu 已提交
842 843 844 845 846 847 848 849 850 851 852 853

/**
 * @brief specific for smaDstVnode
 *
 * @param pVnode
 * @param pCont
 * @param contLen
 * @return int32_t
 */
int32_t vnodeProcessCreateTSma(SVnode *pVnode, void *pCont, uint32_t contLen) {
  return vnodeProcessCreateTSmaReq(pVnode, 1, pCont, contLen, NULL);
}