vnodeSvr.c 34.9 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 44 45 46 47 48
  switch (pMsg->msgType) {
    case TDMT_VND_CREATE_TABLE: {
      int64_t ctime = taosGetTimestampMs();
      int32_t nReqs;

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

      tDecodeI32v(&dc, &nReqs);
      for (int32_t iReq = 0; iReq < nReqs; iReq++) {
        tb_uid_t uid = tGenIdPI64();
H
Hongze Cheng 已提交
49
        char    *name = NULL;
H
Hongze Cheng 已提交
50 51 52
        tStartDecode(&dc);

        tDecodeI32v(&dc, NULL);
H
Hongze Cheng 已提交
53
        tDecodeCStr(&dc, &name);
H
Hongze Cheng 已提交
54 55 56
        *(int64_t *)(dc.data + dc.pos) = uid;
        *(int64_t *)(dc.data + dc.pos + 8) = ctime;

57
        vTrace("vgId:%d, table:%s uid:%" PRId64 " is generated", pVnode->config.vgId, name, uid);
H
Hongze Cheng 已提交
58 59 60 61 62 63 64 65 66 67 68
        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 已提交
69
      tb_uid_t       uid;
H
Hongze Cheng 已提交
70 71 72 73 74 75 76 77

      tInitSubmitMsgIter(pSubmitReq, &msgIter);

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

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

H
Hongze Cheng 已提交
80 81 82 83
          tDecoderInit(&dc, pBlock->data, msgIter.schemaLen);
          tStartDecode(&dc);

          tDecodeI32v(&dc, NULL);
H
Hongze Cheng 已提交
84 85 86 87 88 89
          tDecodeCStr(&dc, &name);

          uid = metaGetTableEntryUidByName(pVnode->pMeta, name);
          if (uid == 0) {
            uid = tGenIdPI64();
          }
H
Hongze Cheng 已提交
90
          *(int64_t *)(dc.data + dc.pos) = uid;
H
Hongze Cheng 已提交
91
          *(int64_t *)(dc.data + dc.pos + 8) = ctime;
H
Hongze Cheng 已提交
92
          pBlock->uid = htobe64(uid);
H
Hongze Cheng 已提交
93 94 95 96 97 98 99

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

    } break;
H
Hongze Cheng 已提交
100 101 102 103 104 105 106 107 108 109
    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 已提交
110 111 112
      if (code) {
        goto _err;
      }
H
Hongze Cheng 已提交
113 114 115 116 117

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

118 119
      ((SMsgHead *)pCont)->contLen = size + sizeof(SMsgHead);
      ((SMsgHead *)pCont)->vgId = TD_VID(pVnode);
H
Hongze Cheng 已提交
120 121 122 123 124 125 126 127 128 129 130

      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 已提交
131 132 133
    default:
      break;
  }
H
Hongze Cheng 已提交
134

S
Shengliang Guan 已提交
135
  return code;
H
Hongze Cheng 已提交
136 137 138 139

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

142
int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg *pRsp) {
143 144 145 146
  void   *ptr = NULL;
  void   *pReq;
  int32_t len;
  int32_t ret;
H
Hongze Cheng 已提交
147

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

H
Hongze Cheng 已提交
151
  pVnode->state.applied = version;
H
Hongze Cheng 已提交
152
  pVnode->state.applyTerm = pMsg->info.conn.applyTerm;
H
Hongze Cheng 已提交
153

H
Hongze Cheng 已提交
154 155 156
  // skip header
  pReq = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
  len = pMsg->contLen - sizeof(SMsgHead);
H
Hongze Cheng 已提交
157 158

  switch (pMsg->msgType) {
H
Hongze Cheng 已提交
159
    /* META */
H
Hongze Cheng 已提交
160
    case TDMT_VND_CREATE_STB:
H
Hongze Cheng 已提交
161
      if (vnodeProcessCreateStbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
162
      break;
H
Hongze Cheng 已提交
163
    case TDMT_VND_ALTER_STB:
H
Hongze Cheng 已提交
164
      if (vnodeProcessAlterStbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
165
      break;
H
Hongze Cheng 已提交
166
    case TDMT_VND_DROP_STB:
H
Hongze Cheng 已提交
167
      if (vnodeProcessDropStbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
168
      break;
H
Hongze Cheng 已提交
169
    case TDMT_VND_CREATE_TABLE:
H
Hongze Cheng 已提交
170
      if (vnodeProcessCreateTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
171 172
      break;
    case TDMT_VND_ALTER_TABLE:
H
Hongze Cheng 已提交
173
      if (vnodeProcessAlterTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
174
      break;
H
Hongze Cheng 已提交
175
    case TDMT_VND_DROP_TABLE:
H
Hongze Cheng 已提交
176
      if (vnodeProcessDropTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
177
      break;
178
    case TDMT_VND_DROP_TTL_TABLE:
179
      if (vnodeProcessDropTtlTbReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
180
      break;
S
Shengliang Guan 已提交
181 182 183 184
    case TDMT_VND_TRIM:
      if (vnodeProcessTrimReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
      break;
    case TDMT_VND_CREATE_SMA:
185
      if (vnodeProcessCreateTSmaReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
S
Shengliang Guan 已提交
186
      break;
H
Hongze Cheng 已提交
187
    /* TSDB */
H
Hongze Cheng 已提交
188
    case TDMT_VND_SUBMIT:
H
Hongze Cheng 已提交
189
      if (vnodeProcessSubmitReq(pVnode, version, pMsg->pCont, pMsg->contLen, pRsp) < 0) goto _err;
H
Hongze Cheng 已提交
190
      break;
D
dapan1121 已提交
191
    case TDMT_VND_DELETE:
H
Hongze Cheng 已提交
192
      if (vnodeProcessDeleteReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
D
dapan1121 已提交
193
      break;
194 195 196
    case TDMT_VND_BATCH_DEL:
      if (vnodeProcessBatchDeleteReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
      break;
H
Hongze Cheng 已提交
197
    /* TQ */
L
Liu Jicong 已提交
198
    case TDMT_VND_MQ_VG_CHANGE:
199
      if (tqProcessVgChangeReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
L
Liu Jicong 已提交
200
                               pMsg->contLen - sizeof(SMsgHead)) < 0) {
201
        goto _err;
L
Liu Jicong 已提交
202 203
      }
      break;
L
Liu Jicong 已提交
204
    case TDMT_VND_MQ_VG_DELETE:
205
      if (tqProcessVgDeleteReq(pVnode->pTq, version, pMsg->pCont, pMsg->contLen) < 0) {
206 207 208 209
        goto _err;
      }
      break;
    case TDMT_VND_MQ_COMMIT_OFFSET:
210 211
      if (tqProcessOffsetCommitReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
                                   pMsg->contLen - sizeof(SMsgHead)) < 0) {
212
        goto _err;
L
Liu Jicong 已提交
213 214
      }
      break;
215 216 217 218 219 220 221 222 223
    case TDMT_VND_ADD_CHECK_INFO:
      if (tqProcessAddCheckInfoReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
                                   pMsg->contLen - sizeof(SMsgHead)) < 0) {
        goto _err;
      }
      break;
    case TDMT_VND_DELETE_CHECK_INFO:
      if (tqProcessDelCheckInfoReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
                                   pMsg->contLen - sizeof(SMsgHead)) < 0) {
L
Liu Jicong 已提交
224 225 226
        goto _err;
      }
      break;
227
    case TDMT_STREAM_TASK_DEPLOY: {
228
      if (tqProcessTaskDeployReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
L
Liu Jicong 已提交
229
                                 pMsg->contLen - sizeof(SMsgHead)) < 0) {
230
        goto _err;
H
Hongze Cheng 已提交
231 232
      }
    } break;
L
Liu Jicong 已提交
233
    case TDMT_STREAM_TASK_DROP: {
234
      if (tqProcessTaskDropReq(pVnode->pTq, version, pMsg->pCont, pMsg->contLen) < 0) {
L
Liu Jicong 已提交
235 236 237
        goto _err;
      }
    } break;
238 239 240
    case TDMT_VND_ALTER_CONFIRM:
      vnodeProcessAlterConfirmReq(pVnode, version, pReq, len, pRsp);
      break;
S
Shengliang Guan 已提交
241
    case TDMT_VND_ALTER_HASHRANGE:
242
      vnodeProcessAlterHashRangeReq(pVnode, version, pReq, len, pRsp);
S
Shengliang Guan 已提交
243
      break;
244
    case TDMT_VND_ALTER_CONFIG:
245
      vnodeProcessAlterConfigReq(pVnode, version, pReq, len, pRsp);
S
Shengliang Guan 已提交
246
      break;
H
Hongze Cheng 已提交
247 248
    case TDMT_VND_COMMIT:
      goto _do_commit;
H
Hongze Cheng 已提交
249 250 251 252 253
    default:
      ASSERT(0);
      break;
  }

254
  vTrace("vgId:%d, process %s request success, index:%" PRId64, TD_VID(pVnode), TMSG_INFO(pMsg->msgType), version);
H
Hongze Cheng 已提交
255

256 257
  walApplyVer(pVnode->pWal, version);

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

H
Hongze Cheng 已提交
263
  // commit if need
H
Hongze Cheng 已提交
264
  if (vnodeShouldCommit(pVnode)) {
H
Hongze Cheng 已提交
265
  _do_commit:
S
Shengliang Guan 已提交
266
    vInfo("vgId:%d, commit at version %" PRId64, TD_VID(pVnode), version);
H
Hongze Cheng 已提交
267 268 269 270 271
    // commit current change
    vnodeCommit(pVnode);

    // start a new one
    vnodeBegin(pVnode);
H
Hongze Cheng 已提交
272 273 274
  }

  return 0;
H
Hongze Cheng 已提交
275 276

_err:
S
Shengliang Guan 已提交
277
  vError("vgId:%d, process %s request failed since %s, version:%" PRId64, TD_VID(pVnode), TMSG_INFO(pMsg->msgType),
H
Hongze Cheng 已提交
278 279
         tstrerror(terrno), version);
  return -1;
H
Hongze Cheng 已提交
280 281
}

282
int32_t vnodePreprocessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
D
dapan1121 已提交
283
  if (TDMT_SCH_QUERY != pMsg->msgType && TDMT_SCH_MERGE_QUERY != pMsg->msgType) {
D
dapan1121 已提交
284 285 286 287 288 289 290
    return 0;
  }

  return qWorkerPreprocessQueryMsg(pVnode->pQuery, pMsg);
}

int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
D
dapan 已提交
291
  vTrace("message in vnode query queue is processing");
292
  if ((pMsg->msgType == TDMT_SCH_QUERY) && !vnodeIsReadyForRead(pVnode)) {
293 294 295 296
    vnodeRedirectRpcMsg(pVnode, pMsg);
    return 0;
  }

297
  SReadHandle handle = {.meta = pVnode->pMeta, .config = &pVnode->config, .vnode = pVnode, .pMsgCb = &pVnode->msgCb};
H
Hongze Cheng 已提交
298
  switch (pMsg->msgType) {
D
dapan1121 已提交
299
    case TDMT_SCH_QUERY:
D
dapan1121 已提交
300
    case TDMT_SCH_MERGE_QUERY:
D
dapan1121 已提交
301
      return qWorkerProcessQueryMsg(&handle, pVnode->pQuery, pMsg, 0);
D
dapan1121 已提交
302
    case TDMT_SCH_QUERY_CONTINUE:
D
dapan1121 已提交
303
      return qWorkerProcessCQueryMsg(&handle, pVnode->pQuery, pMsg, 0);
H
Hongze Cheng 已提交
304 305 306 307 308 309
    default:
      vError("unknown msg type:%d in query queue", pMsg->msgType);
      return TSDB_CODE_VND_APP_ERROR;
  }
}

310
int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
S
Shengliang Guan 已提交
311
  vTrace("vgId:%d, msg:%p in fetch queue is processing", pVnode->config.vgId, pMsg);
L
Liu Jicong 已提交
312 313
  if ((pMsg->msgType == TDMT_SCH_FETCH || pMsg->msgType == TDMT_VND_TABLE_META || pMsg->msgType == TDMT_VND_TABLE_CFG ||
       pMsg->msgType == TDMT_VND_BATCH_META) &&
314
      !vnodeIsReadyForRead(pVnode)) {
315 316 317 318
    vnodeRedirectRpcMsg(pVnode, pMsg);
    return 0;
  }

H
Hongze Cheng 已提交
319 320
  char   *msgstr = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
  int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);
C
Cary Xu 已提交
321

H
Hongze Cheng 已提交
322
  switch (pMsg->msgType) {
D
dapan1121 已提交
323
    case TDMT_SCH_FETCH:
D
dapan1121 已提交
324
    case TDMT_SCH_MERGE_FETCH:
D
dapan1121 已提交
325
      return qWorkerProcessFetchMsg(pVnode, pVnode->pQuery, pMsg, 0);
D
dapan1121 已提交
326
    case TDMT_SCH_FETCH_RSP:
D
dapan1121 已提交
327
      return qWorkerProcessRspMsg(pVnode, pVnode->pQuery, pMsg, 0);
D
dapan1121 已提交
328
    case TDMT_SCH_CANCEL_TASK:
D
dapan1121 已提交
329
      return qWorkerProcessCancelMsg(pVnode, pVnode->pQuery, pMsg, 0);
D
dapan1121 已提交
330
    case TDMT_SCH_DROP_TASK:
D
dapan1121 已提交
331
      return qWorkerProcessDropMsg(pVnode, pVnode->pQuery, pMsg, 0);
D
dapan1121 已提交
332
    case TDMT_SCH_QUERY_HEARTBEAT:
D
dapan1121 已提交
333
      return qWorkerProcessHbMsg(pVnode, pVnode->pQuery, pMsg, 0);
H
Hongze Cheng 已提交
334
    case TDMT_VND_TABLE_META:
D
dapan1121 已提交
335
      return vnodeGetTableMeta(pVnode, pMsg, true);
D
dapan1121 已提交
336
    case TDMT_VND_TABLE_CFG:
D
dapan1121 已提交
337 338 339
      return vnodeGetTableCfg(pVnode, pMsg, true);
    case TDMT_VND_BATCH_META:
      return vnodeGetBatchMeta(pVnode, pMsg);
H
Hongze Cheng 已提交
340
    case TDMT_VND_CONSUME:
L
Liu Jicong 已提交
341
      return tqProcessPollReq(pVnode->pTq, pMsg);
342 343 344
    case TDMT_STREAM_TASK_RUN:
      return tqProcessTaskRunReq(pVnode->pTq, pMsg);
    case TDMT_STREAM_TASK_DISPATCH:
345
      //      return tqProcessTaskDispatchReq(pVnode->pTq, pMsg, pInfo->workerId != 0);
L
Liu Jicong 已提交
346
      return tqProcessTaskDispatchReq(pVnode->pTq, pMsg, true);
L
Liu Jicong 已提交
347 348
      /*case TDMT_STREAM_TASK_RECOVER:*/
      /*return tqProcessTaskRecoverReq(pVnode->pTq, pMsg);*/
L
Liu Jicong 已提交
349 350
    case TDMT_STREAM_RETRIEVE:
      return tqProcessTaskRetrieveReq(pVnode->pTq, pMsg);
351
    case TDMT_STREAM_TASK_DISPATCH_RSP:
L
Liu Jicong 已提交
352
      return tqProcessTaskDispatchRsp(pVnode->pTq, pMsg);
L
Liu Jicong 已提交
353 354
      /*case TDMT_STREAM_TASK_RECOVER_RSP:*/
      /*return tqProcessTaskRecoverRsp(pVnode->pTq, pMsg);*/
L
Liu Jicong 已提交
355 356
    case TDMT_STREAM_RETRIEVE_RSP:
      return tqProcessTaskRetrieveRsp(pVnode->pTq, pMsg);
H
Hongze Cheng 已提交
357 358 359 360 361 362 363 364 365
    default:
      vError("unknown msg type:%d in fetch queue", pMsg->msgType);
      return TSDB_CODE_VND_APP_ERROR;
  }
}

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

D
dapan1121 已提交
370
void vnodeUpdateMetaRsp(SVnode *pVnode, STableMetaRsp *pMetaRsp) {
371 372 373 374
  if (NULL == pMetaRsp) {
    return;
  }
  
D
dapan1121 已提交
375 376 377 378 379 380
  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 已提交
381
static int32_t vnodeProcessTrimReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
382
  int32_t     code = 0;
S
Shengliang Guan 已提交
383 384
  SVTrimDbReq trimReq = {0};

H
Hongze Cheng 已提交
385 386 387 388
  // decode
  if (tDeserializeSVTrimDbReq(pReq, len, &trimReq) != 0) {
    code = TSDB_CODE_INVALID_MSG;
    goto _exit;
S
Shengliang Guan 已提交
389 390
  }

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

H
Hongze Cheng 已提交
393 394 395 396
  // process
  code = tsdbDoRetention(pVnode->pTsdb, trimReq.timestamp);
  if (code) goto _exit;

C
Cary Xu 已提交
397 398 399
  code = smaDoRetention(pVnode->pSma, trimReq.timestamp);
  if (code) goto _exit;

H
Hongze Cheng 已提交
400 401
_exit:
  return code;
S
Shengliang Guan 已提交
402 403
}

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

S
Shengliang Guan 已提交
408 409 410 411 412 413
  SVDropTtlTableReq ttlReq = {0};
  if (tDeserializeSVDropTtlTableReq(pReq, len, &ttlReq) != 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    goto end;
  }

414
  vDebug("vgId:%d, drop ttl table req will be processed, time:%d", pVnode->config.vgId, ttlReq.timestamp);
S
Shengliang Guan 已提交
415
  int32_t ret = metaTtlDropTable(pVnode->pMeta, ttlReq.timestamp, tbUids);
L
Liu Jicong 已提交
416
  if (ret != 0) {
417 418
    goto end;
  }
L
Liu Jicong 已提交
419
  if (taosArrayGetSize(tbUids) > 0) {
wmmhello's avatar
wmmhello 已提交
420 421
    tqUpdateTbUidList(pVnode->pTq, tbUids, false);
  }
422 423 424 425 426 427

end:
  taosArrayDestroy(tbUids);
  return ret;
}

428
static int32_t vnodeProcessCreateStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
429
  SVCreateStbReq req = {0};
H
Hongze Cheng 已提交
430
  SDecoder       coder;
H
Hongze Cheng 已提交
431

H
Hongze Cheng 已提交
432 433 434 435 436 437
  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 已提交
438
  tDecoderInit(&coder, pReq, len);
H
Hongze Cheng 已提交
439 440

  if (tDecodeSVCreateStbReq(&coder, &req) < 0) {
H
Hongze Cheng 已提交
441 442
    pRsp->code = terrno;
    goto _err;
H
Hongze Cheng 已提交
443 444
  }

H
Hongze Cheng 已提交
445
  if (metaCreateSTable(pVnode->pMeta, version, &req) < 0) {
H
Hongze Cheng 已提交
446 447
    pRsp->code = terrno;
    goto _err;
H
Hongze Cheng 已提交
448 449
  }

450
  if (tdProcessRSmaCreate(pVnode->pSma, &req) < 0) {
451 452 453
    pRsp->code = terrno;
    goto _err;
  }
C
Cary Xu 已提交
454

H
Hongze Cheng 已提交
455
  tDecoderClear(&coder);
H
Hongze Cheng 已提交
456
  return 0;
H
Hongze Cheng 已提交
457 458

_err:
H
Hongze Cheng 已提交
459
  tDecoderClear(&coder);
H
Hongze Cheng 已提交
460
  return -1;
H
Hongze Cheng 已提交
461 462
}

