vnodeSvr.c 38.3 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
  vDebug("vgId:%d, start to process write request %s, index:%" PRId64, TD_VID(pVnode), TMSG_INFO(pMsg->msgType),
H
Hongze Cheng 已提交
182
         version);
H
Hongze Cheng 已提交
183

H
Hongze Cheng 已提交
184
  pVnode->state.applied = version;
H
Hongze Cheng 已提交
185
  pVnode->state.applyTerm = pMsg->info.conn.applyTerm;
H
Hongze Cheng 已提交
186

H
Hongze Cheng 已提交
187 188 189
  // skip header
  pReq = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
  len = pMsg->contLen - sizeof(SMsgHead);
H
Hongze Cheng 已提交
190 191

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

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

295 296
  walApplyVer(pVnode->pWal, version);

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

H
Hongze Cheng 已提交
302
  // commit if need
H
Hongze Cheng 已提交
303
  if (vnodeShouldCommit(pVnode)) {
H
Hongze Cheng 已提交
304
  _do_commit:
S
Shengliang Guan 已提交
305
    vInfo("vgId:%d, commit at version %" PRId64, TD_VID(pVnode), version);
H
Hongze Cheng 已提交
306
    // commit current change
307
    if (vnodeCommit(pVnode) < 0) {
H
Hongze Cheng 已提交
308 309
      vError("vgId:%d, failed to commit vnode since %s.", TD_VID(pVnode), tstrerror(terrno));
      goto _err;
310
    }
H
Hongze Cheng 已提交
311 312

    // start a new one
313
    if (vnodeBegin(pVnode) < 0) {
H
Hongze Cheng 已提交
314 315
      vError("vgId:%d, failed to begin vnode since %s.", TD_VID(pVnode), tstrerror(terrno));
      goto _err;
316
    }
H
Hongze Cheng 已提交
317 318 319
  }

  return 0;
H
Hongze Cheng 已提交
320 321

_err:
S
Shengliang Guan 已提交
322
  vError("vgId:%d, process %s request failed since %s, version:%" PRId64, TD_VID(pVnode), TMSG_INFO(pMsg->msgType),
H
Hongze Cheng 已提交
323 324
         tstrerror(terrno), version);
  return -1;
H
Hongze Cheng 已提交
325 326
}

327
int32_t vnodePreprocessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
D
dapan1121 已提交
328
  if (TDMT_SCH_QUERY != pMsg->msgType && TDMT_SCH_MERGE_QUERY != pMsg->msgType) {
D
dapan1121 已提交
329 330 331
    return 0;
  }

332
  return qWorkerPreprocessQueryMsg(pVnode->pQuery, pMsg, TDMT_SCH_QUERY == pMsg->msgType);
D
dapan1121 已提交
333 334 335
}

int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
D
dapan 已提交
336
  vTrace("message in vnode query queue is processing");
337 338
  // if ((pMsg->msgType == TDMT_SCH_QUERY) && !vnodeIsLeader(pVnode)) {
  if ((pMsg->msgType == TDMT_SCH_QUERY) && !syncIsReadyForRead(pVnode->sync)) {
339 340 341 342
    vnodeRedirectRpcMsg(pVnode, pMsg);
    return 0;
  }

343
  SReadHandle handle = {.meta = pVnode->pMeta, .config = &pVnode->config, .vnode = pVnode, .pMsgCb = &pVnode->msgCb};
H
Hongze Cheng 已提交
344
  switch (pMsg->msgType) {
D
dapan1121 已提交
345
    case TDMT_SCH_QUERY:
D
dapan1121 已提交
346
    case TDMT_SCH_MERGE_QUERY:
D
dapan1121 已提交
347
      return qWorkerProcessQueryMsg(&handle, pVnode->pQuery, pMsg, 0);
D
dapan1121 已提交
348
    case TDMT_SCH_QUERY_CONTINUE:
D
dapan1121 已提交
349
      return qWorkerProcessCQueryMsg(&handle, pVnode->pQuery, pMsg, 0);
H
Hongze Cheng 已提交
350 351
    default:
      vError("unknown msg type:%d in query queue", pMsg->msgType);
352
      return TSDB_CODE_APP_ERROR;
H
Hongze Cheng 已提交
353 354 355
  }
}

356
int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
S
Shengliang Guan 已提交
357
  vTrace("vgId:%d, msg:%p in fetch queue is processing", pVnode->config.vgId, pMsg);
L
Liu Jicong 已提交
358 359
  if ((pMsg->msgType == TDMT_SCH_FETCH || pMsg->msgType == TDMT_VND_TABLE_META || pMsg->msgType == TDMT_VND_TABLE_CFG ||
       pMsg->msgType == TDMT_VND_BATCH_META) &&
360 361
      !syncIsReadyForRead(pVnode->sync)) {
    //      !vnodeIsLeader(pVnode)) {
362 363 364 365
    vnodeRedirectRpcMsg(pVnode, pMsg);
    return 0;
  }

L
Liu Jicong 已提交
366
  if (pMsg->msgType == TDMT_VND_TMQ_CONSUME && !pVnode->restored) {
L
Liu Jicong 已提交
367 368 369 370
    vnodeRedirectRpcMsg(pVnode, pMsg);
    return 0;
  }

