vnodeSvr.c 39.1 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

18 19 20 21 22 23 24 25 26
static int32_t vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessAlterStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessDropStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessAlterTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessAlterConfirmReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
27 28
static int32_t vnodeProcessAlterHashRangeReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
static int32_t vnodeProcessAlterConfigReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
29
static int32_t vnodeProcessDropTtlTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
S
Shengliang Guan 已提交
30
static int32_t vnodeProcessTrimReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
H
Hongze Cheng 已提交
31
static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
32
static int32_t vnodeProcessBatchDeleteReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
H
Hongze Cheng 已提交
33

34
int32_t vnodePreProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg) {
S
Shengliang Guan 已提交
35
  int32_t  code = 0;
H
Hongze Cheng 已提交
36
  SDecoder dc = {0};
H
Hongze Cheng 已提交
37

H
Hongze Cheng 已提交
38 39 40 41 42 43
  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));
H
Hongze Cheng 已提交
44 45 46 47
      if (tStartDecode(&dc) < 0) {
        code = TSDB_CODE_INVALID_MSG;
        return code;
      }
H
Hongze Cheng 已提交
48

H
Hongze Cheng 已提交
49 50 51 52
      if (tDecodeI32v(&dc, &nReqs) < 0) {
        code = TSDB_CODE_INVALID_MSG;
        goto _err;
      }
H
Hongze Cheng 已提交
53 54
      for (int32_t iReq = 0; iReq < nReqs; iReq++) {
        tb_uid_t uid = tGenIdPI64();
H
Hongze Cheng 已提交
55
        char    *name = NULL;
H
Hongze Cheng 已提交
56 57 58 59
        if (tStartDecode(&dc) < 0) {
          code = TSDB_CODE_INVALID_MSG;
          goto _err;
        }
H
Hongze Cheng 已提交
60

H
Hongze Cheng 已提交
61 62 63 64 65 66 67 68
        if (tDecodeI32v(&dc, NULL) < 0) {
          code = TSDB_CODE_INVALID_MSG;
          return code;
        }
        if (tDecodeCStr(&dc, &name) < 0) {
          code = TSDB_CODE_INVALID_MSG;
          return code;
        }
H
Hongze Cheng 已提交
69 70 71
        *(int64_t *)(dc.data + dc.pos) = uid;
        *(int64_t *)(dc.data + dc.pos + 8) = ctime;

72
        vTrace("vgId:%d, table:%s uid:%" PRId64 " is generated", pVnode->config.vgId, name, uid);
H
Hongze Cheng 已提交
73 74 75 76 77 78 79 80 81 82 83
        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 已提交
84
      tb_uid_t       uid;
H
Hongze Cheng 已提交
85

H
Hongze Cheng 已提交
86 87 88 89
      if (tInitSubmitMsgIter(pSubmitReq, &msgIter) < 0) {
        code = terrno;
        goto _err;
      }
H
Hongze Cheng 已提交
90 91 92 93 94 95

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

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

H
Hongze Cheng 已提交
98
          tDecoderInit(&dc, pBlock->data, msgIter.schemaLen);
H
Hongze Cheng 已提交
99 100 101 102
          if (tStartDecode(&dc) < 0) {
            code = TSDB_CODE_INVALID_MSG;
            return code;
          }
H
Hongze Cheng 已提交
103

H
Hongze Cheng 已提交
104 105 106 107 108 109 110 111
          if (tDecodeI32v(&dc, NULL) < 0) {
            code = TSDB_CODE_INVALID_MSG;
            return code;
          }
          if (tDecodeCStr(&dc, &name) < 0) {
            code = TSDB_CODE_INVALID_MSG;
            return code;
          }
H
Hongze Cheng 已提交
112 113 114 115 116

          uid = metaGetTableEntryUidByName(pVnode->pMeta, name);
          if (uid == 0) {
            uid = tGenIdPI64();
          }
H
Hongze Cheng 已提交
117
          *(int64_t *)(dc.data + dc.pos) = uid;
H
Hongze Cheng 已提交
118
          *(int64_t *)(dc.data + dc.pos + 8) = ctime;
H
Hongze Cheng 已提交
119
          pBlock->uid = htobe64(uid);
H
Hongze Cheng 已提交
120 121 122 123 124 125 126

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

    } break;
H
Hongze Cheng 已提交
127 128 129 130 131 132 133 134 135 136
    case TDMT_VND_DELETE: {
      int32_t     size;
      int32_t     ret;
      uint8_t    *pCont;
      SEncoder   *pCoder = &(SEncoder){0};
      SDeleteRes  res = {0};
      SReadHandle handle = {
          .meta = pVnode->pMeta, .config = &pVnode->config, .vnode = pVnode, .pMsgCb = &pVnode->msgCb};

      code = qWorkerProcessDeleteMsg(&handle, pVnode->pQuery, pMsg, &res);
wmmhello's avatar
wmmhello 已提交
137 138 139
      if (code) {
        goto _err;
      }
H
Hongze Cheng 已提交
140 141 142 143 144

      // malloc and encode
      tEncodeSize(tEncodeDeleteRes, &res, size, ret);
      pCont = rpcMallocCont(size + sizeof(SMsgHead));

145 146
      ((SMsgHead *)pCont)->contLen = size + sizeof(SMsgHead);
      ((SMsgHead *)pCont)->vgId = TD_VID(pVnode);
H
Hongze Cheng 已提交
147 148 149 150 151 152 153 154 155 156 157

      tEncoderInit(pCoder, pCont + sizeof(SMsgHead), size);
      tEncodeDeleteRes(pCoder, &res);
      tEncoderClear(pCoder);

      rpcFreeCont(pMsg->pCont);
      pMsg->pCont = pCont;
      pMsg->contLen = size + sizeof(SMsgHead);

      taosArrayDestroy(res.uidList);
    } break;
H
Hongze Cheng 已提交
158 159 160
    default:
      break;
  }
H
Hongze Cheng 已提交
161

S
Shengliang Guan 已提交
162
  return code;
H
Hongze Cheng 已提交
163 164 165 166

_err:
  vError("vgId%d, preprocess request failed since %s", TD_VID(pVnode), tstrerror(code));
  return code;
H
Hongze Cheng 已提交
167 168
}