463
static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
464
  SDecoder           decoder = {0};
465
  SEncoder           encoder = {0};
466
  int32_t            rcode = 0;
H
Hongze Cheng 已提交
467 468
  SVCreateTbBatchReq req = {0};
  SVCreateTbReq     *pCreateReq;
H
Hongze Cheng 已提交
469 470 471
  SVCreateTbBatchRsp rsp = {0};
  SVCreateTbRsp      cRsp = {0};
  char               tbName[TSDB_TABLE_FNAME_LEN];
C
Cary Xu 已提交
472
  STbUidStore       *pStore = NULL;
473
  SArray            *tbUids = NULL;
H
Hongze Cheng 已提交
474 475

  pRsp->msgType = TDMT_VND_CREATE_TABLE_RSP;
H
Hongze Cheng 已提交
476 477 478
  pRsp->code = TSDB_CODE_SUCCESS;
  pRsp->pCont = NULL;
  pRsp->contLen = 0;
H
Hongze Cheng 已提交
479

H
Hongze Cheng 已提交
480
  // decode
H
Hongze Cheng 已提交
481 482
  tDecoderInit(&decoder, pReq, len);
  if (tDecodeSVCreateTbBatchReq(&decoder, &req) < 0) {
H
Hongze Cheng 已提交
483 484 485 486
    rcode = -1;
    terrno = TSDB_CODE_INVALID_MSG;
    goto _exit;
  }