H
Hongze Cheng 已提交
371
  switch (pMsg->msgType) {
D
dapan1121 已提交
372
    case TDMT_SCH_FETCH:
D
dapan1121 已提交
373
    case TDMT_SCH_MERGE_FETCH:
D
dapan1121 已提交
374
      return qWorkerProcessFetchMsg(pVnode, pVnode->pQuery, pMsg, 0);
D
dapan1121 已提交
375
    case TDMT_SCH_FETCH_RSP:
D
dapan1121 已提交
376
      return qWorkerProcessRspMsg(pVnode, pVnode->pQuery, pMsg, 0);
L
Liu Jicong 已提交
377 378
    // case TDMT_SCH_CANCEL_TASK:
    //   return qWorkerProcessCancelMsg(pVnode, pVnode->pQuery, pMsg, 0);
D
dapan1121 已提交
379
    case TDMT_SCH_DROP_TASK:
D
dapan1121 已提交
380
      return qWorkerProcessDropMsg(pVnode, pVnode->pQuery, pMsg, 0);
D
dapan1121 已提交
381
    case TDMT_SCH_QUERY_HEARTBEAT:
D
dapan1121 已提交
382
      return qWorkerProcessHbMsg(pVnode, pVnode->pQuery, pMsg, 0);
H
Hongze Cheng 已提交
383
    case TDMT_VND_TABLE_META:
D
dapan1121 已提交
384
      return vnodeGetTableMeta(pVnode, pMsg, true);
D
dapan1121 已提交
385
    case TDMT_VND_TABLE_CFG:
D
dapan1121 已提交
386 387 388
      return vnodeGetTableCfg(pVnode, pMsg, true);
    case TDMT_VND_BATCH_META:
      return vnodeGetBatchMeta(pVnode, pMsg);
L
Liu Jicong 已提交
389
    case TDMT_VND_TMQ_CONSUME:
L
Liu Jicong 已提交
390
      return tqProcessPollReq(pVnode->pTq, pMsg);
391 392
    case TDMT_STREAM_TASK_RUN:
      return tqProcessTaskRunReq(pVnode->pTq, pMsg);
393
#if 1
394
    case TDMT_STREAM_TASK_DISPATCH:
L
Liu Jicong 已提交
395
      return tqProcessTaskDispatchReq(pVnode->pTq, pMsg, true);
L
Liu Jicong 已提交
396
#endif
397 398
    case TDMT_STREAM_TASK_CHECK:
      return tqProcessStreamTaskCheckReq(pVnode->pTq, pMsg);
399
    case TDMT_STREAM_TASK_DISPATCH_RSP:
L
Liu Jicong 已提交
400
      return tqProcessTaskDispatchRsp(pVnode->pTq, pMsg);
L
Liu Jicong 已提交
401 402
    case TDMT_STREAM_RETRIEVE:
      return tqProcessTaskRetrieveReq(pVnode->pTq, pMsg);
L
Liu Jicong 已提交
403 404
    case TDMT_STREAM_RETRIEVE_RSP:
      return tqProcessTaskRetrieveRsp(pVnode->pTq, pMsg);
L
Liu Jicong 已提交
405 406 407 408 409 410
    case TDMT_VND_STREAM_RECOVER_STEP1:
      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 已提交
411 412
    default:
      vError("unknown msg type:%d in fetch queue", pMsg->msgType);
413
      return TSDB_CODE_APP_ERROR;
H
Hongze Cheng 已提交
414 415 416 417 418 419
  }
}

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

D
dapan1121 已提交
424
void vnodeUpdateMetaRsp(SVnode *pVnode, STableMetaRsp *pMetaRsp) {
425 426 427
  if (NULL == pMetaRsp) {
    return;
  }
L
Liu Jicong 已提交
428

D
dapan1121 已提交
429 430 431 432 433 434
  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 已提交
435
static int32_t vnodeProcessTrimReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
436
  int32_t     code = 0;
S
Shengliang Guan 已提交
437 438
  SVTrimDbReq trimReq = {0};

H
Hongze Cheng 已提交
439 440 441 442
  // decode
  if (tDeserializeSVTrimDbReq(pReq, len, &trimReq) != 0) {
    code = TSDB_CODE_INVALID_MSG;
    goto _exit;
S
Shengliang Guan 已提交
443 444
  }

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

H
Hongze Cheng 已提交
447 448 449 450
  // process
  code = tsdbDoRetention(pVnode->pTsdb, trimReq.timestamp);
  if (code) goto _exit;

C
Cary Xu 已提交
451 452 453
  code = smaDoRetention(pVnode->pSma, trimReq.timestamp);
  if (code) goto _exit;

H
Hongze Cheng 已提交
454 455
_exit:
  return code;
S
Shengliang Guan 已提交
456 457
}

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

S
Shengliang Guan 已提交
462 463 464 465 466 467
  SVDropTtlTableReq ttlReq = {0};
  if (tDeserializeSVDropTtlTableReq(pReq, len, &ttlReq) != 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    goto end;
  }

468
  vDebug("vgId:%d, drop ttl table req will be processed, time:%d", pVnode->config.vgId, ttlReq.timestamp);
S
Shengliang Guan 已提交
469
  int32_t ret = metaTtlDropTable(pVnode->pMeta, ttlReq.timestamp, tbUids);
L
Liu Jicong 已提交
470
  if (ret != 0) {
471 472
    goto end;
  }
L
Liu Jicong 已提交
473
  if (taosArrayGetSize(tbUids) > 0) {
wmmhello's avatar
wmmhello 已提交
474 475
    tqUpdateTbUidList(pVnode->pTq, tbUids, false);
  }
476 477 478 479 480 481

end:
  taosArrayDestroy(tbUids);
  return ret;
}