169
int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg *pRsp) {
170 171 172 173
  void   *ptr = NULL;
  void   *pReq;
  int32_t len;
  int32_t ret;
H
Hongze Cheng 已提交
174

175
  if (!pVnode->inUse) {
176
    terrno = TSDB_CODE_VND_NO_AVAIL_BUFPOOL;
H
Hongze Cheng 已提交
177 178
    vError("vgId:%d, not ready to write since %s", TD_VID(pVnode), terrstr());
    return -1;
179 180
  }

181 182 183
  if (version <= pVnode->state.applied) {
    vError("vgId:%d, duplicate write request. version: %" PRId64 ", applied: %" PRId64 "", TD_VID(pVnode), version,
           pVnode->state.applied);
184
    terrno = TSDB_CODE_VND_DUP_REQUEST;
185 186 187 188
    pRsp->info.handle = NULL;
    return -1;
  }

189
  vDebug("vgId:%d, start to process write request %s, index:%" PRId64, TD_VID(pVnode), TMSG_INFO(pMsg->msgType),
H
Hongze Cheng 已提交
190
         version);
H
Hongze Cheng 已提交
191

192
  ASSERT(pVnode->state.applyTerm <= pMsg->info.conn.applyTerm);
193 194
  ASSERT(pVnode->state.applied + 1 == version);

H
Hongze Cheng 已提交
195
  pVnode->state.applied = version;
H
Hongze Cheng 已提交
196
  pVnode->state.applyTerm = pMsg->info.conn.applyTerm;
H
Hongze Cheng 已提交
197

198 199
  if (!syncUtilUserCommit(pMsg->msgType)) goto _exit;

H
Hongze Cheng 已提交
200 201 202
  // skip header
  pReq = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
  len = pMsg->contLen - sizeof(SMsgHead);
H
Hongze Cheng 已提交
203 204

  switch (pMsg->msgType) {
H
Hongze Cheng 已提交
205
    /* META */
H
Hongze Cheng 已提交
206
    case TDMT_VND_CREATE_STB:
H
Hongze Cheng 已提交
207
      if (vnodeProcessCreateStbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
208
      break;
H
Hongze Cheng 已提交
209
    case TDMT_VND_ALTER_STB:
H
Hongze Cheng 已提交
210
      if (vnodeProcessAlterStbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
211
      break;
H
Hongze Cheng 已提交
212
    case TDMT_VND_DROP_STB:
H
Hongze Cheng 已提交
213
      if (vnodeProcessDropStbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
214
      break;
H
Hongze Cheng 已提交
215
    case TDMT_VND_CREATE_TABLE:
H
Hongze Cheng 已提交
216
      if (vnodeProcessCreateTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
217 218
      break;
    case TDMT_VND_ALTER_TABLE:
H
Hongze Cheng 已提交
219
      if (vnodeProcessAlterTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
220
      break;
H
Hongze Cheng 已提交
221
    case TDMT_VND_DROP_TABLE:
H
Hongze Cheng 已提交
222
      if (vnodeProcessDropTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
223
      break;
224
    case TDMT_VND_DROP_TTL_TABLE:
225
      if (vnodeProcessDropTtlTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
226
      break;
S
Shengliang Guan 已提交
227 228 229 230
    case TDMT_VND_TRIM:
      if (vnodeProcessTrimReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
      break;
    case TDMT_VND_CREATE_SMA:
231
      if (vnodeProcessCreateTSmaReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
S
Shengliang Guan 已提交
232
      break;
H
Hongze Cheng 已提交
233
    /* TSDB */
H
Hongze Cheng 已提交
234
    case TDMT_VND_SUBMIT:
H
Hongze Cheng 已提交
235
      if (vnodeProcessSubmitReq(pVnode, version, pMsg->pCont, pMsg->contLen, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
236
      break;
D
dapan1121 已提交
237
    case TDMT_VND_DELETE:
H
Hongze Cheng 已提交
238
      if (vnodeProcessDeleteReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
D
dapan1121 已提交
239
      break;
240 241 242
    case TDMT_VND_BATCH_DEL:
      if (vnodeProcessBatchDeleteReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
      break;
H
Hongze Cheng 已提交
243
    /* TQ */
L
Liu Jicong 已提交
244
    case TDMT_VND_TMQ_SUBSCRIBE:
245
      if (tqProcessSubscribeReq(pVnode->pTq, version, pReq, len) < 0) {
246
        goto _err;
L
Liu Jicong 已提交
247 248
      }
      break;
L
Liu Jicong 已提交
249 250
    case TDMT_VND_TMQ_DELETE_SUB:
      if (tqProcessDeleteSubReq(pVnode->pTq, version, pMsg->pCont, pMsg->contLen) < 0) {
251 252 253
        goto _err;
      }
      break;
L
Liu Jicong 已提交
254
    case TDMT_VND_TMQ_COMMIT_OFFSET:
255
      if (tqProcessOffsetCommitReq(pVnode->pTq, version, pReq, pMsg->contLen - sizeof(SMsgHead)) < 0) {
256
        goto _err;
L
Liu Jicong 已提交
257 258
      }
      break;
L
Liu Jicong 已提交
259
    case TDMT_VND_TMQ_ADD_CHECKINFO:
260
      if (tqProcessAddCheckInfoReq(pVnode->pTq, version, pReq, len) < 0) {
261 262 263
        goto _err;
      }
      break;
L
Liu Jicong 已提交
264
    case TDMT_VND_TMQ_DEL_CHECKINFO:
265
      if (tqProcessDelCheckInfoReq(pVnode->pTq, version, pReq, len) < 0) {
L
Liu Jicong 已提交
266 267 268
        goto _err;
      }
      break;
269
    case TDMT_STREAM_TASK_DEPLOY: {
270
      if (tqProcessTaskDeployReq(pVnode->pTq, version, pReq, len) < 0) {
271
        goto _err;
H
Hongze Cheng 已提交
272 273
      }
    } break;
L
Liu Jicong 已提交
274
    case TDMT_STREAM_TASK_DROP: {
275
      if (tqProcessTaskDropReq(pVnode->pTq, version, pMsg->pCont, pMsg->contLen) < 0) {
L
Liu Jicong 已提交
276 277 278
        goto _err;
      }
    } break;
L
Liu Jicong 已提交
279
    case TDMT_VND_STREAM_RECOVER_BLOCKING_STAGE: {
L
Liu Jicong 已提交
280 281 282 283
      if (tqProcessTaskRecover2Req(pVnode->pTq, version, pMsg->pCont, pMsg->contLen) < 0) {
        goto _err;
      }
    } break;
284 285 286 287 288
    case TDMT_STREAM_TASK_CHECK_RSP: {
      if (tqProcessStreamTaskCheckRsp(pVnode->pTq, version, pReq, len) < 0) {
        goto _err;
      }
    } break;
289 290 291
    case TDMT_VND_ALTER_CONFIRM:
      vnodeProcessAlterConfirmReq(pVnode, version, pReq, len, pRsp);
      break;
S
Shengliang Guan 已提交
292
    case TDMT_VND_ALTER_HASHRANGE:
293
      vnodeProcessAlterHashRangeReq(pVnode, version, pReq, len, pRsp);
S
Shengliang Guan 已提交
294
      break;
295
    case TDMT_VND_ALTER_CONFIG:
296
      vnodeProcessAlterConfigReq(pVnode, version, pReq, len, pRsp);
S
Shengliang Guan 已提交
297
      break;
H
Hongze Cheng 已提交
298
    case TDMT_VND_COMMIT:
H
Hongze Cheng 已提交
299 300 301
      vnodeSyncCommit(pVnode);
      vnodeBegin(pVnode);
      goto _exit;
H
Hongze Cheng 已提交
302
    default:
L
Liu Jicong 已提交
303 304
      vError("vgId:%d, unprocessed msg, %d", TD_VID(pVnode), pMsg->msgType);
      return -1;
H
Hongze Cheng 已提交
305 306
  }

S
Shengliang Guan 已提交
307 308
  vTrace("vgId:%d, process %s request, code:0x%x index:%" PRId64, TD_VID(pVnode), TMSG_INFO(pMsg->msgType), pRsp->code,
         version);
H
Hongze Cheng 已提交
309

310 311
  walApplyVer(pVnode->pWal, version);

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

H
Hongze Cheng 已提交
317
  // commit if need
H
Hongze Cheng 已提交
318
  if (vnodeShouldCommit(pVnode)) {
S
Shengliang Guan 已提交
319
    vInfo("vgId:%d, commit at version %" PRId64, TD_VID(pVnode), version);
H
Hongze Cheng 已提交
320
#if 0
H
Hongze Cheng 已提交
321 322
    vnodeSyncCommit(pVnode);
#else
H
Hongze Cheng 已提交
323
    vnodeAsyncCommit(pVnode);
H
Hongze Cheng 已提交
324
#endif
H
Hongze Cheng 已提交
325 326

    // start a new one
327
    if (vnodeBegin(pVnode) < 0) {
H
Hongze Cheng 已提交
328 329
      vError("vgId:%d, failed to begin vnode since %s.", TD_VID(pVnode), tstrerror(terrno));
      goto _err;
330
    }
H
Hongze Cheng 已提交
331 332
  }

H
Hongze Cheng 已提交
333
_exit:
H
Hongze Cheng 已提交
334
  return 0;
H
Hongze Cheng 已提交
335 336

_err:
S
Shengliang Guan 已提交
337
  vError("vgId:%d, process %s request failed since %s, version:%" PRId64, TD_VID(pVnode), TMSG_INFO(pMsg->msgType),
H
Hongze Cheng 已提交
338 339
         tstrerror(terrno), version);
  return -1;
H
Hongze Cheng 已提交
340 341
}

342
int32_t vnodePreprocessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
D
dapan1121 已提交
343
  if (TDMT_SCH_QUERY != pMsg->msgType && TDMT_SCH_MERGE_QUERY != pMsg->msgType) {
D
dapan1121 已提交
344 345 346
    return 0;
  }

347
  return qWorkerPreprocessQueryMsg(pVnode->pQuery, pMsg, TDMT_SCH_QUERY == pMsg->msgType);
D
dapan1121 已提交
348 349 350
}

int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
D
dapan 已提交
351
  vTrace("message in vnode query queue is processing");
352 353
  // if ((pMsg->msgType == TDMT_SCH_QUERY) && !vnodeIsLeader(pVnode)) {
  if ((pMsg->msgType == TDMT_SCH_QUERY) && !syncIsReadyForRead(pVnode->sync)) {
354
    vnodeRedirectRpcMsg(pVnode, pMsg, terrno);
355 356 357
    return 0;
  }

358
  SReadHandle handle = {.meta = pVnode->pMeta, .config = &pVnode->config, .vnode = pVnode, .pMsgCb = &pVnode->msgCb};
H
Hongze Cheng 已提交
359
  switch (pMsg->msgType) {
D
dapan1121 已提交
360
    case TDMT_SCH_QUERY:
D
dapan1121 已提交
361
    case TDMT_SCH_MERGE_QUERY:
D
dapan1121 已提交
362
      return qWorkerProcessQueryMsg(&handle, pVnode->pQuery, pMsg, 0);
D
dapan1121 已提交
363
    case TDMT_SCH_QUERY_CONTINUE:
D
dapan1121 已提交
364
      return qWorkerProcessCQueryMsg(&handle, pVnode->pQuery, pMsg, 0);
H
Hongze Cheng 已提交
365 366
    default:
      vError("unknown msg type:%d in query queue", pMsg->msgType);
367
      return TSDB_CODE_APP_ERROR;
H
Hongze Cheng 已提交
368 369 370
  }
}

371
int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
S
Shengliang Guan 已提交
372
  vTrace("vgId:%d, msg:%p in fetch queue is processing", pVnode->config.vgId, pMsg);
L
Liu Jicong 已提交
373 374
  if ((pMsg->msgType == TDMT_SCH_FETCH || pMsg->msgType == TDMT_VND_TABLE_META || pMsg->msgType == TDMT_VND_TABLE_CFG ||
       pMsg->msgType == TDMT_VND_BATCH_META) &&
375 376
      !syncIsReadyForRead(pVnode->sync)) {
    //      !vnodeIsLeader(pVnode)) {
377
    vnodeRedirectRpcMsg(pVnode, pMsg, terrno);
378 379 380
    return 0;
  }

L
Liu Jicong 已提交
381
  if (pMsg->msgType == TDMT_VND_TMQ_CONSUME && !pVnode->restored) {
382
    vnodeRedirectRpcMsg(pVnode, pMsg, TSDB_CODE_SYN_RESTORING);
L
Liu Jicong 已提交
383 384 385
    return 0;
  }

H
Hongze Cheng 已提交
386
  switch (pMsg->msgType) {
D
dapan1121 已提交
387
    case TDMT_SCH_FETCH:
D
dapan1121 已提交
388
    case TDMT_SCH_MERGE_FETCH:
D
dapan1121 已提交
389
      return qWorkerProcessFetchMsg(pVnode, pVnode->pQuery, pMsg, 0);
D
dapan1121 已提交
390
    case TDMT_SCH_FETCH_RSP:
D
dapan1121 已提交
391
      return qWorkerProcessRspMsg(pVnode, pVnode->pQuery, pMsg, 0);
L
Liu Jicong 已提交
392 393
    // case TDMT_SCH_CANCEL_TASK:
    //   return qWorkerProcessCancelMsg(pVnode, pVnode->pQuery, pMsg, 0);
D
dapan1121 已提交
394
    case TDMT_SCH_DROP_TASK:
D
dapan1121 已提交
395
      return qWorkerProcessDropMsg(pVnode, pVnode->pQuery, pMsg, 0);
D
dapan1121 已提交
396
    case TDMT_SCH_QUERY_HEARTBEAT:
D
dapan1121 已提交
397
      return qWorkerProcessHbMsg(pVnode, pVnode->pQuery, pMsg, 0);
H
Hongze Cheng 已提交
398
    case TDMT_VND_TABLE_META:
D
dapan1121 已提交
399
      return vnodeGetTableMeta(pVnode, pMsg, true);
D
dapan1121 已提交
400
    case TDMT_VND_TABLE_CFG:
D
dapan1121 已提交
401 402 403
      return vnodeGetTableCfg(pVnode, pMsg, true);
    case TDMT_VND_BATCH_META:
      return vnodeGetBatchMeta(pVnode, pMsg);
L
Liu Jicong 已提交
404
    case TDMT_VND_TMQ_CONSUME:
L
Liu Jicong 已提交
405
      return tqProcessPollReq(pVnode->pTq, pMsg);
406 407
    case TDMT_STREAM_TASK_RUN:
      return tqProcessTaskRunReq(pVnode->pTq, pMsg);
408
#if 1
409
    case TDMT_STREAM_TASK_DISPATCH:
L
Liu Jicong 已提交
410
      return tqProcessTaskDispatchReq(pVnode->pTq, pMsg, true);
L
Liu Jicong 已提交
411
#endif
412 413
    case TDMT_STREAM_TASK_CHECK:
      return tqProcessStreamTaskCheckReq(pVnode->pTq, pMsg);
414
    case TDMT_STREAM_TASK_DISPATCH_RSP:
L
Liu Jicong 已提交
415
      return tqProcessTaskDispatchRsp(pVnode->pTq, pMsg);
L
Liu Jicong 已提交
416 417
    case TDMT_STREAM_RETRIEVE:
      return tqProcessTaskRetrieveReq(pVnode->pTq, pMsg);
L
Liu Jicong 已提交
418 419
    case TDMT_STREAM_RETRIEVE_RSP:
      return tqProcessTaskRetrieveRsp(pVnode->pTq, pMsg);
L
Liu Jicong 已提交
420
    case TDMT_VND_STREAM_RECOVER_NONBLOCKING_STAGE:
L
Liu Jicong 已提交
421 422 423 424 425
      return tqProcessTaskRecover1Req(pVnode->pTq, pMsg);
    case TDMT_STREAM_RECOVER_FINISH:
      return tqProcessTaskRecoverFinishReq(pVnode->pTq, pMsg);
    case TDMT_STREAM_RECOVER_FINISH_RSP:
      return tqProcessTaskRecoverFinishRsp(pVnode->pTq, pMsg);
H
Hongze Cheng 已提交
426 427
    default:
      vError("unknown msg type:%d in fetch queue", pMsg->msgType);
428
      return TSDB_CODE_APP_ERROR;
H
Hongze Cheng 已提交
429 430 431 432 433 434
  }
}

// TODO: remove the function
void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data) {
  // TODO
C
Cary Xu 已提交
435
  // blockDebugShowDataBlocks(data, __func__);
436
  tdProcessTSmaInsert(((SVnode *)pVnode)->pSma, smaId, (const char *)data);
H
Hongze Cheng 已提交
437 438
}

D
dapan1121 已提交
439
void vnodeUpdateMetaRsp(SVnode *pVnode, STableMetaRsp *pMetaRsp) {
440 441 442
  if (NULL == pMetaRsp) {
    return;
  }
L
Liu Jicong 已提交
443

D
dapan1121 已提交
444 445 446 447 448 449
  strcpy(pMetaRsp->dbFName, pVnode->config.dbname);
  pMetaRsp->dbId = pVnode->config.dbId;
  pMetaRsp->vgId = TD_VID(pVnode);
  pMetaRsp->precision = pVnode->config.tsdbCfg.precision;
}

S
Shengliang Guan 已提交
450
static int32_t vnodeProcessTrimReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
451
  int32_t     code = 0;
S
Shengliang Guan 已提交
452 453
  SVTrimDbReq trimReq = {0};

H
Hongze Cheng 已提交
454 455 456 457
  // decode
  if (tDeserializeSVTrimDbReq(pReq, len, &trimReq) != 0) {
    code = TSDB_CODE_INVALID_MSG;
    goto _exit;
S
Shengliang Guan 已提交
458 459
  }

C
Cary Xu 已提交
460 461
  vInfo("vgId:%d, trim vnode request will be processed, time:%d", pVnode->config.vgId, trimReq.timestamp);

H
Hongze Cheng 已提交
462 463 464 465
  // process
  code = tsdbDoRetention(pVnode->pTsdb, trimReq.timestamp);
  if (code) goto _exit;

C
Cary Xu 已提交
466 467 468
  code = smaDoRetention(pVnode->pSma, trimReq.timestamp);
  if (code) goto _exit;

H
Hongze Cheng 已提交
469 470
_exit:
  return code;
S
Shengliang Guan 已提交
471 472
}

L
Liu Jicong 已提交
473
static int32_t vnodeProcessDropTtlTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
474 475 476
  SArray *tbUids = taosArrayInit(8, sizeof(int64_t));
  if (tbUids == NULL) return TSDB_CODE_OUT_OF_MEMORY;

S
Shengliang Guan 已提交
477 478 479 480 481 482
  SVDropTtlTableReq ttlReq = {0};
  if (tDeserializeSVDropTtlTableReq(pReq, len, &ttlReq) != 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    goto end;
  }

483
  vDebug("vgId:%d, drop ttl table req will be processed, time:%d", pVnode->config.vgId, ttlReq.timestamp);
S
Shengliang Guan 已提交
484
  int32_t ret = metaTtlDropTable(pVnode->pMeta, ttlReq.timestamp, tbUids);
L
Liu Jicong 已提交
485
  if (ret != 0) {
486 487
    goto end;
  }
L
Liu Jicong 已提交
488
  if (taosArrayGetSize(tbUids) > 0) {
wmmhello's avatar
wmmhello 已提交
489 490
    tqUpdateTbUidList(pVnode->pTq, tbUids, false);
  }
491 492 493 494 495 496

end:
  taosArrayDestroy(tbUids);
  return ret;
}

497
static int32_t vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
498
  SVCreateStbReq req = {0};
H
Hongze Cheng 已提交
499
  SDecoder       coder;
H
Hongze Cheng 已提交
500

H
Hongze Cheng 已提交
501 502 503 504 505 506
  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 已提交
507
  tDecoderInit(&coder, pReq, len);
H
Hongze Cheng 已提交
508 509

  if (tDecodeSVCreateStbReq(&coder, &req) < 0) {
H
Hongze Cheng 已提交
510 511
    pRsp->code = terrno;
    goto _err;
H
Hongze Cheng 已提交
512 513
  }

H
Hongze Cheng 已提交
514
  if (metaCreateSTable(pVnode->pMeta, version, &req) < 0) {
H
Hongze Cheng 已提交
515 516
    pRsp->code = terrno;
    goto _err;
H
Hongze Cheng 已提交
517 518
  }

519
  if (tdProcessRSmaCreate(pVnode->pSma, &req) < 0) {
520 521 522
    pRsp->code = terrno;
    goto _err;
  }
C
Cary Xu 已提交
523

H
Hongze Cheng 已提交
524
  tDecoderClear(&coder);
H
Hongze Cheng 已提交
525
  return 0;
H
Hongze Cheng 已提交
526 527

_err:
H
Hongze Cheng 已提交
528
  tDecoderClear(&coder);
H
Hongze Cheng 已提交
529
  return -1;
H
Hongze Cheng 已提交
530 531
}

532
static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
533
  SDecoder           decoder = {0};
534
  SEncoder           encoder = {0};
535
  int32_t            rcode = 0;
H
Hongze Cheng 已提交
536 537
  SVCreateTbBatchReq req = {0};
  SVCreateTbReq     *pCreateReq;
H
Hongze Cheng 已提交
538 539 540
  SVCreateTbBatchRsp rsp = {0};
  SVCreateTbRsp      cRsp = {0};
  char               tbName[TSDB_TABLE_FNAME_LEN];
C
Cary Xu 已提交
541
  STbUidStore       *pStore = NULL;
542
  SArray            *tbUids = NULL;
H
Hongze Cheng 已提交
543 544

  pRsp->msgType = TDMT_VND_CREATE_TABLE_RSP;
H
Hongze Cheng 已提交
545 546 547
  pRsp->code = TSDB_CODE_SUCCESS;
  pRsp->pCont = NULL;
  pRsp->contLen = 0;
H
Hongze Cheng 已提交
548

H
Hongze Cheng 已提交
549
  // decode
H
Hongze Cheng 已提交
550 551
  tDecoderInit(&decoder, pReq, len);
  if (tDecodeSVCreateTbBatchReq(&decoder, &req) < 0) {
H
Hongze Cheng 已提交
552 553 554 555
    rcode = -1;
    terrno = TSDB_CODE_INVALID_MSG;
    goto _exit;
  }
H
Hongze Cheng 已提交
556

H
Hongze Cheng 已提交
557
  rsp.pArray = taosArrayInit(req.nReqs, sizeof(cRsp));
558 559
  tbUids = taosArrayInit(req.nReqs, sizeof(int64_t));
  if (rsp.pArray == NULL || tbUids == NULL) {
H
Hongze Cheng 已提交
560 561 562 563 564
    rcode = -1;
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

H
Hongze Cheng 已提交
565
  // loop to create table
566
  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
H
Hongze Cheng 已提交
567
    pCreateReq = req.pReqs + iReq;
D
dapan1121 已提交
568
    memset(&cRsp, 0, sizeof(cRsp));
H
Hongze Cheng 已提交
569

C
Cary Xu 已提交
570 571 572 573
    if ((terrno = grantCheck(TSDB_GRANT_TIMESERIES)) < 0) {
      rcode = -1;
      goto _exit;
    }
L
Liu Jicong 已提交
574

wafwerar's avatar
wafwerar 已提交
575 576 577 578 579
    if ((terrno = grantCheck(TSDB_GRANT_TABLE)) < 0) {
      rcode = -1;
      goto _exit;
    }

H
Hongze Cheng 已提交
580 581 582 583 584 585 586 587 588
    // 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
589
    if (metaCreateTable(pVnode->pMeta, version, pCreateReq, &cRsp.pMeta) < 0) {
H
Hongze Cheng 已提交
590 591 592 593 594
      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 已提交
595
    } else {
H
Hongze Cheng 已提交
596
      cRsp.code = TSDB_CODE_SUCCESS;
597
      tdFetchTbUidList(pVnode->pSma, &pStore, pCreateReq->ctb.suid, pCreateReq->uid);
598
      taosArrayPush(tbUids, &pCreateReq->uid);
L
Liu Jicong 已提交
599
      vnodeUpdateMetaRsp(pVnode, cRsp.pMeta);
H
Hongze Cheng 已提交
600
    }
H
Hongze Cheng 已提交
601 602

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

H
Haojun Liao 已提交
605
  vDebug("vgId:%d, add %d new created tables into query table list", TD_VID(pVnode), (int32_t)taosArrayGetSize(tbUids));
606
  tqUpdateTbUidList(pVnode->pTq, tbUids, true);
607
  if (tdUpdateTbUidList(pVnode->pSma, pStore, true) < 0) {
C
Cary Xu 已提交
608 609
    goto _exit;
  }
610
  tdUidStoreFree(pStore);
C
Cary Xu 已提交
611

H
Hongze Cheng 已提交
612
  // prepare rsp
613
  int32_t ret = 0;
wafwerar's avatar
wafwerar 已提交
614
  tEncodeSize(tEncodeSVCreateTbBatchRsp, &rsp, pRsp->contLen, ret);
H
Hongze Cheng 已提交
615 616 617 618 619 620
  pRsp->pCont = rpcMallocCont(pRsp->contLen);
  if (pRsp->pCont == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    rcode = -1;
    goto _exit;
  }
H
Hongze Cheng 已提交
621 622
  tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
  tEncodeSVCreateTbBatchRsp(&encoder, &rsp);
H
Hongze Cheng 已提交
623

H
Hongze Cheng 已提交
624
_exit:
wmmhello's avatar
wmmhello 已提交
625 626
  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
    pCreateReq = req.pReqs + iReq;
627
    taosMemoryFree(pCreateReq->comment);
wmmhello's avatar
wmmhello 已提交
628 629
    taosArrayDestroy(pCreateReq->ctb.tagName);
  }
630
  taosArrayDestroyEx(rsp.pArray, tFreeSVCreateTbRsp);
631
  taosArrayDestroy(tbUids);
H
Hongze Cheng 已提交
632 633
  tDecoderClear(&decoder);
  tEncoderClear(&encoder);
H
Hongze Cheng 已提交
634
  return rcode;
H
Hongze Cheng 已提交
635 636
}

637
static int32_t vnodeProcessAlterStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
638 639 640 641 642 643 644 645 646 647 648 649 650 651 652
  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 已提交
653
  }
H
Hongze Cheng 已提交
654 655 656 657 658 659 660 661 662

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

  tDecoderClear(&dc);

H
Hongze Cheng 已提交
663 664 665
  return 0;
}

666
static int32_t vnodeProcessDropStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
667
  SVDropStbReq req = {0};
668
  int32_t      rcode = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
669
  SDecoder     decoder = {0};
670
  SArray      *tbUidList = NULL;
H
Hongze Cheng 已提交
671 672 673 674 675 676

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

  // decode request
H
Hongze Cheng 已提交
677 678
  tDecoderInit(&decoder, pReq, len);
  if (tDecodeSVDropStbReq(&decoder, &req) < 0) {
H
Hongze Cheng 已提交
679 680 681 682 683
    rcode = TSDB_CODE_INVALID_MSG;
    goto _exit;
  }

  // process request
684 685 686 687 688 689 690 691
  tbUidList = taosArrayInit(8, sizeof(int64_t));
  if (tbUidList == NULL) goto _exit;
  if (metaDropSTable(pVnode->pMeta, version, &req, tbUidList) < 0) {
    rcode = terrno;
    goto _exit;
  }

  if (tqUpdateTbUidList(pVnode->pTq, tbUidList, false) < 0) {
692 693 694
    rcode = terrno;
    goto _exit;
  }
H
Hongze Cheng 已提交
695

696 697 698 699 700
  if (tdProcessRSmaDrop(pVnode->pSma, &req) < 0) {
    rcode = terrno;
    goto _exit;
  }

H
Hongze Cheng 已提交
701 702
  // return rsp
_exit:
703
  if (tbUidList) taosArrayDestroy(tbUidList);
H
Hongze Cheng 已提交
704
  pRsp->code = rcode;
H
Hongze Cheng 已提交
705
  tDecoderClear(&decoder);
H
Hongze Cheng 已提交
706 707 708
  return 0;
}

709
static int32_t vnodeProcessAlterTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
D
dapan1121 已提交
710 711 712
  SVAlterTbReq  vAlterTbReq = {0};
  SVAlterTbRsp  vAlterTbRsp = {0};
  SDecoder      dc = {0};
713 714
  int32_t       rcode = 0;
  int32_t       ret;
D
dapan1121 已提交
715 716
  SEncoder      ec = {0};
  STableMetaRsp vMetaRsp = {0};
H
Hongze Cheng 已提交
717 718 719 720 721 722 723 724 725 726

  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 已提交
727
    vAlterTbRsp.code = TSDB_CODE_INVALID_MSG;
H
Hongze Cheng 已提交
728
    tDecoderClear(&dc);
H
Hongze Cheng 已提交
729 730
    rcode = -1;
    goto _exit;
H
Hongze Cheng 已提交
731 732 733
  }

  // process
D
dapan1121 已提交
734
  if (metaAlterTable(pVnode->pMeta, version, &vAlterTbReq, &vMetaRsp) < 0) {
wmmhello's avatar
wmmhello 已提交
735
    vAlterTbRsp.code = terrno;
H
Hongze Cheng 已提交
736
    tDecoderClear(&dc);
H
Hongze Cheng 已提交
737 738
    rcode = -1;
    goto _exit;
H
Hongze Cheng 已提交
739 740
  }
  tDecoderClear(&dc);
H
Hongze Cheng 已提交
741

D
dapan1121 已提交
742 743 744 745 746
  if (NULL != vMetaRsp.pSchemas) {
    vnodeUpdateMetaRsp(pVnode, &vMetaRsp);
    vAlterTbRsp.pMeta = &vMetaRsp;
  }

H
Hongze Cheng 已提交
747 748 749 750 751 752
_exit:
  tEncodeSize(tEncodeSVAlterTbRsp, &vAlterTbRsp, pRsp->contLen, ret);
  pRsp->pCont = rpcMallocCont(pRsp->contLen);
  tEncoderInit(&ec, pRsp->pCont, pRsp->contLen);
  tEncodeSVAlterTbRsp(&ec, &vAlterTbRsp);
  tEncoderClear(&ec);
M
Minglei Jin 已提交
753 754 755
  if (vMetaRsp.pSchemas) {
    taosMemoryFree(vMetaRsp.pSchemas);
  }
H
Hongze Cheng 已提交
756 757 758
  return 0;
}