H
Hongze Cheng 已提交
487

H
Hongze Cheng 已提交
488
  rsp.pArray = taosArrayInit(req.nReqs, sizeof(cRsp));
489 490
  tbUids = taosArrayInit(req.nReqs, sizeof(int64_t));
  if (rsp.pArray == NULL || tbUids == NULL) {
H
Hongze Cheng 已提交
491 492 493 494 495
    rcode = -1;
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

H
Hongze Cheng 已提交
496
  // loop to create table
497
  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
H
Hongze Cheng 已提交
498
    pCreateReq = req.pReqs + iReq;
H
Hongze Cheng 已提交
499

C
Cary Xu 已提交
500 501 502 503
    if ((terrno = grantCheck(TSDB_GRANT_TIMESERIES)) < 0) {
      rcode = -1;
      goto _exit;
    }
L
Liu Jicong 已提交
504

wafwerar's avatar
wafwerar 已提交
505 506 507 508 509
    if ((terrno = grantCheck(TSDB_GRANT_TABLE)) < 0) {
      rcode = -1;
      goto _exit;
    }

H
Hongze Cheng 已提交
510 511 512 513 514 515 516 517 518
    // 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
519
    if (metaCreateTable(pVnode->pMeta, version, pCreateReq, &cRsp.pMeta) < 0) {
H
Hongze Cheng 已提交
520 521 522 523 524
      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 已提交
525
    } else {
H
Hongze Cheng 已提交
526
      cRsp.code = TSDB_CODE_SUCCESS;
527
      tdFetchTbUidList(pVnode->pSma, &pStore, pCreateReq->ctb.suid, pCreateReq->uid);
528
      taosArrayPush(tbUids, &pCreateReq->uid);
529
      vnodeUpdateMetaRsp(pVnode, cRsp.pMeta); 
H
Hongze Cheng 已提交
530
    }
H
Hongze Cheng 已提交
531 532

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

535
  tqUpdateTbUidList(pVnode->pTq, tbUids, true);
C
Cary Xu 已提交
536 537 538
  if (tdUpdateTbUidList(pVnode->pSma, pStore) < 0) {
    goto _exit;
  }
539
  tdUidStoreFree(pStore);
C
Cary Xu 已提交
540

H
Hongze Cheng 已提交
541
  // prepare rsp
542
  int32_t ret = 0;
wafwerar's avatar
wafwerar 已提交
543
  tEncodeSize(tEncodeSVCreateTbBatchRsp, &rsp, pRsp->contLen, ret);
H
Hongze Cheng 已提交
544 545 546 547 548 549
  pRsp->pCont = rpcMallocCont(pRsp->contLen);
  if (pRsp->pCont == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    rcode = -1;
    goto _exit;
  }
H
Hongze Cheng 已提交
550 551
  tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
  tEncodeSVCreateTbBatchRsp(&encoder, &rsp);
H
Hongze Cheng 已提交
552

H
Hongze Cheng 已提交
553
_exit:
wmmhello's avatar
wmmhello 已提交
554 555 556 557
  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
    pCreateReq = req.pReqs + iReq;
    taosArrayDestroy(pCreateReq->ctb.tagName);
  }