482
static int32_t vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
483
  SVCreateStbReq req = {0};
H
Hongze Cheng 已提交
484
  SDecoder       coder;
H
Hongze Cheng 已提交
485

H
Hongze Cheng 已提交
486 487 488 489 490 491
  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 已提交
492
  tDecoderInit(&coder, pReq, len);
H
Hongze Cheng 已提交
493 494

  if (tDecodeSVCreateStbReq(&coder, &req) < 0) {
H
Hongze Cheng 已提交
495 496
    pRsp->code = terrno;
    goto _err;
H
Hongze Cheng 已提交
497 498
  }

H
Hongze Cheng 已提交
499
  if (metaCreateSTable(pVnode->pMeta, version, &req) < 0) {
H
Hongze Cheng 已提交
500 501
    pRsp->code = terrno;
    goto _err;
H
Hongze Cheng 已提交
502 503
  }

504
  if (tdProcessRSmaCreate(pVnode->pSma, &req) < 0) {
505 506 507
    pRsp->code = terrno;
    goto _err;
  }
C
Cary Xu 已提交
508

H
Hongze Cheng 已提交
509
  tDecoderClear(&coder);
H
Hongze Cheng 已提交
510
  return 0;
H
Hongze Cheng 已提交
511 512

_err:
H
Hongze Cheng 已提交
513
  tDecoderClear(&coder);
H
Hongze Cheng 已提交
514
  return -1;
H
Hongze Cheng 已提交
515 516
}

517
static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
518
  SDecoder           decoder = {0};
519
  SEncoder           encoder = {0};
520
  int32_t            rcode = 0;
H
Hongze Cheng 已提交
521 522
  SVCreateTbBatchReq req = {0};
  SVCreateTbReq     *pCreateReq;
H
Hongze Cheng 已提交
523 524 525
  SVCreateTbBatchRsp rsp = {0};
  SVCreateTbRsp      cRsp = {0};
  char               tbName[TSDB_TABLE_FNAME_LEN];
C
Cary Xu 已提交
526
  STbUidStore       *pStore = NULL;
527
  SArray            *tbUids = NULL;
H
Hongze Cheng 已提交
528 529

  pRsp->msgType = TDMT_VND_CREATE_TABLE_RSP;
H
Hongze Cheng 已提交
530 531 532
  pRsp->code = TSDB_CODE_SUCCESS;
  pRsp->pCont = NULL;
  pRsp->contLen = 0;
H
Hongze Cheng 已提交
533

H
Hongze Cheng 已提交
534
  // decode
H
Hongze Cheng 已提交
535 536
  tDecoderInit(&decoder, pReq, len);
  if (tDecodeSVCreateTbBatchReq(&decoder, &req) < 0) {
H
Hongze Cheng 已提交
537 538 539 540
    rcode = -1;
    terrno = TSDB_CODE_INVALID_MSG;
    goto _exit;
  }
H
Hongze Cheng 已提交
541

H
Hongze Cheng 已提交
542
  rsp.pArray = taosArrayInit(req.nReqs, sizeof(cRsp));
543 544
  tbUids = taosArrayInit(req.nReqs, sizeof(int64_t));
  if (rsp.pArray == NULL || tbUids == NULL) {
H
Hongze Cheng 已提交
545 546 547 548 549
    rcode = -1;
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

H
Hongze Cheng 已提交
550
  // loop to create table
551
  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
H
Hongze Cheng 已提交
552
    pCreateReq = req.pReqs + iReq;
D
dapan1121 已提交
553
    memset(&cRsp, 0, sizeof(cRsp));
H
Hongze Cheng 已提交
554

C
Cary Xu 已提交
555 556 557 558
    if ((terrno = grantCheck(TSDB_GRANT_TIMESERIES)) < 0) {
      rcode = -1;
      goto _exit;
    }
L
Liu Jicong 已提交
559

wafwerar's avatar
wafwerar 已提交
560 561 562 563 564
    if ((terrno = grantCheck(TSDB_GRANT_TABLE)) < 0) {
      rcode = -1;
      goto _exit;
    }

H
Hongze Cheng 已提交
565 566 567 568 569 570 571 572 573
    // 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
574
    if (metaCreateTable(pVnode->pMeta, version, pCreateReq, &cRsp.pMeta) < 0) {
H
Hongze Cheng 已提交
575 576 577 578 579
      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 已提交
580
    } else {
H
Hongze Cheng 已提交
581
      cRsp.code = TSDB_CODE_SUCCESS;
582
      tdFetchTbUidList(pVnode->pSma, &pStore, pCreateReq->ctb.suid, pCreateReq->uid);
583
      taosArrayPush(tbUids, &pCreateReq->uid);
L
Liu Jicong 已提交
584
      vnodeUpdateMetaRsp(pVnode, cRsp.pMeta);
H
Hongze Cheng 已提交
585
    }
H
Hongze Cheng 已提交
586 587

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

H
Haojun Liao 已提交
590
  vDebug("vgId:%d, add %d new created tables into query table list", TD_VID(pVnode), (int32_t)taosArrayGetSize(tbUids));
591
  tqUpdateTbUidList(pVnode->pTq, tbUids, true);
592
  if (tdUpdateTbUidList(pVnode->pSma, pStore, true) < 0) {
C
Cary Xu 已提交
593 594
    goto _exit;
  }
595
  tdUidStoreFree(pStore);
C
Cary Xu 已提交
596

H
Hongze Cheng 已提交
597
  // prepare rsp
598
  int32_t ret = 0;
wafwerar's avatar
wafwerar 已提交
599
  tEncodeSize(tEncodeSVCreateTbBatchRsp, &rsp, pRsp->contLen, ret);
H
Hongze Cheng 已提交
600 601 602 603 604 605
  pRsp->pCont = rpcMallocCont(pRsp->contLen);
  if (pRsp->pCont == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    rcode = -1;
    goto _exit;
  }
H
Hongze Cheng 已提交
606 607
  tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
  tEncodeSVCreateTbBatchRsp(&encoder, &rsp);
H
Hongze Cheng 已提交
608

H
Hongze Cheng 已提交
609
_exit:
wmmhello's avatar
wmmhello 已提交
610 611 612 613
  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
    pCreateReq = req.pReqs + iReq;
    taosArrayDestroy(pCreateReq->ctb.tagName);
  }