759
static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
760 761
  SVDropTbBatchReq req = {0};
  SVDropTbBatchRsp rsp = {0};
H
Hongze Cheng 已提交
762
  SDecoder         decoder = {0};
H
Hongze Cheng 已提交
763
  SEncoder         encoder = {0};
764
  int32_t          ret;
765
  SArray          *tbUids = NULL;
766
  STbUidStore     *pStore = NULL;
H
Hongze Cheng 已提交
767

H
Hongze Cheng 已提交
768
  pRsp->msgType = TDMT_VND_DROP_TABLE_RSP;
H
Hongze Cheng 已提交
769 770 771
  pRsp->pCont = NULL;
  pRsp->contLen = 0;
  pRsp->code = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
772 773

  // decode req
H
Hongze Cheng 已提交
774 775
  tDecoderInit(&decoder, pReq, len);
  ret = tDecodeSVDropTbBatchReq(&decoder, &req);
H
Hongze Cheng 已提交
776 777 778 779 780
  if (ret < 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    pRsp->code = terrno;
    goto _exit;
  }
H
Hongze Cheng 已提交
781 782

  // process req
783
  tbUids = taosArrayInit(req.nReqs, sizeof(int64_t));
H
Hongze Cheng 已提交
784
  rsp.pArray = taosArrayInit(req.nReqs, sizeof(SVDropTbRsp));