558
  taosArrayDestroyEx(rsp.pArray, tFreeSVCreateTbRsp);
559
  taosArrayDestroy(tbUids);
H
Hongze Cheng 已提交
560 561
  tDecoderClear(&decoder);
  tEncoderClear(&encoder);
H
Hongze Cheng 已提交
562
  return rcode;
H
Hongze Cheng 已提交
563 564
}

565
static int32_t vnodeProcessAlterStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
566 567 568 569 570 571 572 573 574 575 576 577 578 579 580
  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 已提交
581
  }
H
Hongze Cheng 已提交
582 583 584 585 586 587 588 589 590

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

  tDecoderClear(&dc);

H
Hongze Cheng 已提交
591 592 593
  return 0;
}

594
static int32_t vnodeProcessDropStbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
595
  SVDropStbReq req = {0};
596
  int32_t      rcode = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
597
  SDecoder     decoder = {0};
598
  SArray      *tbUidList = NULL;
H
Hongze Cheng 已提交
599 600 601 602 603 604

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

  // decode request
H
Hongze Cheng 已提交
605 606
  tDecoderInit(&decoder, pReq, len);
  if (tDecodeSVDropStbReq(&decoder, &req) < 0) {
H
Hongze Cheng 已提交
607 608 609 610 611
    rcode = TSDB_CODE_INVALID_MSG;
    goto _exit;
  }

  // process request