614
  taosArrayDestroyEx(rsp.pArray, tFreeSVCreateTbRsp);
615
  taosArrayDestroy(tbUids);
H
Hongze Cheng 已提交
616 617
  tDecoderClear(&decoder);
  tEncoderClear(&encoder);
H
Hongze Cheng 已提交
618
  return rcode;
H
Hongze Cheng 已提交
619 620
}

621
static int32_t vnodeProcessAlterStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
622 623 624 625 626 627 628 629 630 631 632 633 634 635 636
  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 已提交
637
  }
H
Hongze Cheng 已提交
638 639 640 641 642 643 644 645 646

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

  tDecoderClear(&dc);

H
Hongze Cheng 已提交
647 648 649
  return 0;
}

650
static int32_t vnodeProcessDropStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
651
  SVDropStbReq req = {0};
652
  int32_t      rcode = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
653
  SDecoder     decoder = {0};
654
  SArray      *tbUidList = NULL;
H
Hongze Cheng 已提交
655 656 657 658 659 660

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

  // decode request
H
Hongze Cheng 已提交
661 662
  tDecoderInit(&decoder, pReq, len);
  if (tDecodeSVDropStbReq(&decoder, &req) < 0) {
H
Hongze Cheng 已提交
663 664 665 666 667
    rcode = TSDB_CODE_INVALID_MSG;
    goto _exit;
  }

  // process request
668 669 670 671 672 673 674 675
  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) {
676 677 678
    rcode = terrno;
    goto _exit;
  }
H
Hongze Cheng 已提交
679

680 681 682 683 684
  if (tdProcessRSmaDrop(pVnode->pSma, &req) < 0) {
    rcode = terrno;
    goto _exit;
  }

H
Hongze Cheng 已提交
685 686
  // return rsp
_exit:
687
  if (tbUidList) taosArrayDestroy(tbUidList);
H
Hongze Cheng 已提交
688
  pRsp->code = rcode;
H
Hongze Cheng 已提交
689
  tDecoderClear(&decoder);
H
Hongze Cheng 已提交
690 691 692
  return 0;
}

693
static int32_t vnodeProcessAlterTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
D
dapan1121 已提交
694 695 696
  SVAlterTbReq  vAlterTbReq = {0};
  SVAlterTbRsp  vAlterTbRsp = {0};
  SDecoder      dc = {0};
697 698
  int32_t       rcode = 0;
  int32_t       ret;
D
dapan1121 已提交
699 700
  SEncoder      ec = {0};
  STableMetaRsp vMetaRsp = {0};
H
Hongze Cheng 已提交
701 702 703 704 705 706 707 708 709 710

  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 已提交
711
    vAlterTbRsp.code = TSDB_CODE_INVALID_MSG;
H
Hongze Cheng 已提交
712
    tDecoderClear(&dc);
H
Hongze Cheng 已提交
713 714
    rcode = -1;
    goto _exit;
H
Hongze Cheng 已提交
715 716 717
  }

  // process
D
dapan1121 已提交
718
  if (metaAlterTable(pVnode->pMeta, version, &vAlterTbReq, &vMetaRsp) < 0) {
wmmhello's avatar
wmmhello 已提交
719
    vAlterTbRsp.code = terrno;
H
Hongze Cheng 已提交
720
    tDecoderClear(&dc);
H
Hongze Cheng 已提交
721 722
    rcode = -1;
    goto _exit;
H
Hongze Cheng 已提交
723 724
  }
  tDecoderClear(&dc);
H
Hongze Cheng 已提交
725

D
dapan1121 已提交
726 727 728 729 730
  if (NULL != vMetaRsp.pSchemas) {
    vnodeUpdateMetaRsp(pVnode, &vMetaRsp);
    vAlterTbRsp.pMeta = &vMetaRsp;
  }

H
Hongze Cheng 已提交
731 732 733 734 735 736
_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 已提交
737 738 739
  if (vMetaRsp.pSchemas) {
    taosMemoryFree(vMetaRsp.pSchemas);
  }
H
Hongze Cheng 已提交
740 741 742
  return 0;
}

743
static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
744 745
  SVDropTbBatchReq req = {0};
  SVDropTbBatchRsp rsp = {0};
H
Hongze Cheng 已提交
746
  SDecoder         decoder = {0};
H
Hongze Cheng 已提交
747
  SEncoder         encoder = {0};
748
  int32_t          ret;
749
  SArray          *tbUids = NULL;
750
  STbUidStore     *pStore = NULL;
H
Hongze Cheng 已提交
751

H
Hongze Cheng 已提交
752
  pRsp->msgType = TDMT_VND_DROP_TABLE_RSP;