785 786
  if (tbUids == NULL || rsp.pArray == NULL) goto _exit;

787
  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
H
Hongze Cheng 已提交
788 789
    SVDropTbReq *pDropTbReq = req.pReqs + iReq;
    SVDropTbRsp  dropTbRsp = {0};
790
    tb_uid_t     tbUid = 0;
H
Hongze Cheng 已提交
791

H
Hongze Cheng 已提交
792
    /* code */
793
    ret = metaDropTable(pVnode->pMeta, version, pDropTbReq, tbUids, &tbUid);
H
Hongze Cheng 已提交
794
    if (ret < 0) {
795
      if (pDropTbReq->igNotExists && terrno == TSDB_CODE_TDB_TABLE_NOT_EXIST) {
H
Hongze Cheng 已提交
796 797 798 799
        dropTbRsp.code = TSDB_CODE_SUCCESS;
      } else {
        dropTbRsp.code = terrno;
      }
H
Hongze Cheng 已提交
800
    } else {
H
Hongze Cheng 已提交
801
      dropTbRsp.code = TSDB_CODE_SUCCESS;
802
      if (tbUid > 0) tdFetchTbUidList(pVnode->pSma, &pStore, pDropTbReq->suid, tbUid);
H
Hongze Cheng 已提交
803 804 805 806 807
    }

    taosArrayPush(rsp.pArray, &dropTbRsp);
  }