612 613 614 615 616 617 618 619
  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) {
620 621 622
    rcode = terrno;
    goto _exit;
  }
H
Hongze Cheng 已提交
623

624 625 626 627 628
  if (tdProcessRSmaDrop(pVnode->pSma, &req) < 0) {
    rcode = terrno;
    goto _exit;
  }

H
Hongze Cheng 已提交
629 630
  // return rsp
_exit:
631
  if (tbUidList) taosArrayDestroy(tbUidList);
H
Hongze Cheng 已提交
632
  pRsp->code = rcode;
H
Hongze Cheng 已提交
633
  tDecoderClear(&decoder);
H
Hongze Cheng 已提交
634 635 636
  return 0;
}

637
static int32_t vnodeProcessAlterTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
D
dapan1121 已提交
638 639 640
  SVAlterTbReq  vAlterTbReq = {0};
  SVAlterTbRsp  vAlterTbRsp = {0};
  SDecoder      dc = {0};
641 642
  int32_t       rcode = 0;
  int32_t       ret;
D
dapan1121 已提交
643 644
  SEncoder      ec = {0};
  STableMetaRsp vMetaRsp = {0};
H
Hongze Cheng 已提交
645 646 647 648 649 650 651 652 653 654

  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 已提交
655
    vAlterTbRsp.code = TSDB_CODE_INVALID_MSG;
H
Hongze Cheng 已提交
656
    tDecoderClear(&dc);
H
Hongze Cheng 已提交
657 658
    rcode = -1;
    goto _exit;
H
Hongze Cheng 已提交
659 660 661
  }

  // process
D
dapan1121 已提交
662
  if (metaAlterTable(pVnode->pMeta, version, &vAlterTbReq, &vMetaRsp) < 0) {
wmmhello's avatar
wmmhello 已提交
663
    vAlterTbRsp.code = terrno;
H
Hongze Cheng 已提交
664
    tDecoderClear(&dc);
H
Hongze Cheng 已提交
665 666
    rcode = -1;
    goto _exit;
H
Hongze Cheng 已提交
667 668
  }
  tDecoderClear(&dc);
H
Hongze Cheng 已提交
669

D
dapan1121 已提交
670 671 672 673 674
  if (NULL != vMetaRsp.pSchemas) {
    vnodeUpdateMetaRsp(pVnode, &vMetaRsp);
    vAlterTbRsp.pMeta = &vMetaRsp;
  }

H
Hongze Cheng 已提交
675 676 677 678 679 680
_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 已提交
681 682 683
  if (vMetaRsp.pSchemas) {
    taosMemoryFree(vMetaRsp.pSchemas);
  }
H
Hongze Cheng 已提交
684 685 686
  return 0;
}

687
static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
688 689
  SVDropTbBatchReq req = {0};
  SVDropTbBatchRsp rsp = {0};
H
Hongze Cheng 已提交
690
  SDecoder         decoder = {0};
H
Hongze Cheng 已提交
691
  SEncoder         encoder = {0};
692
  int32_t          ret;
693
  SArray          *tbUids = NULL;
H
Hongze Cheng 已提交
694

H
Hongze Cheng 已提交
695
  pRsp->msgType = TDMT_VND_DROP_TABLE_RSP;
H
Hongze Cheng 已提交
696 697 698
  pRsp->pCont = NULL;
  pRsp->contLen = 0;
  pRsp->code = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
699 700

  // decode req
H
Hongze Cheng 已提交
701 702
  tDecoderInit(&decoder, pReq, len);
  ret = tDecodeSVDropTbBatchReq(&decoder, &req);
H
Hongze Cheng 已提交
703 704 705 706 707
  if (ret < 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    pRsp->code = terrno;
    goto _exit;
  }
H
Hongze Cheng 已提交
708 709

  // process req
710
  tbUids = taosArrayInit(req.nReqs, sizeof(int64_t));
H
Hongze Cheng 已提交
711
  rsp.pArray = taosArrayInit(req.nReqs, sizeof(SVDropTbRsp));
712 713
  if (tbUids == NULL || rsp.pArray == NULL) goto _exit;

714
  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
H
Hongze Cheng 已提交
715 716
    SVDropTbReq *pDropTbReq = req.pReqs + iReq;
    SVDropTbRsp  dropTbRsp = {0};
H
Hongze Cheng 已提交
717

H
Hongze Cheng 已提交
718
    /* code */
719
    ret = metaDropTable(pVnode->pMeta, version, pDropTbReq, tbUids);