H
Hongze Cheng 已提交
753 754 755
  pRsp->pCont = NULL;
  pRsp->contLen = 0;
  pRsp->code = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
756 757

  // decode req
H
Hongze Cheng 已提交
758 759
  tDecoderInit(&decoder, pReq, len);
  ret = tDecodeSVDropTbBatchReq(&decoder, &req);
H
Hongze Cheng 已提交
760 761 762 763 764
  if (ret < 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    pRsp->code = terrno;
    goto _exit;
  }
H
Hongze Cheng 已提交
765 766

  // process req
767
  tbUids = taosArrayInit(req.nReqs, sizeof(int64_t));
H
Hongze Cheng 已提交
768
  rsp.pArray = taosArrayInit(req.nReqs, sizeof(SVDropTbRsp));
769 770
  if (tbUids == NULL || rsp.pArray == NULL) goto _exit;

771
  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
H
Hongze Cheng 已提交
772 773
    SVDropTbReq *pDropTbReq = req.pReqs + iReq;
    SVDropTbRsp  dropTbRsp = {0};
774
    tb_uid_t     tbUid = 0;
H
Hongze Cheng 已提交
775

H
Hongze Cheng 已提交
776
    /* code */
777
    ret = metaDropTable(pVnode->pMeta, version, pDropTbReq, tbUids, &tbUid);
H
Hongze Cheng 已提交
778
    if (ret < 0) {
779
      if (pDropTbReq->igNotExists && terrno == TSDB_CODE_TDB_TABLE_NOT_EXIST) {
H
Hongze Cheng 已提交
780 781 782 783
        dropTbRsp.code = TSDB_CODE_SUCCESS;
      } else {
        dropTbRsp.code = terrno;
      }
H
Hongze Cheng 已提交
784
    } else {
H
Hongze Cheng 已提交
785
      dropTbRsp.code = TSDB_CODE_SUCCESS;
786
      if (tbUid > 0) tdFetchTbUidList(pVnode->pSma, &pStore, pDropTbReq->suid, tbUid);
H
Hongze Cheng 已提交
787 788 789 790 791
    }

    taosArrayPush(rsp.pArray, &dropTbRsp);
  }

792
  tqUpdateTbUidList(pVnode->pTq, tbUids, false);
793
  tdUpdateTbUidList(pVnode->pSma, pStore, false);
794

H
Hongze Cheng 已提交
795
_exit:
796
  taosArrayDestroy(tbUids);
797
  tdUidStoreFree(pStore);
H
Hongze Cheng 已提交
798
  tDecoderClear(&decoder);
H
Hongze Cheng 已提交
799 800 801 802 803
  tEncodeSize(tEncodeSVDropTbBatchRsp, &rsp, pRsp->contLen, ret);
  pRsp->pCont = rpcMallocCont(pRsp->contLen);
  tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
  tEncodeSVDropTbBatchRsp(&encoder, &rsp);
  tEncoderClear(&encoder);
804
  taosArrayDestroy(rsp.pArray);
H
Hongze Cheng 已提交
805 806 807
  return 0;
}

808 809
static int32_t vnodeDebugPrintSingleSubmitMsg(SMeta *pMeta, SSubmitBlk *pBlock, SSubmitMsgIter *msgIter,
                                              const char *tags) {
D
dapan 已提交
810 811 812 813
  SSubmitBlkIter blkIter = {0};
  STSchema      *pSchema = NULL;
  tb_uid_t       suid = 0;
  STSRow        *row = NULL;
C
Cary Xu 已提交
814
  int32_t        rv = -1;
D
dapan 已提交
815 816 817

  tInitSubmitBlkIter(msgIter, pBlock, &blkIter);
  if (blkIter.row == NULL) return 0;
818 819 820 821 822

  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 已提交
823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838
  }
  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;
}

839
static int32_t vnodeDebugPrintSubmitMsg(SVnode *pVnode, SSubmitReq *pMsg, const char *tags) {
C
Cary Xu 已提交
840 841 842 843 844 845 846 847 848
  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 已提交
849

D
dapan 已提交
850 851
    vnodeDebugPrintSingleSubmitMsg(pMeta, pBlock, &msgIter, tags);
  }
C
Cary Xu 已提交
852 853 854 855

  return 0;
}

856
static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
857
  SSubmitReq    *pSubmitReq = (SSubmitReq *)pReq;
H
Hongze Cheng 已提交
858
  SSubmitRsp     submitRsp = {0};
H
Hongze Cheng 已提交
859
  SSubmitMsgIter msgIter = {0};
C
Cary Xu 已提交
860
  SSubmitBlk    *pBlock = NULL;
H
Hongze Cheng 已提交
861
  SVCreateTbReq  createTbReq = {0};
H
Hongze Cheng 已提交
862
  SDecoder       decoder = {0};
C
Cary Xu 已提交
863
  int32_t        nRows = 0;
H
Hongze Cheng 已提交
864 865
  int32_t        tsize, ret;
  SEncoder       encoder = {0};
866
  SArray        *newTbUids = NULL;
867
  SVStatis       statis = {0};
D
dapan1121 已提交
868
  bool           tbCreated = false;
869
  terrno = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
870 871

  pRsp->code = 0;
C
Cary Xu 已提交
872
  pSubmitReq->version = version;
873
  statis.nBatchInsert = 1;
C
Cary Xu 已提交
874

C
Cary Xu 已提交
875 876 877 878
#ifdef TD_DEBUG_PRINT_ROW
  vnodeDebugPrintSubmitMsg(pVnode, pReq, __func__);