808
  tqUpdateTbUidList(pVnode->pTq, tbUids, false);
809
  tdUpdateTbUidList(pVnode->pSma, pStore, false);
810

H
Hongze Cheng 已提交
811
_exit:
812
  taosArrayDestroy(tbUids);
813
  tdUidStoreFree(pStore);
H
Hongze Cheng 已提交
814
  tDecoderClear(&decoder);
H
Hongze Cheng 已提交
815 816 817 818 819
  tEncodeSize(tEncodeSVDropTbBatchRsp, &rsp, pRsp->contLen, ret);
  pRsp->pCont = rpcMallocCont(pRsp->contLen);
  tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
  tEncodeSVDropTbBatchRsp(&encoder, &rsp);
  tEncoderClear(&encoder);
820
  taosArrayDestroy(rsp.pArray);
H
Hongze Cheng 已提交
821 822 823
  return 0;
}

824 825
static int32_t vnodeDebugPrintSingleSubmitMsg(SMeta *pMeta, SSubmitBlk *pBlock, SSubmitMsgIter *msgIter,
                                              const char *tags) {
D
dapan 已提交
826 827 828 829
  SSubmitBlkIter blkIter = {0};
  STSchema      *pSchema = NULL;
  tb_uid_t       suid = 0;
  STSRow        *row = NULL;
C
Cary Xu 已提交
830
  int32_t        rv = -1;
D
dapan 已提交
831 832 833

  tInitSubmitBlkIter(msgIter, pBlock, &blkIter);
  if (blkIter.row == NULL) return 0;
834 835 836 837 838

  pSchema = metaGetTbTSchema(pMeta, msgIter->suid, TD_ROW_SVER(blkIter.row), 1);  // TODO: use the real schema
  if (pSchema) {
    suid = msgIter->suid;
    rv = TD_ROW_SVER(blkIter.row);
D
dapan 已提交
839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854
  }
  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;
}