H
Hongze Cheng 已提交
720
    if (ret < 0) {
H
Hongze Cheng 已提交
721 722 723 724 725
      if (pDropTbReq->igNotExists && terrno == TSDB_CODE_VND_TABLE_NOT_EXIST) {
        dropTbRsp.code = TSDB_CODE_SUCCESS;
      } else {
        dropTbRsp.code = terrno;
      }
H
Hongze Cheng 已提交
726
    } else {
H
Hongze Cheng 已提交
727
      dropTbRsp.code = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
728 729 730 731 732
    }

    taosArrayPush(rsp.pArray, &dropTbRsp);
  }

733 734
  tqUpdateTbUidList(pVnode->pTq, tbUids, false);

H
Hongze Cheng 已提交
735
_exit:
736
  taosArrayDestroy(tbUids);
H
Hongze Cheng 已提交
737
  tDecoderClear(&decoder);
H
Hongze Cheng 已提交
738 739 740 741 742
  tEncodeSize(tEncodeSVDropTbBatchRsp, &rsp, pRsp->contLen, ret);
  pRsp->pCont = rpcMallocCont(pRsp->contLen);
  tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
  tEncodeSVDropTbBatchRsp(&encoder, &rsp);
  tEncoderClear(&encoder);
743
  taosArrayDestroy(rsp.pArray);
H
Hongze Cheng 已提交
744 745 746
  return 0;
}

747 748
static int32_t vnodeDebugPrintSingleSubmitMsg(SMeta *pMeta, SSubmitBlk *pBlock, SSubmitMsgIter *msgIter,
                                              const char *tags) {
D
dapan 已提交
749 750 751 752
  SSubmitBlkIter blkIter = {0};
  STSchema      *pSchema = NULL;
  tb_uid_t       suid = 0;
  STSRow        *row = NULL;
C
Cary Xu 已提交
753
  int32_t        rv = -1;
D
dapan 已提交
754 755 756

  tInitSubmitBlkIter(msgIter, pBlock, &blkIter);
  if (blkIter.row == NULL) return 0;
C
Cary Xu 已提交
757
  if (!pSchema || (suid != msgIter->suid) || rv != TD_ROW_SVER(blkIter.row)) {
D
dapan 已提交
758 759 760
    if (pSchema) {
      taosMemoryFreeClear(pSchema);
    }
C
Cary Xu 已提交
761
    pSchema = metaGetTbTSchema(pMeta, msgIter->suid, TD_ROW_SVER(blkIter.row));  // TODO: use the real schema
D
dapan 已提交
762 763
    if (pSchema) {
      suid = msgIter->suid;
C
Cary Xu 已提交
764
      rv = TD_ROW_SVER(blkIter.row);
D
dapan 已提交
765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781
    }
  }
  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;
}

782
static int32_t vnodeDebugPrintSubmitMsg(SVnode *pVnode, SSubmitReq *pMsg, const char *tags) {
C
Cary Xu 已提交
783 784 785 786 787 788 789 790 791
  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 已提交
792

D
dapan 已提交
793 794
    vnodeDebugPrintSingleSubmitMsg(pMeta, pBlock, &msgIter, tags);
  }
C
Cary Xu 已提交
795 796 797 798

  return 0;
}

799
static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
H
Hongze Cheng 已提交
800
  SSubmitReq    *pSubmitReq = (SSubmitReq *)pReq;
H
Hongze Cheng 已提交
801
  SSubmitRsp     submitRsp = {0};
H
Hongze Cheng 已提交
802 803 804 805
  SSubmitMsgIter msgIter = {0};
  SSubmitBlk    *pBlock;
  SSubmitRsp     rsp = {0};
  SVCreateTbReq  createTbReq = {0};
H
Hongze Cheng 已提交
806
  SDecoder       decoder = {0};
H
Hongze Cheng 已提交
807
  int32_t        nRows;
H
Hongze Cheng 已提交
808 809
  int32_t        tsize, ret;
  SEncoder       encoder = {0};
810
  SArray        *newTbUids = NULL;
811
  terrno = TSDB_CODE_SUCCESS;
H
Hongze Cheng 已提交
812 813

  pRsp->code = 0;
C
Cary Xu 已提交
814
  pSubmitReq->version = version;
C
Cary Xu 已提交
815

C
Cary Xu 已提交
816 817 818 819
#ifdef TD_DEBUG_PRINT_ROW
  vnodeDebugPrintSubmitMsg(pVnode, pReq, __func__);
#endif

C
Cary Xu 已提交
820 821 822 823 824
  if (tsdbScanAndConvertSubmitMsg(pVnode->pTsdb, pSubmitReq) < 0) {
    pRsp->code = terrno;
    goto _exit;
  }

H
Hongze Cheng 已提交
825
  // handle the request
H
Hongze Cheng 已提交
826 827 828
  if (tInitSubmitMsgIter(pSubmitReq, &msgIter) < 0) {
    pRsp->code = TSDB_CODE_INVALID_MSG;
    goto _exit;
H
Hongze Cheng 已提交
829 830
  }

C
Cary Xu 已提交
831 832
  submitRsp.pArray = taosArrayInit(msgIter.numOfBlocks, sizeof(SSubmitBlkRsp));
  newTbUids = taosArrayInit(msgIter.numOfBlocks, sizeof(int64_t));
C
Cary Xu 已提交
833
  if (!submitRsp.pArray || !newTbUids) {
834 835 836 837
    pRsp->code = TSDB_CODE_OUT_OF_MEMORY;
    goto _exit;
  }