#endif

C
Cary Xu 已提交
879 880 881 882 883
  if (tsdbScanAndConvertSubmitMsg(pVnode->pTsdb, pSubmitReq) < 0) {
    pRsp->code = terrno;
    goto _exit;
  }

H
Hongze Cheng 已提交
884
  // handle the request
H
Hongze Cheng 已提交
885 886 887
  if (tInitSubmitMsgIter(pSubmitReq, &msgIter) < 0) {
    pRsp->code = TSDB_CODE_INVALID_MSG;
    goto _exit;
H
Hongze Cheng 已提交
888 889
  }

C
Cary Xu 已提交
890 891
  submitRsp.pArray = taosArrayInit(msgIter.numOfBlocks, sizeof(SSubmitBlkRsp));
  newTbUids = taosArrayInit(msgIter.numOfBlocks, sizeof(int64_t));
C
Cary Xu 已提交
892
  if (!submitRsp.pArray || !newTbUids) {
893 894 895 896
    pRsp->code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

H
Hongze Cheng 已提交
897
  for (;;) {
H
Hongze Cheng 已提交
898 899 900
    tGetSubmitMsgNext(&msgIter, &pBlock);
    if (pBlock == NULL) break;

H
Hongze Cheng 已提交
901
    SSubmitBlkRsp submitBlkRsp = {0};
D
dapan1121 已提交
902
    tbCreated = false;
H
Hongze Cheng 已提交
903

H
Hongze Cheng 已提交
904 905
    // create table for auto create table mode
    if (msgIter.schemaLen > 0) {
H
Hongze Cheng 已提交
906 907
      tDecoderInit(&decoder, pBlock->data, msgIter.schemaLen);
      if (tDecodeSVCreateTbReq(&decoder, &createTbReq) < 0) {
H
Hongze Cheng 已提交
908
        pRsp->code = TSDB_CODE_INVALID_MSG;
H
Hongze Cheng 已提交
909
        tDecoderClear(&decoder);
wmmhello's avatar
wmmhello 已提交
910
        taosArrayDestroy(createTbReq.ctb.tagName);
H
Hongze Cheng 已提交
911 912 913
        goto _exit;
      }

C
Cary Xu 已提交
914 915 916 917 918 919 920
      if ((terrno = grantCheck(TSDB_GRANT_TIMESERIES)) < 0) {
        pRsp->code = terrno;
        tDecoderClear(&decoder);
        taosArrayDestroy(createTbReq.ctb.tagName);
        goto _exit;
      }

wafwerar's avatar
wafwerar 已提交
921 922 923 924 925 926 927
      if ((terrno = grantCheck(TSDB_GRANT_TABLE)) < 0) {
        pRsp->code = terrno;
        tDecoderClear(&decoder);
        taosArrayDestroy(createTbReq.ctb.tagName);
        goto _exit;
      }

928
      if (metaCreateTable(pVnode->pMeta, version, &createTbReq, &submitBlkRsp.pMeta) < 0) {
H
Hongze Cheng 已提交
929
        if (terrno != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
H
Hongze Cheng 已提交
930
          submitBlkRsp.code = terrno;
C
Cary Xu 已提交
931
          pRsp->code = terrno;
H
Hongze Cheng 已提交
932
          tDecoderClear(&decoder);
wmmhello's avatar
wmmhello 已提交
933
          taosArrayDestroy(createTbReq.ctb.tagName);
H
Hongze Cheng 已提交
934 935
          goto _exit;
        }
936 937 938 939
      } else {
        if (NULL != submitBlkRsp.pMeta) {
          vnodeUpdateMetaRsp(pVnode, submitBlkRsp.pMeta);
        }
940 941

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

D
dapan1121 已提交
943 944 945 946 947 948
        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;
      }
      
H
Hongze Cheng 已提交
949 950 951 952 953 954 955
      msgIter.uid = createTbReq.uid;
      if (createTbReq.type == TSDB_CHILD_TABLE) {
        msgIter.suid = createTbReq.ctb.suid;
      } else {
        msgIter.suid = 0;
      }

wmmhello's avatar
wmmhello 已提交
956
#ifdef TD_DEBUG_PRINT_ROW
D
dapan 已提交
957
      vnodeDebugPrintSingleSubmitMsg(pVnode->pMeta, pBlock, &msgIter, "real uid");
wmmhello's avatar
wmmhello 已提交
958
#endif
H
Hongze Cheng 已提交
959
      tDecoderClear(&decoder);
wmmhello's avatar
wmmhello 已提交
960
      taosArrayDestroy(createTbReq.ctb.tagName);
D
dapan1121 已提交
961
    } 
H
Hongze Cheng 已提交
962

H
Hongze Cheng 已提交
963
    if (tsdbInsertTableData(pVnode->pTsdb, version, &msgIter, pBlock, &submitBlkRsp) < 0) {
H
Hongze Cheng 已提交
964
      submitBlkRsp.code = terrno;
H
Hongze Cheng 已提交
965 966
    }

H
Hongze Cheng 已提交
967 968
    submitRsp.numOfRows += submitBlkRsp.numOfRows;
    submitRsp.affectedRows += submitBlkRsp.affectedRows;
D
dapan1121 已提交
969 970 971
    if (tbCreated || submitBlkRsp.code) {
      taosArrayPush(submitRsp.pArray, &submitBlkRsp);
    }
H
Hongze Cheng 已提交
972
  }
H
Haojun Liao 已提交
973 974

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

979
  tqUpdateTbUidList(pVnode->pTq, newTbUids, true);
980

H
Hongze Cheng 已提交
981
_exit:
982
  taosArrayDestroy(newTbUids);
H
Hongze Cheng 已提交
983 984 985 986 987 988 989
  tEncodeSize(tEncodeSSubmitRsp, &submitRsp, tsize, ret);
  pRsp->pCont = rpcMallocCont(tsize);
  pRsp->contLen = tsize;
  tEncoderInit(&encoder, pRsp->pCont, tsize);
  tEncodeSSubmitRsp(&encoder, &submitRsp);
  tEncoderClear(&encoder);

990
  taosArrayDestroyEx(submitRsp.pArray, tFreeSSubmitBlkRsp);
H
Hongze Cheng 已提交
991

992
  // TODO: the partial success scenario and the error case
H
Hongze Cheng 已提交
993 994
  // => If partial success, extract the success submitted rows and reconstruct a new submit msg, and push to level
  // 1/level 2.
995
  // TODO: refactor
C
Cary Xu 已提交
996
  if ((terrno == TSDB_CODE_SUCCESS) && (pRsp->code == TSDB_CODE_SUCCESS)) {
997
    statis.nBatchInsertSuccess = 1;
L
Liu Jicong 已提交
998
    tdProcessRSmaSubmit(pVnode->pSma, pReq, STREAM_INPUT__DATA_SUBMIT);
999
  }
C
Cary Xu 已提交
1000

1001
  // N.B. not strict as the following procedure is not atomic
1002 1003 1004
  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);