855
static int32_t vnodeDebugPrintSubmitMsg(SVnode *pVnode, SSubmitReq *pMsg, const char *tags) {
C
Cary Xu 已提交
856 857 858 859 860 861 862 863 864
  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 已提交
865

D
dapan 已提交
866 867
    vnodeDebugPrintSingleSubmitMsg(pMeta, pBlock, &msgIter, tags);
  }
C
Cary Xu 已提交
868 869 870 871

  return 0;
}

872
static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
873
  SSubmitReq    *pSubmitReq = (SSubmitReq *)pReq;
H
Hongze Cheng 已提交
874
  SSubmitRsp     submitRsp = {0};
H
Hongze Cheng 已提交
875
  SSubmitMsgIter msgIter = {0};
C
Cary Xu 已提交
876
  SSubmitBlk    *pBlock = NULL;
H
Hongze Cheng 已提交
877
  SVCreateTbReq  createTbReq = {0};
H
Hongze Cheng 已提交
878
  SDecoder       decoder = {0};
C
Cary Xu 已提交
879
  int32_t        nRows = 0;
H
Hongze Cheng 已提交
880 881
  int32_t        tsize, ret;
  SEncoder       encoder = {0};
882
  SArray        *newTbUids = NULL;
883
  SVStatis       statis = {0};
D
dapan1121 已提交
884
  bool           tbCreated = false;
885
  terrno = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
886 887

  pRsp->code = 0;
C
Cary Xu 已提交
888
  pSubmitReq->version = version;
889
  statis.nBatchInsert = 1;
C
Cary Xu 已提交
890

C
Cary Xu 已提交
891 892 893 894
#ifdef TD_DEBUG_PRINT_ROW
  vnodeDebugPrintSubmitMsg(pVnode, pReq, __func__);
#endif

C
Cary Xu 已提交
895 896 897 898 899
  if (tsdbScanAndConvertSubmitMsg(pVnode->pTsdb, pSubmitReq) < 0) {
    pRsp->code = terrno;
    goto _exit;
  }

H
Hongze Cheng 已提交
900
  // handle the request
H
Hongze Cheng 已提交
901 902 903
  if (tInitSubmitMsgIter(pSubmitReq, &msgIter) < 0) {
    pRsp->code = TSDB_CODE_INVALID_MSG;
    goto _exit;
H
Hongze Cheng 已提交
904 905
  }

C
Cary Xu 已提交
906 907
  submitRsp.pArray = taosArrayInit(msgIter.numOfBlocks, sizeof(SSubmitBlkRsp));
  newTbUids = taosArrayInit(msgIter.numOfBlocks, sizeof(int64_t));
C
Cary Xu 已提交
908
  if (!submitRsp.pArray || !newTbUids) {
909 910 911 912
    pRsp->code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

H
Hongze Cheng 已提交
913
  for (;;) {
H
Hongze Cheng 已提交
914 915 916
    tGetSubmitMsgNext(&msgIter, &pBlock);
    if (pBlock == NULL) break;

H
Hongze Cheng 已提交
917
    SSubmitBlkRsp submitBlkRsp = {0};
D
dapan1121 已提交
918
    tbCreated = false;
H
Hongze Cheng 已提交
919

H
Hongze Cheng 已提交
920 921
    // create table for auto create table mode
    if (msgIter.schemaLen > 0) {
H
Hongze Cheng 已提交
922 923
      tDecoderInit(&decoder, pBlock->data, msgIter.schemaLen);
      if (tDecodeSVCreateTbReq(&decoder, &createTbReq) < 0) {
H
Hongze Cheng 已提交
924
        pRsp->code = TSDB_CODE_INVALID_MSG;
H
Hongze Cheng 已提交
925
        tDecoderClear(&decoder);
wmmhello's avatar
wmmhello 已提交
926
        taosArrayDestroy(createTbReq.ctb.tagName);
H
Hongze Cheng 已提交
927 928 929
        goto _exit;
      }

C
Cary Xu 已提交
930 931 932 933 934 935 936
      if ((terrno = grantCheck(TSDB_GRANT_TIMESERIES)) < 0) {
        pRsp->code = terrno;
        tDecoderClear(&decoder);
        taosArrayDestroy(createTbReq.ctb.tagName);
        goto _exit;
      }

wafwerar's avatar
wafwerar 已提交
937 938 939 940 941 942 943
      if ((terrno = grantCheck(TSDB_GRANT_TABLE)) < 0) {
        pRsp->code = terrno;
        tDecoderClear(&decoder);
        taosArrayDestroy(createTbReq.ctb.tagName);
        goto _exit;
      }

944
      if (metaCreateTable(pVnode->pMeta, version, &createTbReq, &submitBlkRsp.pMeta) < 0) {
H
Hongze Cheng 已提交
945
        if (terrno != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
H
Hongze Cheng 已提交
946
          submitBlkRsp.code = terrno;
C
Cary Xu 已提交
947
          pRsp->code = terrno;
H
Hongze Cheng 已提交
948
          tDecoderClear(&decoder);
wmmhello's avatar
wmmhello 已提交
949
          taosArrayDestroy(createTbReq.ctb.tagName);
H
Hongze Cheng 已提交
950 951
          goto _exit;
        }
952 953 954 955
      } else {
        if (NULL != submitBlkRsp.pMeta) {
          vnodeUpdateMetaRsp(pVnode, submitBlkRsp.pMeta);
        }
956 957

        taosArrayPush(newTbUids, &createTbReq.uid);
H
Hongze Cheng 已提交
958

D
dapan1121 已提交
959 960 961 962 963
        submitBlkRsp.uid = createTbReq.uid;
        submitBlkRsp.tblFName = taosMemoryMalloc(strlen(pVnode->config.dbname) + strlen(createTbReq.name) + 2);
        sprintf(submitBlkRsp.tblFName, "%s.%s", pVnode->config.dbname, createTbReq.name);
        tbCreated = true;
      }
964

H
Hongze Cheng 已提交
965 966 967 968 969 970 971
      msgIter.uid = createTbReq.uid;
      if (createTbReq.type == TSDB_CHILD_TABLE) {
        msgIter.suid = createTbReq.ctb.suid;
      } else {
        msgIter.suid = 0;
      }

wmmhello's avatar
wmmhello 已提交
972
#ifdef TD_DEBUG_PRINT_ROW
D
dapan 已提交
973
      vnodeDebugPrintSingleSubmitMsg(pVnode->pMeta, pBlock, &msgIter, "real uid");
wmmhello's avatar
wmmhello 已提交
974
#endif
H
Hongze Cheng 已提交
975
      tDecoderClear(&decoder);
wmmhello's avatar
wmmhello 已提交
976
      taosArrayDestroy(createTbReq.ctb.tagName);
977
    }
H
Hongze Cheng 已提交
978

H
Hongze Cheng 已提交
979
    if (tsdbInsertTableData(pVnode->pTsdb, version, &msgIter, pBlock, &submitBlkRsp) < 0) {
H
Hongze Cheng 已提交
980
      submitBlkRsp.code = terrno;
H
Hongze Cheng 已提交
981 982
    }

H
Hongze Cheng 已提交
983 984
    submitRsp.numOfRows += submitBlkRsp.numOfRows;
    submitRsp.affectedRows += submitBlkRsp.affectedRows;
D
dapan1121 已提交
985 986 987
    if (tbCreated || submitBlkRsp.code) {
      taosArrayPush(submitRsp.pArray, &submitBlkRsp);
    }
H
Hongze Cheng 已提交
988
  }
H
Haojun Liao 已提交
989 990

  if (taosArrayGetSize(newTbUids) > 0) {
L
Liu Jicong 已提交
991 992
    vDebug("vgId:%d, add %d table into query table list in handling submit", TD_VID(pVnode),
           (int32_t)taosArrayGetSize(newTbUids));
H
Haojun Liao 已提交
993 994
  }

995
  tqUpdateTbUidList(pVnode->pTq, newTbUids, true);
996

H
Hongze Cheng 已提交
997
_exit:
998
  taosArrayDestroy(newTbUids);
H
Hongze Cheng 已提交
999 1000 1001 1002 1003 1004 1005
  tEncodeSize(tEncodeSSubmitRsp, &submitRsp, tsize, ret);
  pRsp->pCont = rpcMallocCont(tsize);
  pRsp->contLen = tsize;
  tEncoderInit(&encoder, pRsp->pCont, tsize);
  tEncodeSSubmitRsp(&encoder, &submitRsp);
  tEncoderClear(&encoder);

1006
  taosArrayDestroyEx(submitRsp.pArray, tFreeSSubmitBlkRsp);
H
Hongze Cheng 已提交
1007

1008
  // TODO: the partial success scenario and the error case
H
Hongze Cheng 已提交
1009 1010
  // => If partial success, extract the success submitted rows and reconstruct a new submit msg, and push to level
  // 1/level 2.
1011
  // TODO: refactor
C
Cary Xu 已提交
1012
  if ((terrno == TSDB_CODE_SUCCESS) && (pRsp->code == TSDB_CODE_SUCCESS)) {
1013
    statis.nBatchInsertSuccess = 1;
L
Liu Jicong 已提交
1014
    tdProcessRSmaSubmit(pVnode->pSma, pReq, STREAM_INPUT__DATA_SUBMIT);
1015
  }
C
Cary Xu 已提交
1016

1017
  // N.B. not strict as the following procedure is not atomic
1018 1019 1020
  atomic_add_fetch_64(&pVnode->statis.nInsert, submitRsp.numOfRows);
  atomic_add_fetch_64(&pVnode->statis.nInsertSuccess, submitRsp.affectedRows);
  atomic_add_fetch_64(&pVnode->statis.nBatchInsert, statis.nBatchInsert);
1021
  atomic_add_fetch_64(&pVnode->statis.nBatchInsertSuccess, statis.nBatchInsertSuccess);
1022

S
Shengliang Guan 已提交
1023
  vDebug("vgId:%d, submit success, index:%" PRId64, pVnode->config.vgId, version);
H
Hongze Cheng 已提交
1024
  return 0;
L
Liu Jicong 已提交
1025
}
1026