H
Hongze Cheng 已提交
838
  for (;;) {
H
Hongze Cheng 已提交
839 840 841
    tGetSubmitMsgNext(&msgIter, &pBlock);
    if (pBlock == NULL) break;

H
Hongze Cheng 已提交
842 843
    SSubmitBlkRsp submitBlkRsp = {0};

H
Hongze Cheng 已提交
844 845
    // create table for auto create table mode
    if (msgIter.schemaLen > 0) {
H
Hongze Cheng 已提交
846 847
      submitBlkRsp.hashMeta = 1;

H
Hongze Cheng 已提交
848 849
      tDecoderInit(&decoder, pBlock->data, msgIter.schemaLen);
      if (tDecodeSVCreateTbReq(&decoder, &createTbReq) < 0) {
H
Hongze Cheng 已提交
850
        pRsp->code = TSDB_CODE_INVALID_MSG;
H
Hongze Cheng 已提交
851
        tDecoderClear(&decoder);
wmmhello's avatar
wmmhello 已提交
852
        taosArrayDestroy(createTbReq.ctb.tagName);
H
Hongze Cheng 已提交
853 854 855
        goto _exit;
      }

C
Cary Xu 已提交
856 857 858 859 860 861 862
      if ((terrno = grantCheck(TSDB_GRANT_TIMESERIES)) < 0) {
        pRsp->code = terrno;
        tDecoderClear(&decoder);
        taosArrayDestroy(createTbReq.ctb.tagName);
        goto _exit;
      }

wafwerar's avatar
wafwerar 已提交
863 864 865 866 867 868 869
      if ((terrno = grantCheck(TSDB_GRANT_TABLE)) < 0) {
        pRsp->code = terrno;
        tDecoderClear(&decoder);
        taosArrayDestroy(createTbReq.ctb.tagName);
        goto _exit;
      }

870
      if (metaCreateTable(pVnode->pMeta, version, &createTbReq, &submitBlkRsp.pMeta) < 0) {
H
Hongze Cheng 已提交
871
        if (terrno != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
H
Hongze Cheng 已提交
872
          submitBlkRsp.code = terrno;
C
Cary Xu 已提交
873
          pRsp->code = terrno;
H
Hongze Cheng 已提交
874
          tDecoderClear(&decoder);
wmmhello's avatar
wmmhello 已提交
875
          taosArrayDestroy(createTbReq.ctb.tagName);
H
Hongze Cheng 已提交
876 877
          goto _exit;
        }
878 879 880 881
      } else {
        if (NULL != submitBlkRsp.pMeta) {
          vnodeUpdateMetaRsp(pVnode, submitBlkRsp.pMeta);
        }
H
Hongze Cheng 已提交
882
      }
883
      taosArrayPush(newTbUids, &createTbReq.uid);
H
Hongze Cheng 已提交
884

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

H
Hongze Cheng 已提交
889 890 891 892 893 894 895
      msgIter.uid = createTbReq.uid;
      if (createTbReq.type == TSDB_CHILD_TABLE) {
        msgIter.suid = createTbReq.ctb.suid;
      } else {
        msgIter.suid = 0;
      }

wmmhello's avatar
wmmhello 已提交
896
#ifdef TD_DEBUG_PRINT_ROW
D
dapan 已提交
897
      vnodeDebugPrintSingleSubmitMsg(pVnode->pMeta, pBlock, &msgIter, "real uid");
wmmhello's avatar
wmmhello 已提交
898
#endif
H
Hongze Cheng 已提交
899
      tDecoderClear(&decoder);
wmmhello's avatar
wmmhello 已提交
900
      taosArrayDestroy(createTbReq.ctb.tagName);
D
dapan1121 已提交
901 902 903
    } else {
      submitBlkRsp.tblFName = taosMemoryMalloc(TSDB_TABLE_FNAME_LEN);
      sprintf(submitBlkRsp.tblFName, "%s.", pVnode->config.dbname);
H
Hongze Cheng 已提交
904 905
    }

H
Hongze Cheng 已提交
906
    if (tsdbInsertTableData(pVnode->pTsdb, version, &msgIter, pBlock, &submitBlkRsp) < 0) {
H
Hongze Cheng 已提交
907
      submitBlkRsp.code = terrno;
H
Hongze Cheng 已提交
908 909
    }

H
Hongze Cheng 已提交
910 911 912
    submitRsp.numOfRows += submitBlkRsp.numOfRows;
    submitRsp.affectedRows += submitBlkRsp.affectedRows;
    taosArrayPush(submitRsp.pArray, &submitBlkRsp);
H
Hongze Cheng 已提交
913
  }
914
  tqUpdateTbUidList(pVnode->pTq, newTbUids, true);
915

H
Hongze Cheng 已提交
916
_exit:
917
  taosArrayDestroy(newTbUids);
H
Hongze Cheng 已提交
918 919 920 921 922 923 924
  tEncodeSize(tEncodeSSubmitRsp, &submitRsp, tsize, ret);
  pRsp->pCont = rpcMallocCont(tsize);
  pRsp->contLen = tsize;
  tEncoderInit(&encoder, pRsp->pCont, tsize);
  tEncodeSSubmitRsp(&encoder, &submitRsp);
  tEncoderClear(&encoder);

925
  taosArrayDestroyEx(submitRsp.pArray, tFreeSSubmitBlkRsp);
H
Hongze Cheng 已提交
926

927
  // TODO: the partial success scenario and the error case
H
Hongze Cheng 已提交
928 929
  // => If partial success, extract the success submitted rows and reconstruct a new submit msg, and push to level
  // 1/level 2.
930
  // TODO: refactor
C
Cary Xu 已提交
931
  if ((terrno == TSDB_CODE_SUCCESS) && (pRsp->code == TSDB_CODE_SUCCESS)) {
L
Liu Jicong 已提交
932
    tdProcessRSmaSubmit(pVnode->pSma, pReq, STREAM_INPUT__DATA_SUBMIT);
933
  }
C
Cary Xu 已提交
934

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

939
static int32_t vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
940
  SVCreateTSmaReq req = {0};
C
Cary Xu 已提交
941
  SDecoder        coder = {0};
942

C
Cary Xu 已提交
943 944 945 946 947 948
  if (pRsp) {
    pRsp->msgType = TDMT_VND_CREATE_SMA_RSP;
    pRsp->code = TSDB_CODE_SUCCESS;
    pRsp->pCont = NULL;
    pRsp->contLen = 0;
  }
949 950 951 952 953

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

  if (tDecodeSVCreateTSmaReq(&coder, &req) < 0) {
C
Cary Xu 已提交
954 955
    terrno = TSDB_CODE_MSG_DECODE_ERROR;
    if (pRsp) pRsp->code = terrno;
956
    goto _err;
957
  }