1005
  atomic_add_fetch_64(&pVnode->statis.nBatchInsertSuccess, statis.nBatchInsertSuccess);
1006

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

1011
static int32_t vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
1012
  SVCreateTSmaReq req = {0};
C
Cary Xu 已提交
1013
  SDecoder        coder = {0};
1014

C
Cary Xu 已提交
1015 1016 1017 1018 1019 1020
  if (pRsp) {
    pRsp->msgType = TDMT_VND_CREATE_SMA_RSP;
    pRsp->code = TSDB_CODE_SUCCESS;
    pRsp->pCont = NULL;
    pRsp->contLen = 0;
  }
1021 1022 1023 1024 1025

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

  if (tDecodeSVCreateTSmaReq(&coder, &req) < 0) {
C
Cary Xu 已提交
1026 1027
    terrno = TSDB_CODE_MSG_DECODE_ERROR;
    if (pRsp) pRsp->code = terrno;
1028
    goto _err;
1029
  }
C
Cary Xu 已提交
1030

C
Cary Xu 已提交
1031
  if (tdProcessTSmaCreate(pVnode->pSma, version, (const char *)&req) < 0) {
C
Cary Xu 已提交
1032
    if (pRsp) pRsp->code = terrno;
1033
    goto _err;
1034
  }
C
Cary Xu 已提交
1035

1036
  tDecoderClear(&coder);
S
Shengliang Guan 已提交
1037
  vDebug("vgId:%d, success to create tsma %s:%" PRIi64 " version %" PRIi64 " for table %" PRIi64, TD_VID(pVnode),
C
Cary Xu 已提交
1038
         req.indexName, req.indexUid, version, req.tableUid);
H
Hongze Cheng 已提交
1039
  return 0;
1040 1041 1042

_err:
  tDecoderClear(&coder);
S
Shengliang Guan 已提交
1043
  vError("vgId:%d, failed to create tsma %s:%" PRIi64 " version %" PRIi64 "for table %" PRIi64 " since %s",
C
Cary Xu 已提交
1044
         TD_VID(pVnode), req.indexName, req.indexUid, version, req.tableUid, terrstr());
1045
  return -1;
L
Liu Jicong 已提交
1046
}
C
Cary Xu 已提交
1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058