1027
static int32_t vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
1028
  SVCreateTSmaReq req = {0};
C
Cary Xu 已提交
1029
  SDecoder        coder = {0};
1030

C
Cary Xu 已提交
1031 1032 1033 1034 1035 1036
  if (pRsp) {
    pRsp->msgType = TDMT_VND_CREATE_SMA_RSP;
    pRsp->code = TSDB_CODE_SUCCESS;
    pRsp->pCont = NULL;
    pRsp->contLen = 0;
  }
1037 1038 1039 1040 1041

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

  if (tDecodeSVCreateTSmaReq(&coder, &req) < 0) {
C
Cary Xu 已提交
1042 1043
    terrno = TSDB_CODE_MSG_DECODE_ERROR;
    if (pRsp) pRsp->code = terrno;
1044
    goto _err;
1045
  }
C
Cary Xu 已提交
1046

C
Cary Xu 已提交
1047
  if (tdProcessTSmaCreate(pVnode->pSma, version, (const char *)&req) < 0) {
C
Cary Xu 已提交
1048
    if (pRsp) pRsp->code = terrno;
1049
    goto _err;
1050
  }
C
Cary Xu 已提交
1051

1052
  tDecoderClear(&coder);
S
Shengliang Guan 已提交
1053
  vDebug("vgId:%d, success to create tsma %s:%" PRIi64 " version %" PRIi64 " for table %" PRIi64, TD_VID(pVnode),
C
Cary Xu 已提交
1054
         req.indexName, req.indexUid, version, req.tableUid);
H
Hongze Cheng 已提交
1055
  return 0;
1056 1057 1058

_err:
  tDecoderClear(&coder);
S
Shengliang Guan 已提交
1059
  vError("vgId:%d, failed to create tsma %s:%" PRIi64 " version %" PRIi64 "for table %" PRIi64 " since %s",
C
Cary Xu 已提交
1060
         TD_VID(pVnode), req.indexName, req.indexUid, version, req.tableUid, terrstr());
1061
  return -1;
L
Liu Jicong 已提交
1062
}
C
Cary Xu 已提交
1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074