C
Cary Xu 已提交
958

C
Cary Xu 已提交
959
  if (tdProcessTSmaCreate(pVnode->pSma, version, (const char *)&req) < 0) {
C
Cary Xu 已提交
960
    if (pRsp) pRsp->code = terrno;
961
    goto _err;
962
  }
C
Cary Xu 已提交
963

964
  tDecoderClear(&coder);
S
Shengliang Guan 已提交
965
  vDebug("vgId:%d, success to create tsma %s:%" PRIi64 " version %" PRIi64 " for table %" PRIi64, TD_VID(pVnode),
C
Cary Xu 已提交
966
         req.indexName, req.indexUid, version, req.tableUid);
H
Hongze Cheng 已提交
967
  return 0;
968 969 970

_err:
  tDecoderClear(&coder);
S
Shengliang Guan 已提交
971
  vError("vgId:%d, failed to create tsma %s:%" PRIi64 " version %" PRIi64 "for table %" PRIi64 " since %s",
C
Cary Xu 已提交
972
         TD_VID(pVnode), req.indexName, req.indexUid, version, req.tableUid, terrstr());
973
  return -1;
L
Liu Jicong 已提交
974
}
C
Cary Xu 已提交
975 976 977 978 979 980 981 982 983 984 985 986

/**
 * @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);
}
987 988 989 990 991 992 993 994 995

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

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

1001 1002
  // todo
  // 1. stop work
S
Shengliang Guan 已提交
1003 1004 1005
  // 2. adjust hash range / compact / remove wals / rename vgroups
  // 3. reload sync
  return 0;
M
Minghao Li 已提交
1006
}
H
Hongze Cheng 已提交
1007

1008 1009
static int32_t vnodeProcessAlterConfigReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
  SAlterVnodeReq alterReq = {0};
1010
  bool           walChanged = false;
M
Minglei Jin 已提交
1011
  bool           tsdbChanged = false;
1012

1013 1014 1015 1016 1017 1018 1019
  if (tDeserializeSAlterVnodeReq(pReq, len, &alterReq) != 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    return TSDB_CODE_INVALID_MSG;
  }

  vInfo("vgId:%d, start to alter vnode config, cacheLast:%d cacheLastSize:%d", TD_VID(pVnode), alterReq.cacheLast,
        alterReq.cacheLastSize);
1020 1021 1022 1023
  if (pVnode->config.cacheLastSize != alterReq.cacheLastSize) {
    pVnode->config.cacheLastSize = alterReq.cacheLastSize;
    tsdbCacheSetCapacity(pVnode, (size_t)pVnode->config.cacheLastSize * 1024 * 1024);
  }
1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041

  if (pVnode->config.cacheLast != alterReq.cacheLast) {
    pVnode->config.cacheLast = alterReq.cacheLast;
  }

  if (pVnode->config.walCfg.fsyncPeriod != alterReq.walFsyncPeriod) {
    pVnode->config.walCfg.fsyncPeriod = alterReq.walFsyncPeriod;

    walChanged = true;
  }

  if (pVnode->config.walCfg.level != alterReq.walLevel) {
    pVnode->config.walCfg.level = alterReq.walLevel;

    walChanged = true;
  }

  if (pVnode->config.tsdbCfg.keep0 != alterReq.daysToKeep0) {
M
Minglei Jin 已提交
1042
    pVnode->config.tsdbCfg.keep0 = alterReq.daysToKeep0;
1043
    if (!VND_IS_RSMA(pVnode)) {
M
Minglei Jin 已提交
1044
      tsdbChanged = true;
1045 1046 1047 1048
    }
  }

  if (pVnode->config.tsdbCfg.keep1 != alterReq.daysToKeep1) {
M
Minglei Jin 已提交
1049
    pVnode->config.tsdbCfg.keep1 = alterReq.daysToKeep1;
1050
    if (!VND_IS_RSMA(pVnode)) {
M
Minglei Jin 已提交
1051
      tsdbChanged = true;
1052 1053 1054 1055
    }
  }

  if (pVnode->config.tsdbCfg.keep2 != alterReq.daysToKeep2) {
M
Minglei Jin 已提交
1056
    pVnode->config.tsdbCfg.keep2 = alterReq.daysToKeep2;
1057
    if (!VND_IS_RSMA(pVnode)) {
M
Minglei Jin 已提交
1058
      tsdbChanged = true;
1059 1060 1061 1062
    }
  }

  if (walChanged) {
M
Minglei Jin 已提交
1063 1064 1065 1066 1067
    walAlter(pVnode->pWal, &pVnode->config.walCfg);
  }

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

1070 1071 1072
  return 0;
}

1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086
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);
    if (code) {
      // TODO
    }
  }
1087
  taosArrayDestroy(deleteReq.deleteReqs);
1088 1089 1090
  return 0;
}

H
Hongze Cheng 已提交
1091 1092 1093 1094 1095
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 已提交
1096 1097 1098 1099 1100
  pRsp->msgType = TDMT_VND_DELETE_RSP;
  pRsp->pCont = NULL;
  pRsp->contLen = 0;
  pRsp->code = TSDB_CODE_SUCCESS;

H
Hongze Cheng 已提交
1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117
  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);

  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 已提交
1118 1119

  SVDeleteRsp rsp = {.affectedRows = pRes->affectedRows};
L
Liu Jicong 已提交
1120
  int32_t     ret = 0;
wmmhello's avatar
wmmhello 已提交
1121 1122
  tEncodeSize(tEncodeSVDeleteRsp, &rsp, pRsp->contLen, ret);
  pRsp->pCont = rpcMallocCont(pRsp->contLen);
L
Liu Jicong 已提交
1123
  SEncoder ec = {0};
wmmhello's avatar
wmmhello 已提交
1124 1125 1126
  tEncoderInit(&ec, pRsp->pCont, pRsp->contLen);
  tEncodeSVDeleteRsp(&ec, &rsp);
  tEncoderClear(&ec);
H
Hongze Cheng 已提交
1127 1128 1129 1130
  return code;

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