/**
 * @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);
}
1059 1060 1061 1062 1063 1064 1065 1066 1067

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;
1068
}
S
Shengliang Guan 已提交
1069

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

1073 1074
  // todo
  // 1. stop work
S
Shengliang Guan 已提交
1075 1076 1077
  // 2. adjust hash range / compact / remove wals / rename vgroups
  // 3. reload sync
  return 0;
M
Minghao Li 已提交
1078
}
H
Hongze Cheng 已提交
1079

1080
static int32_t vnodeProcessAlterConfigReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
S
Shengliang Guan 已提交
1081 1082
  bool walChanged = false;
  bool tsdbChanged = false;
1083

S
Shengliang Guan 已提交
1084 1085
  SAlterVnodeConfigReq req = {0};
  if (tDeserializeSAlterVnodeConfigReq(pReq, len, &req) != 0) {
1086 1087 1088 1089
    terrno = TSDB_CODE_INVALID_MSG;
    return TSDB_CODE_INVALID_MSG;
  }

1090
  vInfo("vgId:%d, start to alter vnode config, page:%d pageSize:%d buffer:%d szPage:%d szBuf:%" PRIu64
S
Shengliang Guan 已提交
1091
        " cacheLast:%d cacheLastSize:%d days:%d keep0:%d keep1:%d keep2:%d fsync:%d level:%d",
1092 1093
        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 已提交
1094
        req.walFsyncPeriod, req.walLevel);
1095 1096 1097

  if (pVnode->config.cacheLastSize != req.cacheLastSize) {
    pVnode->config.cacheLastSize = req.cacheLastSize;
1098 1099
    tsdbCacheSetCapacity(pVnode, (size_t)pVnode->config.cacheLastSize * 1024 * 1024);
  }
1100

1101
  if (pVnode->config.szBuf != req.buffer * 1024LL * 1024LL) {
S
Shengliang Guan 已提交
1102
    vInfo("vgId:%d, vnode buffer is changed from %" PRId64 " to %" PRId64, TD_VID(pVnode), pVnode->config.szBuf,
1103
          (uint64_t)(req.buffer * 1024LL * 1024LL));
1104
    pVnode->config.szBuf = req.buffer * 1024LL * 1024LL;
H
Hongze Cheng 已提交
1105 1106
  }

1107 1108
  if (pVnode->config.szCache != req.pages) {
    if (metaAlterCache(pVnode->pMeta, req.pages) < 0) {
S
Shengliang Guan 已提交
1109
      vError("vgId:%d, failed to change vnode pages from %d to %d failed since %s", TD_VID(pVnode),
1110
             pVnode->config.szCache, req.pages, tstrerror(errno));
H
Hongze Cheng 已提交
1111 1112
      return errno;
    } else {
S
Shengliang Guan 已提交
1113
      vInfo("vgId:%d, vnode pages is changed from %d to %d", TD_VID(pVnode), pVnode->config.szCache, req.pages);
1114
      pVnode->config.szCache = req.pages;
H
Hongze Cheng 已提交
1115
    }
H
Hongze Cheng 已提交
1116 1117
  }

1118 1119
  if (pVnode->config.cacheLast != req.cacheLast) {
    pVnode->config.cacheLast = req.cacheLast;
1120 1121
  }

1122 1123
  if (pVnode->config.walCfg.fsyncPeriod != req.walFsyncPeriod) {
    pVnode->config.walCfg.fsyncPeriod = req.walFsyncPeriod;
1124 1125 1126 1127

    walChanged = true;
  }

1128 1129
  if (pVnode->config.walCfg.level != req.walLevel) {
    pVnode->config.walCfg.level = req.walLevel;
1130 1131 1132 1133

    walChanged = true;
  }

1134 1135
  if (pVnode->config.tsdbCfg.keep0 != req.daysToKeep0) {
    pVnode->config.tsdbCfg.keep0 = req.daysToKeep0;
1136
    if (!VND_IS_RSMA(pVnode)) {
M
Minglei Jin 已提交
1137
      tsdbChanged = true;
1138 1139 1140
    }
  }

1141 1142
  if (pVnode->config.tsdbCfg.keep1 != req.daysToKeep1) {
    pVnode->config.tsdbCfg.keep1 = req.daysToKeep1;
1143
    if (!VND_IS_RSMA(pVnode)) {
M
Minglei Jin 已提交
1144
      tsdbChanged = true;
1145 1146 1147
    }
  }

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

  if (walChanged) {
M
Minglei Jin 已提交
1156 1157 1158 1159 1160
    walAlter(pVnode->pWal, &pVnode->config.walCfg);
  }

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

1163 1164 1165
  return 0;
}

1166 1167 1168 1169 1170 1171 1172 1173 1174 1175
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);

  int32_t sz = taosArrayGetSize(deleteReq.deleteReqs);
  for (int32_t i = 0; i < sz; i++) {
    SSingleDeleteReq *pOneReq = taosArrayGet(deleteReq.deleteReqs, i);
    int32_t code = tsdbDeleteTableData(pVnode->pTsdb, version, deleteReq.suid, pOneReq->uid, pOneReq->ts, pOneReq->ts);
1176 1177 1178 1179
    if (code < 0) {
      terrno = code;
      vError("vgId:%d, delete error since %s, suid:%" PRId64 ", uid:%" PRId64 ", start ts:%" PRId64 ", end ts:%" PRId64,
             TD_VID(pVnode), terrstr(), deleteReq.suid, pOneReq->uid, pOneReq->ts, pOneReq->ts);
1180 1181
    }
  }
1182
  taosArrayDestroy(deleteReq.deleteReqs);
1183 1184 1185
  return 0;
}

H
Hongze Cheng 已提交
1186 1187 1188 1189 1190
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 已提交
1191 1192 1193 1194 1195
  pRsp->msgType = TDMT_VND_DELETE_RSP;
  pRsp->pCont = NULL;
  pRsp->contLen = 0;
  pRsp->code = TSDB_CODE_SUCCESS;

H
Hongze Cheng 已提交
1196 1197 1198 1199 1200 1201 1202 1203
  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 已提交
1204
  ASSERT(taosArrayGetSize(pRes->uidList) == 0 || (pRes->skey != 0 && pRes->ekey != 0));
H
Hongze Cheng 已提交
1205 1206 1207 1208 1209 1210 1211 1212 1213

  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 已提交
1214 1215

  SVDeleteRsp rsp = {.affectedRows = pRes->affectedRows};
L
Liu Jicong 已提交
1216
  int32_t     ret = 0;
wmmhello's avatar
wmmhello 已提交
1217 1218
  tEncodeSize(tEncodeSVDeleteRsp, &rsp, pRsp->contLen, ret);
  pRsp->pCont = rpcMallocCont(pRsp->contLen);
L
Liu Jicong 已提交
1219
  SEncoder ec = {0};
wmmhello's avatar
wmmhello 已提交
1220 1221 1222
  tEncoderInit(&ec, pRsp->pCont, pRsp->contLen);
  tEncodeSVDeleteRsp(&ec, &rsp);
  tEncoderClear(&ec);
H
Hongze Cheng 已提交
1223 1224 1225 1226
  return code;

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