/**
 * @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);
}
1075 1076 1077 1078 1079 1080 1081 1082 1083

static int32_t vnodeProcessAlterConfirmReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
  vInfo("vgId:%d, alter replica confim msg is processed", TD_VID(pVnode));
  pRsp->msgType = TDMT_VND_ALTER_CONFIRM_RSP;
  pRsp->code = TSDB_CODE_SUCCESS;
  pRsp->pCont = NULL;
  pRsp->contLen = 0;

  return 0;
1084
}
S
Shengliang Guan 已提交
1085

1086
static int32_t vnodeProcessAlterHashRangeReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
S
Shengliang Guan 已提交
1087 1088
  vInfo("vgId:%d, alter hashrange msg will be processed", TD_VID(pVnode));

1089 1090
  // todo
  // 1. stop work
S
Shengliang Guan 已提交
1091 1092 1093
  // 2. adjust hash range / compact / remove wals / rename vgroups
  // 3. reload sync
  return 0;
M
Minghao Li 已提交
1094
}
H
Hongze Cheng 已提交
1095

1096
static int32_t vnodeProcessAlterConfigReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
S
Shengliang Guan 已提交
1097 1098
  bool walChanged = false;
  bool tsdbChanged = false;
1099

S
Shengliang Guan 已提交
1100 1101
  SAlterVnodeConfigReq req = {0};
  if (tDeserializeSAlterVnodeConfigReq(pReq, len, &req) != 0) {
1102 1103 1104 1105
    terrno = TSDB_CODE_INVALID_MSG;
    return TSDB_CODE_INVALID_MSG;
  }

1106
  vInfo("vgId:%d, start to alter vnode config, page:%d pageSize:%d buffer:%d szPage:%d szBuf:%" PRIu64
S
Shengliang Guan 已提交
1107
        " cacheLast:%d cacheLastSize:%d days:%d keep0:%d keep1:%d keep2:%d fsync:%d level:%d",
1108 1109
        TD_VID(pVnode), req.pages, req.pageSize, req.buffer, req.pageSize * 1024, (uint64_t)req.buffer * 1024 * 1024,
        req.cacheLast, req.cacheLastSize, req.daysPerFile, req.daysToKeep0, req.daysToKeep1, req.daysToKeep2,
S
Shengliang Guan 已提交
1110
        req.walFsyncPeriod, req.walLevel);
1111 1112 1113

  if (pVnode->config.cacheLastSize != req.cacheLastSize) {
    pVnode->config.cacheLastSize = req.cacheLastSize;
1114 1115
    tsdbCacheSetCapacity(pVnode, (size_t)pVnode->config.cacheLastSize * 1024 * 1024);
  }
1116

1117
  if (pVnode->config.szBuf != req.buffer * 1024LL * 1024LL) {
S
Shengliang Guan 已提交
1118
    vInfo("vgId:%d, vnode buffer is changed from %" PRId64 " to %" PRId64, TD_VID(pVnode), pVnode->config.szBuf,
1119
          (uint64_t)(req.buffer * 1024LL * 1024LL));
1120
    pVnode->config.szBuf = req.buffer * 1024LL * 1024LL;
H
Hongze Cheng 已提交
1121 1122
  }

1123 1124
  if (pVnode->config.szCache != req.pages) {
    if (metaAlterCache(pVnode->pMeta, req.pages) < 0) {
S
Shengliang Guan 已提交
1125
      vError("vgId:%d, failed to change vnode pages from %d to %d failed since %s", TD_VID(pVnode),
1126
             pVnode->config.szCache, req.pages, tstrerror(errno));
H
Hongze Cheng 已提交
1127 1128
      return errno;
    } else {
S
Shengliang Guan 已提交
1129
      vInfo("vgId:%d, vnode pages is changed from %d to %d", TD_VID(pVnode), pVnode->config.szCache, req.pages);
1130
      pVnode->config.szCache = req.pages;
H
Hongze Cheng 已提交
1131
    }
H
Hongze Cheng 已提交
1132 1133
  }

1134 1135
  if (pVnode->config.cacheLast != req.cacheLast) {
    pVnode->config.cacheLast = req.cacheLast;
1136 1137
  }

1138 1139
  if (pVnode->config.walCfg.fsyncPeriod != req.walFsyncPeriod) {
    pVnode->config.walCfg.fsyncPeriod = req.walFsyncPeriod;
1140 1141 1142 1143

    walChanged = true;
  }

1144 1145
  if (pVnode->config.walCfg.level != req.walLevel) {
    pVnode->config.walCfg.level = req.walLevel;
1146 1147 1148 1149

    walChanged = true;
  }

1150 1151
  if (pVnode->config.tsdbCfg.keep0 != req.daysToKeep0) {
    pVnode->config.tsdbCfg.keep0 = req.daysToKeep0;
1152
    if (!VND_IS_RSMA(pVnode)) {
M
Minglei Jin 已提交
1153
      tsdbChanged = true;
1154 1155 1156
    }
  }

1157 1158
  if (pVnode->config.tsdbCfg.keep1 != req.daysToKeep1) {
    pVnode->config.tsdbCfg.keep1 = req.daysToKeep1;
1159
    if (!VND_IS_RSMA(pVnode)) {
M
Minglei Jin 已提交
1160
      tsdbChanged = true;
1161 1162 1163
    }
  }

1164 1165
  if (pVnode->config.tsdbCfg.keep2 != req.daysToKeep2) {
    pVnode->config.tsdbCfg.keep2 = req.daysToKeep2;
1166
    if (!VND_IS_RSMA(pVnode)) {
M
Minglei Jin 已提交
1167
      tsdbChanged = true;
1168 1169 1170 1171
    }
  }

  if (walChanged) {
M
Minglei Jin 已提交
1172 1173 1174 1175 1176
    walAlter(pVnode->pWal, &pVnode->config.walCfg);
  }

  if (tsdbChanged) {
    tsdbSetKeepCfg(pVnode->pTsdb, &pVnode->config.tsdbCfg);
1177 1178
  }

1179 1180 1181
  return 0;
}

1182 1183 1184 1185 1186 1187
static int32_t vnodeProcessBatchDeleteReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
  SBatchDeleteReq deleteReq;
  SDecoder        decoder;
  tDecoderInit(&decoder, pReq, len);
  tDecodeSBatchDeleteReq(&decoder, &deleteReq);

1188 1189 1190
  SMetaReader mr = {0};
  metaReaderInit(&mr, pVnode->pMeta, 0);

1191 1192 1193
  int32_t sz = taosArrayGetSize(deleteReq.deleteReqs);
  for (int32_t i = 0; i < sz; i++) {
    SSingleDeleteReq *pOneReq = taosArrayGet(deleteReq.deleteReqs, i);
1194 1195 1196 1197 1198 1199 1200 1201
    char             *name = pOneReq->tbname;
    if (metaGetTableEntryByName(&mr, name) < 0) {
      vDebug("stream delete msg, skip vgId:%d since no table: %s", pVnode->config.vgId, name);
      continue;
    }

    int64_t uid = mr.me.uid;

L
Liu Jicong 已提交
1202
    int32_t code = tsdbDeleteTableData(pVnode->pTsdb, version, deleteReq.suid, uid, pOneReq->startTs, pOneReq->endTs);
1203 1204 1205
    if (code < 0) {
      terrno = code;
      vError("vgId:%d, delete error since %s, suid:%" PRId64 ", uid:%" PRId64 ", start ts:%" PRId64 ", end ts:%" PRId64,
L
Liu Jicong 已提交
1206
             TD_VID(pVnode), terrstr(), deleteReq.suid, uid, pOneReq->startTs, pOneReq->endTs);
1207
    }
1208 1209

    tDecoderClear(&mr.coder);
1210
  }
1211
  metaReaderClear(&mr);
1212
  taosArrayDestroy(deleteReq.deleteReqs);
1213 1214 1215
  return 0;
}

H
Hongze Cheng 已提交
1216 1217 1218 1219 1220
static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
  int32_t     code = 0;
  SDecoder   *pCoder = &(SDecoder){0};
  SDeleteRes *pRes = &(SDeleteRes){0};

wmmhello's avatar
wmmhello 已提交
1221 1222 1223 1224 1225
  pRsp->msgType = TDMT_VND_DELETE_RSP;
  pRsp->pCont = NULL;
  pRsp->contLen = 0;
  pRsp->code = TSDB_CODE_SUCCESS;

H
Hongze Cheng 已提交
1226 1227 1228 1229 1230 1231 1232 1233
  pRes->uidList = taosArrayInit(0, sizeof(tb_uid_t));
  if (pRes->uidList == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _err;
  }

  tDecoderInit(pCoder, pReq, len);
  tDecodeDeleteRes(pCoder, pRes);
L
Liu Jicong 已提交
1234
  ASSERT(taosArrayGetSize(pRes->uidList) == 0 || (pRes->skey != 0 && pRes->ekey != 0));
H
Hongze Cheng 已提交
1235 1236 1237 1238 1239 1240 1241 1242 1243

  for (int32_t iUid = 0; iUid < taosArrayGetSize(pRes->uidList); iUid++) {
    code = tsdbDeleteTableData(pVnode->pTsdb, version, pRes->suid, *(uint64_t *)taosArrayGet(pRes->uidList, iUid),
                               pRes->skey, pRes->ekey);
    if (code) goto _err;
  }

  tDecoderClear(pCoder);
  taosArrayDestroy(pRes->uidList);
wmmhello's avatar
wmmhello 已提交
1244 1245

  SVDeleteRsp rsp = {.affectedRows = pRes->affectedRows};
L
Liu Jicong 已提交
1246
  int32_t     ret = 0;
wmmhello's avatar
wmmhello 已提交
1247 1248
  tEncodeSize(tEncodeSVDeleteRsp, &rsp, pRsp->contLen, ret);
  pRsp->pCont = rpcMallocCont(pRsp->contLen);
L
Liu Jicong 已提交
1249
  SEncoder ec = {0};
wmmhello's avatar
wmmhello 已提交
1250 1251 1252
  tEncoderInit(&ec, pRsp->pCont, pRsp->contLen);
  tEncodeSVDeleteRsp(&ec, &rsp);
  tEncoderClear(&ec);
H
Hongze Cheng 已提交
1253 1254 1255 1256
  return code;

_err:
  return code;
M
Minglei Jin 已提交
1257
}