taosdata/TDengine | Build 3621 | source/dnode/vnode/src/vnd/vnodeSvr.c | Coveralls

UNCOV

static int32_t vnodePreprocessCreateTableReq(SVnode *pVnode, SDecoder *pCoder, int64_t btime, int64_t *pUid) {

×

UNCOV

64

  int32_t code = 0;

×

UNCOV

65

  int32_t lino = 0;

×

UNCOV

67

  if (tStartDecode(pCoder) < 0) {

×

68

    code = TSDB_CODE_INVALID_MSG;

×

69

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

73

  if (tDecodeI32v(pCoder, NULL) < 0) {

×

74

    code = TSDB_CODE_INVALID_MSG;

×

75

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

79

  char *name = NULL;

×

UNCOV

80

  if (tDecodeCStr(pCoder, &name) < 0) {

×

81

    code = TSDB_CODE_INVALID_MSG;

×

82

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

86

  int64_t uid = metaGetTableEntryUidByName(pVnode->pMeta, name);

×

UNCOV

87

  if (uid == 0) {

×

UNCOV

88

    uid = tGenIdPI64();

×

UNCOV

90

  *(int64_t *)(pCoder->data + pCoder->pos) = uid;

×

UNCOV

93

  *(int64_t *)(pCoder->data + pCoder->pos + 8) = btime;

×

UNCOV

95

  tEndDecode(pCoder);

×

UNCOV

97

_exit:

×

UNCOV

98

  if (code) {

×

99

    vError("vgId:%d %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));

×

UNCOV

101

    vTrace("vgId:%d %s done, table:%s uid generated:%" PRId64, TD_VID(pVnode), __func__, name, uid);

×

UNCOV

102

    if (pUid) *pUid = uid;

×

UNCOV

104

  return code;

×

UNCOV

106

static int32_t vnodePreProcessCreateTableMsg(SVnode *pVnode, SRpcMsg *pMsg) {

×

UNCOV

107

  int32_t code = 0;

×

UNCOV

108

  int32_t lino = 0;

×

UNCOV

110

  int64_t  btime = taosGetTimestampMs();

×

UNCOV

111

  SDecoder dc = {0};

×

UNCOV

114

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

×

UNCOV

115

  if (tStartDecode(&dc) < 0) {

×

116

    code = TSDB_CODE_INVALID_MSG;

×

117

    return code;

×

UNCOV

120

  if (tDecodeI32v(&dc, &nReqs) < 0) {

×

121

    code = TSDB_CODE_INVALID_MSG;

×

122

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

124

  for (int32_t iReq = 0; iReq < nReqs; iReq++) {

×

UNCOV

125

    code = vnodePreprocessCreateTableReq(pVnode, &dc, btime, NULL);

×

UNCOV

126

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

129

  tEndDecode(&dc);

×

UNCOV

131

_exit:

×

UNCOV

132

  tDecoderClear(&dc);

×

UNCOV

133

  if (code) {

×

134

    vError("vgId:%d, %s:%d failed to preprocess submit request since %s, msg type:%s", TD_VID(pVnode), __func__, lino,

×

UNCOV

137

  return code;

×

UNCOV

140

static int32_t vnodePreProcessAlterTableMsg(SVnode *pVnode, SRpcMsg *pMsg) {

×

UNCOV

141

  int32_t code = TSDB_CODE_INVALID_MSG;

×

UNCOV

142

  int32_t lino = 0;

×

UNCOV

144

  SDecoder dc = {0};

×

UNCOV

145

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

×

UNCOV

147

  SVAlterTbReq vAlterTbReq = {0};

×

UNCOV

148

  int64_t      ctimeMs = taosGetTimestampMs();

×

UNCOV

149

  if (tDecodeSVAlterTbReqSetCtime(&dc, &vAlterTbReq, ctimeMs) < 0) {

×

150

    taosArrayDestroy(vAlterTbReq.pMultiTag);

×

151

    vAlterTbReq.pMultiTag = NULL;

×

152

    goto _exit;

×

UNCOV

154

  taosArrayDestroy(vAlterTbReq.pMultiTag);

×

UNCOV

155

  vAlterTbReq.pMultiTag = NULL;

×

UNCOV

157

  code = 0;

×

UNCOV

159

_exit:

×

UNCOV

160

  tDecoderClear(&dc);

×

UNCOV

161

  if (code) {

×

162

    vError("vgId:%d %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));

×

UNCOV

164

    vTrace("vgId:%d %s done, table:%s ctimeMs generated:%" PRId64, TD_VID(pVnode), __func__, vAlterTbReq.tbName,

×

UNCOV

167

  return code;

×

UNCOV

170

static int32_t vnodePreProcessDropTtlMsg(SVnode *pVnode, SRpcMsg *pMsg) {

×

UNCOV

171

  int32_t code = TSDB_CODE_INVALID_MSG;

×

UNCOV

172

  int32_t lino = 0;

×

UNCOV

174

  SMsgHead *pContOld = pMsg->pCont;

×

UNCOV

175

  int32_t   reqLenOld = pMsg->contLen - sizeof(SMsgHead);

×

UNCOV

177

  SArray *tbUids = NULL;

×

UNCOV

178

  int64_t timestampMs = 0;

×

UNCOV

180

  SVDropTtlTableReq ttlReq = {0};

×

UNCOV

181

  if (tDeserializeSVDropTtlTableReq((char *)pContOld + sizeof(SMsgHead), reqLenOld, &ttlReq) != 0) {

×

182

    code = TSDB_CODE_INVALID_MSG;

×

183

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

187

    tbUids = taosArrayInit(8, sizeof(tb_uid_t));

×

UNCOV

188

    if (tbUids == NULL) {

×

189

      code = terrno;

×

190

      TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

193

    timestampMs = (int64_t)ttlReq.timestampSec * 1000;

×

UNCOV

194

    code = metaTtlFindExpired(pVnode->pMeta, timestampMs, tbUids, ttlReq.ttlDropMaxCount);

×

UNCOV

195

    if (code != 0) {

×

196

      code = TSDB_CODE_INVALID_MSG;

×

197

      TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

200

    ttlReq.nUids = taosArrayGetSize(tbUids);

×

UNCOV

201

    ttlReq.pTbUids = tbUids;

×

UNCOV

204

  if (ttlReq.nUids == 0) {

×

UNCOV

205

    code = TSDB_CODE_MSG_PREPROCESSED;

×

UNCOV

206

    TSDB_CHECK_CODE(code, lino, _exit);

×

210

    int32_t reqLenNew = tSerializeSVDropTtlTableReq(NULL, 0, &ttlReq);

×

UNCOV

211

    int32_t contLenNew = reqLenNew + sizeof(SMsgHead);

×

UNCOV

213

    SMsgHead *pContNew = rpcMallocCont(contLenNew);

×

UNCOV

214

    if (pContNew == NULL) {

×

215

      code = terrno;

×

216

      TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

219

    if (tSerializeSVDropTtlTableReq((char *)pContNew + sizeof(SMsgHead), reqLenNew, &ttlReq) != 0) {

×

UNCOV

220

      vError("vgId:%d %s:%d failed to serialize drop ttl request", TD_VID(pVnode), __func__, lino);

×

UNCOV

222

    pContNew->contLen = htonl(reqLenNew);

×

UNCOV

223

    pContNew->vgId = pContOld->vgId;

×

UNCOV

225

    rpcFreeCont(pContOld);

×

UNCOV

226

    pMsg->pCont = pContNew;

×

UNCOV

227

    pMsg->contLen = contLenNew;

×

UNCOV

230

  code = 0;

×

UNCOV

232

_exit:

×

UNCOV

233

  taosArrayDestroy(tbUids);

×

UNCOV

235

  if (code && code != TSDB_CODE_MSG_PREPROCESSED) {

×

236

    vError("vgId:%d, %s:%d failed to preprocess drop ttl request since %s, msg type:%s", TD_VID(pVnode), __func__, lino,

×

UNCOV

239

    vTrace("vgId:%d, %s done, timestampSec:%d, nUids:%d", TD_VID(pVnode), __func__, ttlReq.timestampSec, ttlReq.nUids);

×

UNCOV

242

  return code;

×

UNCOV

246

static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int64_t btimeMs, int64_t ctimeMs) {

×

UNCOV

247

  int32_t code = 0;

×

UNCOV

248

  int32_t lino = 0;

×

UNCOV

250

  if (tStartDecode(pCoder) < 0) {

×

251

    code = TSDB_CODE_INVALID_MSG;

×

252

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

257

  if (tDecodeI32v(pCoder, &submitTbData.flags) < 0) {

×

258

    code = TSDB_CODE_INVALID_MSG;

×

259

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

261

  version = (submitTbData.flags >> 8) & 0xff;

×

UNCOV

262

  submitTbData.flags = submitTbData.flags & 0xff;

×

UNCOV

265

  if (submitTbData.flags & SUBMIT_REQ_AUTO_CREATE_TABLE) {

×

UNCOV

266

    code = vnodePreprocessCreateTableReq(pVnode, pCoder, btimeMs, &uid);

×

UNCOV

267

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

271

  if (tDecodeI64(pCoder, &submitTbData.suid) < 0) {

×

272

    code = TSDB_CODE_INVALID_MSG;

×

273

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

276

  if (submitTbData.flags & SUBMIT_REQ_AUTO_CREATE_TABLE) {

×

UNCOV

277

    *(int64_t *)(pCoder->data + pCoder->pos) = uid;

×

UNCOV

278

    pCoder->pos += sizeof(int64_t);

×

UNCOV

280

    if (tDecodeI64(pCoder, &submitTbData.uid) < 0) {

×

281

      code = TSDB_CODE_INVALID_MSG;

×

282

      TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

286

  if (tDecodeI32v(pCoder, &submitTbData.sver) < 0) {

×

287

    code = TSDB_CODE_INVALID_MSG;

×

288

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

292

  TSKEY now = btimeMs;

×

UNCOV

293

  if (pVnode->config.tsdbCfg.precision == TSDB_TIME_PRECISION_MICRO) {

×

UNCOV

294

    now *= 1000;

×

UNCOV

295

  } else if (pVnode->config.tsdbCfg.precision == TSDB_TIME_PRECISION_NANO) {

×

UNCOV

296

    now *= 1000000;

×

UNCOV

299

  int32_t keep = pVnode->config.tsdbCfg.keep2;

×

UNCOV

311

  TSKEY minKey = now - tsTickPerMin[pVnode->config.tsdbCfg.precision] * keep;

×

UNCOV

312

  TSKEY maxKey = tsMaxKeyByPrecision[pVnode->config.tsdbCfg.precision];

×

UNCOV

313

  if (submitTbData.flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {

×

UNCOV

315

    if (tDecodeU64v(pCoder, &nColData) < 0) {

×

316

      code = TSDB_CODE_INVALID_MSG;

×

UNCOV

317

      TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

320

    SColData colData = {0};

×

UNCOV

321

    code = tDecodeColData(version, pCoder, &colData);

×

UNCOV

322

    if (code) {

×

323

      code = TSDB_CODE_INVALID_MSG;

×

324

      TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

327

    if (colData.flag != HAS_VALUE) {

×

328

      code = TSDB_CODE_INVALID_MSG;

×

329

      TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

332

    for (int32_t iRow = 0; iRow < colData.nVal; iRow++) {

×

UNCOV

333

      if (((TSKEY *)colData.pData)[iRow] < minKey || ((TSKEY *)colData.pData)[iRow] > maxKey) {

×

UNCOV

334

        code = TSDB_CODE_TDB_TIMESTAMP_OUT_OF_RANGE;

×

UNCOV

335

        TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

339

    for (uint64_t i = 1; i < nColData; i++) {

×

UNCOV

340

      code = tDecodeColData(version, pCoder, &colData);

×

UNCOV

341

      if (code) {

×

342

        code = TSDB_CODE_INVALID_MSG;

×

343

        TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

348

    if (tDecodeU64v(pCoder, &nRow) < 0) {

×

349

      code = TSDB_CODE_INVALID_MSG;

×

UNCOV

350

      TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

353

    for (int32_t iRow = 0; iRow < nRow; ++iRow) {

×

UNCOV

354

      SRow *pRow = (SRow *)(pCoder->data + pCoder->pos);

×

UNCOV

355

      pCoder->pos += pRow->len;

×

UNCOV

357

      if (pRow->ts < minKey || pRow->ts > maxKey) {

×

UNCOV

358

        code = TSDB_CODE_TDB_TIMESTAMP_OUT_OF_RANGE;

×

UNCOV

359

        TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

364

  if (!tDecodeIsEnd(pCoder)) {

×

UNCOV

365

    *(int64_t *)(pCoder->data + pCoder->pos) = ctimeMs;

×

UNCOV

366

    pCoder->pos += sizeof(int64_t);

×

UNCOV

369

  tEndDecode(pCoder);

×

UNCOV

371

_exit:

×

UNCOV

372

  if (code) {

×

UNCOV

373

    vError("vgId:%d, %s:%d failed to vnodePreProcessSubmitTbData submit request since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));

×

UNCOV

375

  return code;

×

UNCOV

377

static int32_t vnodePreProcessSubmitMsg(SVnode *pVnode, SRpcMsg *pMsg) {

×

UNCOV

378

  int32_t code = 0;

×

UNCOV

379

  int32_t lino = 0;

×

UNCOV

381

  if (tsBypassFlag & TSDB_BYPASS_RA_RPC_RECV_SUBMIT) {

×

382

    return TSDB_CODE_MSG_PREPROCESSED;

×

UNCOV

385

  SDecoder *pCoder = &(SDecoder){0};

×

UNCOV

387

  if (taosHton64(((SSubmitReq2Msg *)pMsg->pCont)->version) != 1) {

×

388

    code = TSDB_CODE_INVALID_MSG;

×

389

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

392

  tDecoderInit(pCoder, (uint8_t *)pMsg->pCont + sizeof(SSubmitReq2Msg), pMsg->contLen - sizeof(SSubmitReq2Msg));

×

UNCOV

394

  if (tStartDecode(pCoder) < 0) {

×

395

    code = TSDB_CODE_INVALID_MSG;

×

396

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

400

  if (tDecodeU64v(pCoder, &nSubmitTbData) < 0) {

×

401

    code = TSDB_CODE_INVALID_MSG;

×

402

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

405

  int64_t btimeMs = taosGetTimestampMs();

×

UNCOV

406

  int64_t ctimeMs = btimeMs;

×

UNCOV

407

  for (int32_t i = 0; i < nSubmitTbData; i++) {

×

UNCOV

408

    code = vnodePreProcessSubmitTbData(pVnode, pCoder, btimeMs, ctimeMs);

×

UNCOV

409

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

412

  tEndDecode(pCoder);

×

UNCOV

414

_exit:

×

UNCOV

415

  tDecoderClear(pCoder);

×

UNCOV

416

  if (code) {

×

UNCOV

417

    vError("vgId:%d, %s:%d failed to preprocess submit request since %s, msg type:%s", TD_VID(pVnode), __func__, lino,

×

UNCOV

420

  return code;

×

UNCOV

423

static int32_t vnodePreProcessDeleteMsg(SVnode *pVnode, SRpcMsg *pMsg) {

×

UNCOV

424

  int32_t code = 0;

×

UNCOV

429

  SEncoder  *pCoder = &(SEncoder){0};

×

UNCOV

430

  SDeleteRes res = {0};

×

UNCOV

432

  SReadHandle handle = {.vnode = pVnode, .pMsgCb = &pVnode->msgCb, .skipRollup = 1};

×

UNCOV

433

  initStorageAPI(&handle.api);

×

UNCOV

435

  code = qWorkerProcessDeleteMsg(&handle, pVnode->pQuery, pMsg, &res);

×

UNCOV

436

  if (code) goto _exit;

×

UNCOV

438

  res.ctimeMs = taosGetTimestampMs();

×

UNCOV

440

  tEncodeSize(tEncodeDeleteRes, &res, size, ret);

×

UNCOV

441

  pCont = rpcMallocCont(size + sizeof(SMsgHead));

×

UNCOV

443

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

×

UNCOV

444

  ((SMsgHead *)pCont)->vgId = TD_VID(pVnode);

×

UNCOV

446

  tEncoderInit(pCoder, pCont + sizeof(SMsgHead), size);

×

UNCOV

447

  if (tEncodeDeleteRes(pCoder, &res) != 0) {

×

448

    vError("vgId:%d %s failed to encode delete response", TD_VID(pVnode), __func__);

×

UNCOV

450

  tEncoderClear(pCoder);

×

UNCOV

452

  rpcFreeCont(pMsg->pCont);

×

UNCOV

453

  pMsg->pCont = pCont;

×

UNCOV

454

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

×

UNCOV

456

  taosArrayDestroy(res.uidList);

×

UNCOV

458

_exit:

×

UNCOV

459

  return code;

×

UNCOV

462

static int32_t vnodePreProcessBatchDeleteMsg(SVnode *pVnode, SRpcMsg *pMsg) {

×

UNCOV

463

  int32_t code = 0;

×

UNCOV

464

  int32_t lino = 0;

×

UNCOV

466

  int64_t         ctimeMs = taosGetTimestampMs();

×

UNCOV

467

  SBatchDeleteReq pReq = {0};

×

UNCOV

468

  SDecoder       *pCoder = &(SDecoder){0};

×

UNCOV

470

  tDecoderInit(pCoder, (uint8_t *)pMsg->pCont + sizeof(SMsgHead), pMsg->contLen - sizeof(SMsgHead));

×

UNCOV

472

  if (tDecodeSBatchDeleteReqSetCtime(pCoder, &pReq, ctimeMs) < 0) {

×

473

    code = TSDB_CODE_INVALID_MSG;

×

UNCOV

476

  tDecoderClear(pCoder);

×

UNCOV

477

  taosArrayDestroy(pReq.deleteReqs);

×

UNCOV

479

  if (code) {

×

480

    vError("vgId:%d %s failed at line %d since %s", TD_VID(pVnode), __func__, lino, tstrerror(code));

×

UNCOV

482

    vTrace("vgId:%d %s done, ctimeMs generated:%" PRId64, TD_VID(pVnode), __func__, ctimeMs);

×

UNCOV

484

  return code;

×

UNCOV

487

static int32_t vnodePreProcessArbCheckSyncMsg(SVnode *pVnode, SRpcMsg *pMsg) {

×

UNCOV

488

  SVArbCheckSyncReq syncReq = {0};

×

UNCOV

490

  if (tDeserializeSVArbCheckSyncReq((char *)pMsg->pCont + sizeof(SMsgHead), pMsg->contLen - sizeof(SMsgHead),

×

492

    return TSDB_CODE_INVALID_MSG;

×

UNCOV

495

  int32_t ret = vnodeCheckToken(pVnode, syncReq.member0Token, syncReq.member1Token);

×

UNCOV

496

  if (ret != 0) {

×

497

    vError("vgId:%d, failed to preprocess arb check sync request since %s", TD_VID(pVnode), tstrerror(ret));

×

UNCOV

500

  int32_t code = terrno;

×

UNCOV

501

  tFreeSVArbCheckSyncReq(&syncReq);

×

UNCOV

503

  return code;

×

UNCOV

506

int32_t vnodePreProcessDropTbMsg(SVnode *pVnode, SRpcMsg *pMsg) {

×

UNCOV

507

  int32_t          code = TSDB_CODE_SUCCESS;

×

UNCOV

508

  int32_t          lino = 0;

×

UNCOV

509

  int32_t          size = 0;

×

UNCOV

510

  SDecoder         dc = {0};

×

UNCOV

511

  SEncoder         ec = {0};

×

UNCOV

512

  SVDropTbBatchReq receivedBatchReqs = {0};

×

UNCOV

513

  SVDropTbBatchReq sentBatchReqs = {0};

×

UNCOV

515

  tDecoderInit(&dc, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), pMsg->contLen - sizeof(SMsgHead));

×

UNCOV

517

  code = tDecodeSVDropTbBatchReq(&dc, &receivedBatchReqs);

×

UNCOV

518

  if (code < 0) {

×

519

    terrno = code;

×

520

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

522

  sentBatchReqs.pArray = taosArrayInit(receivedBatchReqs.nReqs, sizeof(SVDropTbReq));

×

UNCOV

523

  if (!sentBatchReqs.pArray) {

×

524

    code = terrno;

×

525

    goto _exit;

×

UNCOV

528

  for (int32_t i = 0; i < receivedBatchReqs.nReqs; ++i) {

×

UNCOV

529

    SVDropTbReq *pReq = receivedBatchReqs.pReqs + i;

×

UNCOV

530

    tb_uid_t     uid = metaGetTableEntryUidByName(pVnode->pMeta, pReq->name);

×

UNCOV

531

    if (uid == 0) {

×

532

      vWarn("vgId:%d, preprocess drop ctb: %s not found", TD_VID(pVnode), pReq->name);

×

533

      continue;

×

UNCOV

535

    pReq->uid = uid;

×

UNCOV

536

    vDebug("vgId:%d %s for: %s, uid: %" PRId64, TD_VID(pVnode), __func__, pReq->name, pReq->uid);

×

UNCOV

537

    if (taosArrayPush(sentBatchReqs.pArray, pReq) == NULL) {

×

538

      code = terrno;

×

539

      goto _exit;

×

UNCOV

542

  sentBatchReqs.nReqs = sentBatchReqs.pArray->size;

×

UNCOV

544

  tEncodeSize(tEncodeSVDropTbBatchReq, &sentBatchReqs, size, code);

×

UNCOV

545

  tEncoderInit(&ec, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), size);

×

UNCOV

546

  code = tEncodeSVDropTbBatchReq(&ec, &sentBatchReqs);

×

UNCOV

547

  tEncoderClear(&ec);

×

UNCOV

548

  if (code != TSDB_CODE_SUCCESS) {

×

549

    vError("vgId:%d %s failed to encode drop tb batch req: %s", TD_VID(pVnode), __func__, tstrerror(code));

×

550

    TSDB_CHECK_CODE(code, lino, _exit);

×

UNCOV

553

_exit:

×

UNCOV

554

  tDecoderClear(&dc);

×

UNCOV

555

  if (sentBatchReqs.pArray) {

×

UNCOV

556

    taosArrayDestroy(sentBatchReqs.pArray);

×

UNCOV

558

  return code;

×

UNCOV

561

int32_t vnodePreProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg) {

×

UNCOV

562

  int32_t code = 0;

×

UNCOV

564

  switch (pMsg->msgType) {

×

UNCOV

565

    case TDMT_VND_CREATE_TABLE: {

×

UNCOV

566

      code = vnodePreProcessCreateTableMsg(pVnode, pMsg);

×

UNCOV

567

    } break;

×

UNCOV

568

    case TDMT_VND_ALTER_TABLE: {

×

UNCOV

569

      code = vnodePreProcessAlterTableMsg(pVnode, pMsg);

×

UNCOV

570

    } break;

×

UNCOV

571

    case TDMT_VND_FETCH_TTL_EXPIRED_TBS:

×

UNCOV

573

      code = vnodePreProcessDropTtlMsg(pVnode, pMsg);

×

UNCOV

574

    } break;

×

UNCOV

575

    case TDMT_VND_SUBMIT: {

×

UNCOV

576

      code = vnodePreProcessSubmitMsg(pVnode, pMsg);

×

UNCOV

577

    } break;

×

UNCOV

578

    case TDMT_VND_DELETE: {

×

UNCOV

579

      code = vnodePreProcessDeleteMsg(pVnode, pMsg);

×

UNCOV

580

    } break;

×

UNCOV

581

    case TDMT_VND_BATCH_DEL: {

×

UNCOV

582

      code = vnodePreProcessBatchDeleteMsg(pVnode, pMsg);

×

UNCOV

583

    } break;

×

UNCOV

584

    case TDMT_VND_ARB_CHECK_SYNC: {

×

UNCOV

585

      code = vnodePreProcessArbCheckSyncMsg(pVnode, pMsg);

×

UNCOV

586

    } break;

×

UNCOV

587

    case TDMT_VND_DROP_TABLE: {

×

UNCOV

588

      code = vnodePreProcessDropTbMsg(pVnode, pMsg);

×

UNCOV

589

    } break;

×

UNCOV

590

    default:

×

UNCOV

591

      break;

×

UNCOV

594

  if (code && code != TSDB_CODE_MSG_PREPROCESSED) {

×

UNCOV

595

    vError("vgId:%d, failed to preprocess write request since %s, msg type:%s", TD_VID(pVnode), tstrerror(code),

×

UNCOV

598

  return code;

×

UNCOV

601

int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t ver, SRpcMsg *pRsp) {

×

UNCOV

602

  int32_t code = 0;

×

UNCOV

603

  void   *ptr = NULL;

×

UNCOV

607

  (void)taosThreadMutexLock(&pVnode->mutex);

×

UNCOV

608

  if (pVnode->disableWrite) {

×

609

    (void)taosThreadMutexUnlock(&pVnode->mutex);

×

610

    vError("vgId:%d write is disabled for snapshot, version:%" PRId64, TD_VID(pVnode), ver);

×

611

    return TSDB_CODE_VND_WRITE_DISABLED;

×

UNCOV

613

  (void)taosThreadMutexUnlock(&pVnode->mutex);

×

UNCOV

615

  if (ver <= pVnode->state.applied) {

×

616

    vError("vgId:%d, duplicate write request. ver: %" PRId64 ", applied: %" PRId64 "", TD_VID(pVnode), ver,

×

618

    return terrno = TSDB_CODE_VND_DUP_REQUEST;

×

UNCOV

621

  vDebug("vgId:%d, start to process write request %s, index:%" PRId64 ", applied:%" PRId64 ", state.applyTerm:%" PRId64

×

UNCOV

626

  if (!(pVnode->state.applyTerm <= pMsg->info.conn.applyTerm)) {

×

627

    return terrno = TSDB_CODE_INTERNAL_ERROR;

×

UNCOV

630

  if (!(pVnode->state.applied + 1 == ver)) {

×

631

    return terrno = TSDB_CODE_INTERNAL_ERROR;

×

UNCOV

634

  atomic_store_64(&pVnode->state.applied, ver);

×

UNCOV

635

  atomic_store_64(&pVnode->state.applyTerm, pMsg->info.conn.applyTerm);

×

UNCOV

637

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

×

UNCOV

640

  pReq = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));

×

UNCOV

641

  len = pMsg->contLen - sizeof(SMsgHead);

×

UNCOV

642

  bool needCommit = false;

×

UNCOV

644

  switch (pMsg->msgType) {

×

UNCOV

646

    case TDMT_VND_CREATE_STB:

×

UNCOV

647

      if (vnodeProcessCreateStbReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;

×

UNCOV

648

      break;

×

UNCOV

649

    case TDMT_VND_ALTER_STB:

×

UNCOV

650

      if (vnodeProcessAlterStbReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;

×

UNCOV

651

      break;

×

UNCOV

652

    case TDMT_VND_DROP_STB:

×

UNCOV

653

      if (vnodeProcessDropStbReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;

×

UNCOV

654

      break;

×

UNCOV

655

    case TDMT_VND_CREATE_TABLE:

×

UNCOV

656

      if (vnodeProcessCreateTbReq(pVnode, ver, pReq, len, pRsp, pMsg) < 0) goto _err;

×

UNCOV

657

      break;

×

UNCOV

658

    case TDMT_VND_ALTER_TABLE:

×

UNCOV

659

      if (vnodeProcessAlterTbReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;

×

UNCOV

660

      break;

×

UNCOV

661

    case TDMT_VND_DROP_TABLE:

×

UNCOV

662

      if (vnodeProcessDropTbReq(pVnode, ver, pReq, len, pRsp, pMsg) < 0) goto _err;

×

UNCOV

663

      break;

×

664

    case TDMT_VND_DROP_TTL_TABLE:

×

665

      if (vnodeProcessDropTtlTbReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;

×

666

      break;

×

UNCOV

667

    case TDMT_VND_FETCH_TTL_EXPIRED_TBS:

×

UNCOV

668

      if (vnodeProcessFetchTtlExpiredTbs(pVnode, ver, pReq, len, pRsp) < 0) goto _err;

×

UNCOV

669

      break;

×

UNCOV

670

    case TDMT_VND_TRIM:

×

UNCOV

671

      if (vnodeProcessTrimReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;

×

UNCOV

672

      break;

×

673

    case TDMT_VND_S3MIGRATE:

×

674

      if (vnodeProcessS3MigrateReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;

×

675

      break;

×

UNCOV

676

    case TDMT_VND_CREATE_SMA:

×

UNCOV

677

      if (vnodeProcessCreateTSmaReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;

×

UNCOV

678

      break;

×

UNCOV

680

    case TDMT_VND_SUBMIT:

×

UNCOV

681

      if (vnodeProcessSubmitReq(pVnode, ver, pMsg->pCont, pMsg->contLen, pRsp, pMsg) < 0) goto _err;

×

UNCOV

682

      break;

×

UNCOV

683

    case TDMT_VND_DELETE:

×

UNCOV

684

      if (vnodeProcessDeleteReq(pVnode, ver, pReq, len, pRsp, pMsg) < 0) goto _err;

×

UNCOV

685

      break;

×

UNCOV

686

    case TDMT_VND_BATCH_DEL:

×

UNCOV

687

      if (vnodeProcessBatchDeleteReq(pVnode, ver, pReq, len, pRsp) < 0) goto _err;

×

UNCOV

688

      break;

×

UNCOV

690

    case TDMT_VND_TMQ_SUBSCRIBE:

×

UNCOV

691

      if (tqProcessSubscribeReq(pVnode->pTq, ver, pReq, len) < 0) {

×

UNCOV

692

        goto _err;

×

UNCOV

694

      break;

×

UNCOV

695

    case TDMT_VND_TMQ_DELETE_SUB:

×

UNCOV

696

      if (tqProcessDeleteSubReq(pVnode->pTq, ver, pMsg->pCont, pMsg->contLen) < 0) {

×

697

        goto _err;

×

UNCOV

699

      break;

×

UNCOV

700

    case TDMT_VND_TMQ_COMMIT_OFFSET:

×

UNCOV

701

      if (tqProcessOffsetCommitReq(pVnode->pTq, ver, pReq, len) < 0) {

×

702

        goto _err;

×

UNCOV

704

      break;

×

UNCOV

705

    case TDMT_VND_TMQ_ADD_CHECKINFO:

×

UNCOV

706

      if (tqProcessAddCheckInfoReq(pVnode->pTq, ver, pReq, len) < 0) {

×

707

        goto _err;

×

UNCOV

709

      break;

×

UNCOV

710

    case TDMT_VND_TMQ_DEL_CHECKINFO:

×

UNCOV

711

      if (tqProcessDelCheckInfoReq(pVnode->pTq, ver, pReq, len) < 0) {

×

UNCOV

712

        goto _err;

×

UNCOV

714

      break;

×

UNCOV

715

    case TDMT_STREAM_TASK_DEPLOY: {

×

UNCOV

716

      if ((code = tqProcessTaskDeployReq(pVnode->pTq, ver, pReq, len)) != TSDB_CODE_SUCCESS) {

×

717

        goto _err;

×

UNCOV

719

    } break;

×

UNCOV

720

    case TDMT_STREAM_TASK_DROP: {

×

UNCOV

721

      if ((code = tqProcessTaskDropReq(pVnode->pTq, pMsg->pCont, pMsg->contLen)) < 0) {

×

722

        goto _err;

×

UNCOV

724

    } break;

×

UNCOV

725

    case TDMT_STREAM_TASK_UPDATE_CHKPT: {

×

UNCOV

726

      if ((code = tqProcessTaskUpdateCheckpointReq(pVnode->pTq, pMsg->pCont, pMsg->contLen)) < 0) {

×

727

        goto _err;

×

UNCOV

729

    } break;

×

UNCOV

730

    case TDMT_STREAM_CONSEN_CHKPT: {

×

UNCOV

731

      if (pVnode->restored && (code = tqProcessTaskConsenChkptIdReq(pVnode->pTq, pMsg)) < 0) {

×

732

        goto _err;

×

UNCOV

735

    } break;

×

UNCOV

736

    case TDMT_STREAM_TASK_PAUSE: {

×

UNCOV

737

      if (pVnode->restored && vnodeIsLeader(pVnode) &&

×

UNCOV

738

          (code = tqProcessTaskPauseReq(pVnode->pTq, ver, pMsg->pCont, pMsg->contLen)) < 0) {

×

739

        goto _err;

×

UNCOV

741

    } break;

×

UNCOV

742

    case TDMT_STREAM_TASK_RESUME: {

×

UNCOV

743

      if (pVnode->restored && vnodeIsLeader(pVnode) &&

×

UNCOV

744

          (code = tqProcessTaskResumeReq(pVnode->pTq, ver, pMsg->pCont, pMsg->contLen)) < 0) {

×

745

        goto _err;

×

UNCOV

747

    } break;

×

UNCOV

748

    case TDMT_VND_STREAM_TASK_RESET: {

×

UNCOV

749

      if (pVnode->restored && vnodeIsLeader(pVnode) && (code = tqProcessTaskResetReq(pVnode->pTq, pMsg)) < 0) {

×

750

        goto _err;

×

UNCOV

753

    } break;

×

UNCOV

754

    case TDMT_VND_ALTER_CONFIRM:

×

UNCOV

755

      needCommit = pVnode->config.hashChange;

×

UNCOV

756

      if (vnodeProcessAlterConfirmReq(pVnode, ver, pReq, len, pRsp) < 0) {

×

757

        goto _err;

×

UNCOV

759

      break;

×

UNCOV

760

    case TDMT_VND_ALTER_CONFIG:

×

UNCOV

761

      vnodeProcessAlterConfigReq(pVnode, ver, pReq, len, pRsp);

×

UNCOV

762

      break;

×

UNCOV

763

    case TDMT_VND_COMMIT:

×

UNCOV

764

      needCommit = true;

×

UNCOV

765

      break;

×

UNCOV

766

    case TDMT_VND_CREATE_INDEX:

×

UNCOV

767

      vnodeProcessCreateIndexReq(pVnode, ver, pReq, len, pRsp);

×

UNCOV

768

      break;

×

UNCOV

769

    case TDMT_VND_DROP_INDEX:

×

UNCOV

770

      vnodeProcessDropIndexReq(pVnode, ver, pReq, len, pRsp);

×

UNCOV

771

      break;

×

UNCOV

772

    case TDMT_VND_STREAM_CHECK_POINT_SOURCE:  // always return true

×

UNCOV

773

      tqProcessTaskCheckPointSourceReq(pVnode->pTq, pMsg, pRsp);

×

UNCOV

774

      break;

×

UNCOV

775

    case TDMT_VND_STREAM_TASK_UPDATE:  // always return true

×

UNCOV

776

      tqProcessTaskUpdateReq(pVnode->pTq, pMsg);

×

UNCOV

777

      break;

×

UNCOV

778

    case TDMT_VND_COMPACT:

×

UNCOV

779

      vnodeProcessCompactVnodeReq(pVnode, ver, pReq, len, pRsp);

×

UNCOV

780

      goto _exit;

×

781

    case TDMT_SYNC_CONFIG_CHANGE:

×

782

      vnodeProcessConfigChangeReq(pVnode, ver, pReq, len, pRsp);

×

783

      break;

×

785

    case TDMT_VND_KILL_COMPACT:

×

786

      vnodeProcessKillCompactReq(pVnode, ver, pReq, len, pRsp);

×

787

      break;

×

UNCOV

790

    case TDMT_VND_ARB_CHECK_SYNC:

×

UNCOV

791

      vnodeProcessArbCheckSyncReq(pVnode, pReq, len, pRsp);

×

UNCOV

792

      break;

×

793

    default:

×

794

      vError("vgId:%d, unprocessed msg, %d", TD_VID(pVnode), pMsg->msgType);

×

795

      return TSDB_CODE_INVALID_MSG;

×

UNCOV

798

  vTrace("vgId:%d, process %s request, code:0x%x index:%" PRId64, TD_VID(pVnode), TMSG_INFO(pMsg->msgType), pRsp->code,

×

UNCOV

801

  walApplyVer(pVnode->pWal, ver);

×

UNCOV

803

  code = tqPushMsg(pVnode->pTq, pMsg->msgType);

×

UNCOV

804

  if (code) {

×

UNCOV

805

    vError("vgId:%d, failed to push msg to TQ since %s", TD_VID(pVnode), tstrerror(terrno));

×

UNCOV

806

    return code;

×

UNCOV

810

  if (needCommit) {

×

UNCOV

811

    vInfo("vgId:%d, commit at version %" PRId64, TD_VID(pVnode), ver);

×

UNCOV

812

    code = vnodeAsyncCommit(pVnode);

×

UNCOV

813

    if (code) {

×

814

      vError("vgId:%d, failed to vnode async commit since %s.", TD_VID(pVnode), tstrerror(terrno));

×

815

      goto _err;

×

UNCOV

819

    code = vnodeBegin(pVnode);

×

UNCOV

820

    if (code) {

×

UNCOV

821

      vError("vgId:%d, failed to begin vnode since %s.", TD_VID(pVnode), tstrerror(terrno));

×

822

      goto _err;

×

UNCOV

826

_exit:

×

UNCOV

827

  return 0;

×

UNCOV

829

_err:

×

UNCOV

830

  vError("vgId:%d, process %s request failed since %s, ver:%" PRId64, TD_VID(pVnode), TMSG_INFO(pMsg->msgType),

×

UNCOV

832

  return code;

×

UNCOV

835

int32_t vnodePreprocessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {

×

UNCOV

836

  if (TDMT_SCH_QUERY != pMsg->msgType && TDMT_SCH_MERGE_QUERY != pMsg->msgType) {

×

UNCOV

837

    return 0;

×

UNCOV

840

  return qWorkerPreprocessQueryMsg(pVnode->pQuery, pMsg, TDMT_SCH_QUERY == pMsg->msgType);

×

UNCOV

843

int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {

×

UNCOV

844

  vTrace("message in vnode query queue is processing");

×

UNCOV

845

  if (pMsg->msgType == TDMT_VND_TMQ_CONSUME && !syncIsReadyForRead(pVnode->sync)) {

×

UNCOV

846

    vnodeRedirectRpcMsg(pVnode, pMsg, terrno);

×

UNCOV

847

    return 0;

×

UNCOV

850

  if (pMsg->msgType == TDMT_VND_TMQ_CONSUME && !pVnode->restored) {

×

851

    vnodeRedirectRpcMsg(pVnode, pMsg, TSDB_CODE_SYN_RESTORING);

×

852

    return 0;

×

UNCOV

855

  SReadHandle handle = {.vnode = pVnode, .pMsgCb = &pVnode->msgCb, .pWorkerCb = pInfo->workerCb};

×

UNCOV

856

  initStorageAPI(&handle.api);

×

UNCOV

857

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

858

  bool    redirected = false;

×

UNCOV

860

  switch (pMsg->msgType) {

×

UNCOV

861

    case TDMT_SCH_QUERY:

×

UNCOV

862

      if (!syncIsReadyForRead(pVnode->sync)) {

×

UNCOV

863

        pMsg->code = (terrno) ? terrno : TSDB_CODE_SYN_NOT_LEADER;

×

UNCOV

864

        redirected = true;

×

UNCOV

866

      code = qWorkerProcessQueryMsg(&handle, pVnode->pQuery, pMsg, 0);

×

UNCOV

867

      if (redirected) {

×

UNCOV

868

        vnodeRedirectRpcMsg(pVnode, pMsg, pMsg->code);

×

UNCOV

869

        return 0;

×

UNCOV

871

      return code;

×

UNCOV

872

    case TDMT_SCH_MERGE_QUERY:

×

UNCOV

873

      return qWorkerProcessQueryMsg(&handle, pVnode->pQuery, pMsg, 0);

×

UNCOV

874

    case TDMT_SCH_QUERY_CONTINUE:

×

UNCOV

875

      return qWorkerProcessCQueryMsg(&handle, pVnode->pQuery, pMsg, 0);

×

UNCOV

876

    case TDMT_VND_TMQ_CONSUME:

×

UNCOV

877

      return tqProcessPollReq(pVnode->pTq, pMsg);

×

UNCOV

878

    case TDMT_VND_TMQ_CONSUME_PUSH:

×

UNCOV

879

      return tqProcessPollPush(pVnode->pTq, pMsg);

×

UNCOV

880

    default:

×

UNCOV

881

      vError("unknown msg type:%d in query queue", pMsg->msgType);

×

882

      return TSDB_CODE_APP_ERROR;

×

UNCOV

886

int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {

×

UNCOV

887

  vTrace("vgId:%d, msg:%p in fetch queue is processing", pVnode->config.vgId, pMsg);

×

UNCOV

888

  if ((pMsg->msgType == TDMT_SCH_FETCH || pMsg->msgType == TDMT_VND_TABLE_META || pMsg->msgType == TDMT_VND_TABLE_CFG ||

×

UNCOV

889

       pMsg->msgType == TDMT_VND_BATCH_META || pMsg->msgType == TDMT_VND_TABLE_NAME) &&

×

UNCOV

890

      !syncIsReadyForRead(pVnode->sync)) {

×

UNCOV

891

    vnodeRedirectRpcMsg(pVnode, pMsg, terrno);

×

UNCOV

892

    return 0;

×

UNCOV

895

  switch (pMsg->msgType) {

×

UNCOV

896

    case TDMT_SCH_FETCH:

×

UNCOV

898

      return qWorkerProcessFetchMsg(pVnode, pVnode->pQuery, pMsg, 0);

×

899

    case TDMT_SCH_FETCH_RSP:

×

900

      return qWorkerProcessRspMsg(pVnode, pVnode->pQuery, pMsg, 0);

×

UNCOV

903

    case TDMT_SCH_DROP_TASK:

×

UNCOV

904

      return qWorkerProcessDropMsg(pVnode, pVnode->pQuery, pMsg, 0);

×

UNCOV

905

    case TDMT_SCH_TASK_NOTIFY:

×

UNCOV

906

      return qWorkerProcessNotifyMsg(pVnode, pVnode->pQuery, pMsg, 0);

×

UNCOV

907

    case TDMT_SCH_QUERY_HEARTBEAT:

×

UNCOV

908

      return qWorkerProcessHbMsg(pVnode, pVnode->pQuery, pMsg, 0);

×

UNCOV

909

    case TDMT_VND_TABLE_META:

×

UNCOV

911

      return vnodeGetTableMeta(pVnode, pMsg, true);

×

912

    case TDMT_VND_TABLE_CFG:

×

913

      return vnodeGetTableCfg(pVnode, pMsg, true);

×

UNCOV

914

    case TDMT_VND_BATCH_META:

×

UNCOV

915

      return vnodeGetBatchMeta(pVnode, pMsg);

×

UNCOV

917

    case TDMT_VND_QUERY_COMPACT_PROGRESS:

×

UNCOV

918

      return vnodeQueryCompactProgress(pVnode, pMsg);

×

UNCOV

922

    case TDMT_VND_TMQ_VG_WALINFO:

×

UNCOV

923

      return tqProcessVgWalInfoReq(pVnode->pTq, pMsg);

×

UNCOV

924

    case TDMT_VND_TMQ_VG_COMMITTEDINFO:

×

UNCOV

925

      return tqProcessVgCommittedInfoReq(pVnode->pTq, pMsg);

×

UNCOV

926

    case TDMT_VND_TMQ_SEEK:

×

UNCOV

927

      return tqProcessSeekReq(pVnode->pTq, pMsg);

×

929

    default:

×

930

      vError("unknown msg type:%d in fetch queue", pMsg->msgType);

×

931

      return TSDB_CODE_APP_ERROR;

×

UNCOV

935

int32_t vnodeProcessStreamMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {

×

UNCOV

936

  vTrace("vgId:%d, msg:%p in stream queue is processing", pVnode->config.vgId, pMsg);

×

NEW

937

  if (!syncIsReadyForRead(pVnode->sync)) {

×

UNCOV

938

    vnodeRedirectRpcMsg(pVnode, pMsg, terrno);

×

939

    return 0;

×

UNCOV

942

  switch (pMsg->msgType) {

×

UNCOV

943

    case TDMT_STREAM_TASK_RUN:

×

UNCOV

944

      return tqProcessTaskRunReq(pVnode->pTq, pMsg);

×

UNCOV

945

    case TDMT_STREAM_RETRIEVE:

×

UNCOV

946

      return tqProcessTaskRetrieveReq(pVnode->pTq, pMsg);

×

UNCOV

947

    case TDMT_STREAM_RETRIEVE_RSP:

×

UNCOV

948

      return tqProcessTaskRetrieveRsp(pVnode->pTq, pMsg);

×

UNCOV

949

    case TDMT_VND_GET_STREAM_PROGRESS:

×

UNCOV

950

      return tqStreamProgressRetrieveReq(pVnode->pTq, pMsg);

×

951

    default:

×

952

      vError("unknown msg type:%d in stream queue", pMsg->msgType);

×

953

      return TSDB_CODE_APP_ERROR;

×

UNCOV

957

int32_t vnodeProcessStreamCtrlMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {

×

UNCOV

958

  vTrace("vgId:%d, msg:%p in stream ctrl queue is processing", pVnode->config.vgId, pMsg);

×

UNCOV

959

  if ((pMsg->msgType == TDMT_SCH_FETCH || pMsg->msgType == TDMT_VND_TABLE_META || pMsg->msgType == TDMT_VND_TABLE_CFG ||

×

UNCOV

960

       pMsg->msgType == TDMT_VND_BATCH_META) &&

×

961

      !syncIsReadyForRead(pVnode->sync)) {

×

962

    vnodeRedirectRpcMsg(pVnode, pMsg, terrno);

×

963

    return 0;

×

UNCOV

966

  switch (pMsg->msgType) {

×

UNCOV

967

    case TDMT_MND_STREAM_HEARTBEAT_RSP:

×

UNCOV

968

      return tqProcessStreamHbRsp(pVnode->pTq, pMsg);

×

UNCOV

969

    case TDMT_STREAM_TASK_DISPATCH:

×

UNCOV

970

      return tqProcessTaskDispatchReq(pVnode->pTq, pMsg);

×

UNCOV

971

    case TDMT_STREAM_TASK_DISPATCH_RSP:

×

UNCOV

972

      return tqProcessTaskDispatchRsp(pVnode->pTq, pMsg);

×

UNCOV

973

    case TDMT_VND_STREAM_TASK_CHECK:

×

UNCOV

974

      return tqProcessTaskCheckReq(pVnode->pTq, pMsg);

×

UNCOV

975

    case TDMT_VND_STREAM_TASK_CHECK_RSP:

×

UNCOV

976

      return tqProcessTaskCheckRsp(pVnode->pTq, pMsg);

×

UNCOV

977

    case TDMT_STREAM_TASK_CHECKPOINT_READY:

×

UNCOV

978

      return tqProcessTaskCheckpointReadyMsg(pVnode->pTq, pMsg);

×

UNCOV

979

    case TDMT_STREAM_TASK_CHECKPOINT_READY_RSP:

×

UNCOV

980

      return tqProcessTaskCheckpointReadyRsp(pVnode->pTq, pMsg);

×

UNCOV

981

    case TDMT_STREAM_RETRIEVE_TRIGGER:

×

UNCOV

982

      return tqProcessTaskRetrieveTriggerReq(pVnode->pTq, pMsg);

×

UNCOV

983

    case TDMT_STREAM_RETRIEVE_TRIGGER_RSP:

×

UNCOV

984

      return tqProcessTaskRetrieveTriggerRsp(pVnode->pTq, pMsg);

×

UNCOV

985

    case TDMT_MND_STREAM_REQ_CHKPT_RSP:

×

UNCOV

986

      return tqProcessStreamReqCheckpointRsp(pVnode->pTq, pMsg);

×

UNCOV

987

    case TDMT_MND_STREAM_CHKPT_REPORT_RSP:

×

UNCOV

988

      return tqProcessTaskChkptReportRsp(pVnode->pTq, pMsg);

×

989

    default:

×

990

      vError("unknown msg type:%d in stream ctrl queue", pMsg->msgType);

×

991

      return TSDB_CODE_APP_ERROR;

×

NEW

995

int32_t vnodeProcessStreamLongExecMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {

×

NEW

996

  vTrace("vgId:%d, msg:%p in stream long exec queue is processing", pVnode->config.vgId, pMsg);

×

NEW

997

  if (!syncIsReadyForRead(pVnode->sync)) {

×

NEW

998

    vnodeRedirectRpcMsg(pVnode, pMsg, terrno);

×

NEW

999

    return 0;

×

NEW

1002

  switch (pMsg->msgType) {

×

NEW

1003

    case TDMT_VND_STREAM_SCAN_HISTORY:

×

NEW

1004

      return tqProcessTaskScanHistory(pVnode->pTq, pMsg);

×

NEW

1005

    default:

×

NEW

1006

      vError("unknown msg type:%d in stream long exec queue", pMsg->msgType);

×

NEW

1007

      return TSDB_CODE_APP_ERROR;

×

UNCOV

1011

void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data) {

×

UNCOV

1012

  int32_t code = tdProcessTSmaInsert(((SVnode *)pVnode)->pSma, smaId, (const char *)data);

×

UNCOV

1013

  if (code) {

×

1014

    vError("failed to process sma result since %s", tstrerror(code));

×

UNCOV

1016

×

UNCOV

1018

void vnodeUpdateMetaRsp(SVnode *pVnode, STableMetaRsp *pMetaRsp) {

×

UNCOV

1019

  if (NULL == pMetaRsp) {

×

1020

    return;

×

UNCOV

1023

  tstrncpy(pMetaRsp->dbFName, pVnode->config.dbname, TSDB_DB_FNAME_LEN);

×

UNCOV

1024

  pMetaRsp->dbId = pVnode->config.dbId;

×

UNCOV

1025

  pMetaRsp->vgId = TD_VID(pVnode);

×

UNCOV

1026

  pMetaRsp->precision = pVnode->config.tsdbCfg.precision;

×

UNCOV

1031

static int32_t vnodeProcessTrimReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

UNCOV

1032

  if (!pVnode->restored) {

×

UNCOV

1033

    vInfo("vgId:%d, ignore trim req during restoring. ver:%" PRId64, TD_VID(pVnode), ver);

×

UNCOV

1034

    return 0;

×

UNCOV

1037

  int32_t     code = 0;

×

UNCOV

1038

  SVTrimDbReq trimReq = {0};

×

UNCOV

1041

  if (tDeserializeSVTrimDbReq(pReq, len, &trimReq) != 0) {

×

1042

    code = TSDB_CODE_INVALID_MSG;

×

1043

    goto _exit;

×

UNCOV

1046

  vInfo("vgId:%d, trim vnode request will be processed, time:%d", pVnode->config.vgId, trimReq.timestamp);

×

UNCOV

1048

  code = vnodeAsyncRetention(pVnode, trimReq.timestamp);

×

UNCOV

1050

_exit:

×

UNCOV

1051

  return code;

×

1056

static int32_t vnodeProcessS3MigrateReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

1057

  int32_t          code = 0;

×

1058

  SVS3MigrateDbReq s3migrateReq = {0};

×

1061

  if (tDeserializeSVS3MigrateDbReq(pReq, len, &s3migrateReq) != 0) {

×

1062

    code = TSDB_CODE_INVALID_MSG;

×

1063

    goto _exit;

×

1066

  vInfo("vgId:%d, s3migrate vnode request will be processed, time:%d", pVnode->config.vgId, s3migrateReq.timestamp);

×

1068

  code = vnodeAsyncS3Migrate(pVnode, s3migrateReq.timestamp);

×

1070

_exit:

×

1071

  return code;

×

1074

static int32_t vnodeProcessDropTtlTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

1075

  int               ret = 0;

×

1076

  SVDropTtlTableReq ttlReq = {0};

×

1077

  if (tDeserializeSVDropTtlTableReq(pReq, len, &ttlReq) != 0) {

×

1078

    ret = TSDB_CODE_INVALID_MSG;

×

1079

    goto end;

×

1082

  if (ttlReq.nUids != taosArrayGetSize(ttlReq.pTbUids)) {

×

1083

    ret = TSDB_CODE_INVALID_MSG;

×

1084

    goto end;

×

1087

  if (ttlReq.nUids != 0) {

×

1088

    vInfo("vgId:%d, drop ttl table req will be processed, time:%d, ntbUids:%d", pVnode->config.vgId,

×

1092

  if (ttlReq.nUids > 0) {

×

1093

    int32_t code = metaDropMultipleTables(pVnode->pMeta, ver, ttlReq.pTbUids);

×

1094

    if (code) return code;

×

1096

    code = tqUpdateTbUidList(pVnode->pTq, ttlReq.pTbUids, false);

×

1097

    if (code) {

×

1098

      vError("vgId:%d, failed to update tbUid list since %s", TD_VID(pVnode), tstrerror(code));

×

1102

end:

×

1103

  taosArrayDestroy(ttlReq.pTbUids);

×

1104

  return ret;

×

UNCOV

1107

static int32_t vnodeProcessFetchTtlExpiredTbs(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

UNCOV

1108

  int32_t                 code = -1;

×

UNCOV

1109

  SMetaReader             mr = {0};

×

UNCOV

1110

  SVDropTtlTableReq       ttlReq = {0};

×

UNCOV

1111

  SVFetchTtlExpiredTbsRsp rsp = {0};

×

UNCOV

1112

  SEncoder                encoder = {0};

×

UNCOV

1113

  SArray                 *pNames = NULL;

×

UNCOV

1114

  pRsp->msgType = TDMT_VND_FETCH_TTL_EXPIRED_TBS_RSP;

×

UNCOV

1115

  pRsp->code = TSDB_CODE_SUCCESS;

×

UNCOV

1116

  pRsp->pCont = NULL;

×

UNCOV

1117

  pRsp->contLen = 0;

×

UNCOV

1119

  if (tDeserializeSVDropTtlTableReq(pReq, len, &ttlReq) != 0) {

×

1120

    terrno = TSDB_CODE_INVALID_MSG;

×

1121

    goto _end;

×

UNCOV

1124

  if (!(ttlReq.nUids == taosArrayGetSize(ttlReq.pTbUids))) {

×

1125

    terrno = TSDB_CODE_INVALID_MSG;

×

1126

    goto _end;

×

UNCOV

1131

  SVDropTbReq expiredTb = {.igNotExists = true};

×

UNCOV

1132

  metaReaderDoInit(&mr, pVnode->pMeta, 0);

×

UNCOV

1133

  rsp.vgId = TD_VID(pVnode);

×

UNCOV

1134

  rsp.pExpiredTbs = taosArrayInit(ttlReq.nUids, sizeof(SVDropTbReq));

×

UNCOV

1135

  if (!rsp.pExpiredTbs) goto _end;

×

UNCOV

1137

  pNames = taosArrayInit(ttlReq.nUids, TSDB_TABLE_NAME_LEN);

×

UNCOV

1138

  if (!pNames) {

×

1139

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1140

    goto _end;

×

UNCOV

1143

  for (int32_t i = 0; i < ttlReq.nUids; ++i) {

×

UNCOV

1144

    tb_uid_t *uid = taosArrayGet(ttlReq.pTbUids, i);

×

UNCOV

1145

    expiredTb.suid = *uid;

×

UNCOV

1146

    terrno = metaReaderGetTableEntryByUid(&mr, *uid);

×

UNCOV

1147

    if (terrno < 0) goto _end;

×

UNCOV

1148

    tstrncpy(buf, mr.me.name, TSDB_TABLE_NAME_LEN);

×

UNCOV

1149

    void *p = taosArrayPush(pNames, buf);

×

UNCOV

1150

    if (p == NULL) {

×

1151

      goto _end;

×

UNCOV

1154

    expiredTb.name = p;

×

UNCOV

1155

    if (mr.me.type == TSDB_CHILD_TABLE) {

×

UNCOV

1156

      expiredTb.suid = mr.me.ctbEntry.suid;

×

UNCOV

1159

    if (taosArrayPush(rsp.pExpiredTbs, &expiredTb) == NULL) {

×

1160

      goto _end;

×

UNCOV

1164

  int32_t ret = 0;

×

UNCOV

1165

  tEncodeSize(tEncodeVFetchTtlExpiredTbsRsp, &rsp, pRsp->contLen, ret);

×

UNCOV

1166

  pRsp->pCont = rpcMallocCont(pRsp->contLen);

×

UNCOV

1167

  if (pRsp->pCont == NULL) {

×

1168

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1169

    code = -1;

×

1170

    goto _end;

×

UNCOV

1172

  tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);

×

UNCOV

1173

  terrno = tEncodeVFetchTtlExpiredTbsRsp(&encoder, &rsp);

×

UNCOV

1174

  tEncoderClear(&encoder);

×

UNCOV

1176

  if (terrno == 0) code = 0;

×

1177

_end:

×

UNCOV

1178

  metaReaderClear(&mr);

×

UNCOV

1179

  tFreeFetchTtlExpiredTbsRsp(&rsp);

×

UNCOV

1180

  taosArrayDestroy(ttlReq.pTbUids);

×

UNCOV

1181

  if (pNames) taosArrayDestroy(pNames);

×

UNCOV

1182

  pRsp->code = terrno;

×

UNCOV

1183

  return code;

×

UNCOV

1186

static int32_t vnodeProcessCreateStbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

UNCOV

1187

  int32_t        code = 0;

×

UNCOV

1188

  SVCreateStbReq req = {0};

×

UNCOV

1191

  pRsp->msgType = TDMT_VND_CREATE_STB_RSP;

×

UNCOV

1192

  pRsp->code = TSDB_CODE_SUCCESS;

×

UNCOV

1193

  pRsp->pCont = NULL;

×

UNCOV

1194

  pRsp->contLen = 0;

×

UNCOV

1197

  tDecoderInit(&coder, pReq, len);

×

UNCOV

1199

  code = tDecodeSVCreateStbReq(&coder, &req);

×

UNCOV

1200

  if (code) {

×

1201

    pRsp->code = code;

×

1202

    goto _err;

×

UNCOV

1205

  code = metaCreateSuperTable(pVnode->pMeta, ver, &req);

×

UNCOV

1206

  if (code) {

×

UNCOV

1207

    pRsp->code = code;

×

UNCOV

1208

    goto _err;

×

UNCOV

1211

  if ((code = tdProcessRSmaCreate(pVnode->pSma, &req)) < 0) {

×

1212

    pRsp->code = code;

×

1213

    goto _err;

×

UNCOV

1216

  tDecoderClear(&coder);

×

UNCOV

1217

  return 0;

×

UNCOV

1219

_err:

×

UNCOV

1220

  tDecoderClear(&coder);

×

UNCOV

1221

  return code;

×

UNCOV

1224

static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,

×

UNCOV

1226

  SDecoder           decoder = {0};

×

UNCOV

1227

  SEncoder           encoder = {0};

×

UNCOV

1228

  int32_t            rcode = 0;

×

UNCOV

1229

  SVCreateTbBatchReq req = {0};

×

UNCOV

1231

  SVCreateTbBatchRsp rsp = {0};

×

UNCOV

1232

  SVCreateTbRsp      cRsp = {0};

×

UNCOV

1234

  STbUidStore       *pStore = NULL;

×

UNCOV

1235

  SArray            *tbUids = NULL;

×

UNCOV

1236

  SArray            *tbNames = NULL;

×

UNCOV

1237

  pRsp->msgType = TDMT_VND_CREATE_TABLE_RSP;

×

UNCOV

1238

  pRsp->code = TSDB_CODE_SUCCESS;

×

UNCOV

1239

  pRsp->pCont = NULL;

×

UNCOV

1240

  pRsp->contLen = 0;

×

UNCOV

1243

  tDecoderInit(&decoder, pReq, len);

×

UNCOV

1244

  if (tDecodeSVCreateTbBatchReq(&decoder, &req) < 0) {

×

1245

    rcode = -1;

×

1246

    terrno = TSDB_CODE_INVALID_MSG;

×

1247

    goto _exit;

×

UNCOV

1250

  rsp.pArray = taosArrayInit(req.nReqs, sizeof(cRsp));

×

UNCOV

1251

  tbUids = taosArrayInit(req.nReqs, sizeof(int64_t));

×

UNCOV

1252

  tbNames = taosArrayInit(req.nReqs, sizeof(char *));

×

UNCOV

1253

  if (rsp.pArray == NULL || tbUids == NULL || tbNames == NULL) {

×

UNCOV

1254

    rcode = -1;

×

UNCOV

1255

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1256

    goto _exit;

×

UNCOV

1260

  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {

×

UNCOV

1261

    pCreateReq = req.pReqs + iReq;

×

UNCOV

1262

    memset(&cRsp, 0, sizeof(cRsp));

×

UNCOV

1264

    if (tsEnableAudit && tsEnableAuditCreateTable) {

×

UNCOV

1265

      char *str = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN);

×

UNCOV

1266

      if (str == NULL) {

×

1267

        terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1268

        rcode = -1;

×

1269

        goto _exit;

×

UNCOV

1271

      tstrncpy(str, pCreateReq->name, TSDB_TABLE_FNAME_LEN);

×

UNCOV

1272

      if (taosArrayPush(tbNames, &str) == NULL) {

×

1273

        terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1274

        rcode = -1;

×

1275

        goto _exit;

×

UNCOV

1280

    (void)tsnprintf(tbName, TSDB_TABLE_FNAME_LEN, "%s.%s", pVnode->config.dbname, pCreateReq->name);

×

UNCOV

1281

    if (vnodeValidateTableHash(pVnode, tbName) < 0) {

×

UNCOV

1282

      cRsp.code = TSDB_CODE_VND_HASH_MISMATCH;

×

UNCOV

1283

      if (taosArrayPush(rsp.pArray, &cRsp) == NULL) {

×

1284

        terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1285

        rcode = -1;

×

1286

        goto _exit;

×

UNCOV

1288

      vError("vgId:%d create-table:%s failed due to hash value mismatch", TD_VID(pVnode), tbName);

×

UNCOV

1289

      continue;

×

UNCOV

1293

    if (metaCreateTable2(pVnode->pMeta, ver, pCreateReq, &cRsp.pMeta) < 0) {

×

UNCOV

1294

      if (pCreateReq->flags & TD_CREATE_IF_NOT_EXISTS && terrno == TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {

×

UNCOV

1295

        cRsp.code = TSDB_CODE_SUCCESS;

×

UNCOV

1297

        cRsp.code = terrno;

×

UNCOV

1300

      cRsp.code = TSDB_CODE_SUCCESS;

×

UNCOV

1301

      if (tdFetchTbUidList(pVnode->pSma, &pStore, pCreateReq->ctb.suid, pCreateReq->uid) < 0) {

×

1302

        vError("vgId:%d, failed to fetch tbUid list", TD_VID(pVnode));

×

UNCOV

1304

      if (taosArrayPush(tbUids, &pCreateReq->uid) == NULL) {

×

1305

        terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1306

        rcode = -1;

×

1307

        goto _exit;

×

UNCOV

1309

      vnodeUpdateMetaRsp(pVnode, cRsp.pMeta);

×

UNCOV

1312

    if (taosArrayPush(rsp.pArray, &cRsp) == NULL) {

×

1313

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1314

      rcode = -1;

×

1315

      goto _exit;

×

UNCOV

1319

  vDebug("vgId:%d, add %d new created tables into query table list", TD_VID(pVnode), (int32_t)taosArrayGetSize(tbUids));

×

UNCOV

1320

  if (tqUpdateTbUidList(pVnode->pTq, tbUids, true) < 0) {

×

1321

    vError("vgId:%d, failed to update tbUid list since %s", TD_VID(pVnode), tstrerror(terrno));

×

UNCOV

1323

  if (tdUpdateTbUidList(pVnode->pSma, pStore, true) < 0) {

×

1324

    goto _exit;

×

UNCOV

1326

  pStore = tdUidStoreFree(pStore);

×

UNCOV

1329

  int32_t ret = 0;

×

UNCOV

1330

  tEncodeSize(tEncodeSVCreateTbBatchRsp, &rsp, pRsp->contLen, ret);

×

UNCOV

1331

  pRsp->pCont = rpcMallocCont(pRsp->contLen);

×

UNCOV

1332

  if (pRsp->pCont == NULL) {

×

1333

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1334

    rcode = -1;

×

1335

    goto _exit;

×

UNCOV

1337

  tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);

×

UNCOV

1338

  if (tEncodeSVCreateTbBatchRsp(&encoder, &rsp) != 0) {

×

1339

    vError("vgId:%d, failed to encode create table batch response", TD_VID(pVnode));

×

UNCOV

1342

  if (tsEnableAudit && tsEnableAuditCreateTable) {

×

UNCOV

1343

    int64_t clusterId = pVnode->config.syncCfg.nodeInfo[0].clusterId;

×

UNCOV

1345

    SName name = {0};

×

UNCOV

1346

    if (tNameFromString(&name, pVnode->config.dbname, T_NAME_ACCT | T_NAME_DB) < 0) {

×

1347

      vError("vgId:%d, failed to get name from string", TD_VID(pVnode));

×

UNCOV

1350

    SStringBuilder sb = {0};

×

UNCOV

1351

    for (int32_t i = 0; i < tbNames->size; i++) {

×

UNCOV

1352

      char **key = (char **)taosArrayGet(tbNames, i);

×

UNCOV

1353

      taosStringBuilderAppendStringLen(&sb, *key, strlen(*key));

×

UNCOV

1354

      if (i < tbNames->size - 1) {

×

UNCOV

1355

        taosStringBuilderAppendChar(&sb, ',');

×

UNCOV

1360

    size_t len = 0;

×

UNCOV

1361

    char  *keyJoined = taosStringBuilderGetResult(&sb, &len);

×

UNCOV

1363

    if (pOriginRpc->info.conn.user != NULL && strlen(pOriginRpc->info.conn.user) > 0) {

×

UNCOV

1364

      auditAddRecord(pOriginRpc, clusterId, "createTable", name.dbname, "", keyJoined, len);

×

UNCOV

1367

    taosStringBuilderDestroy(&sb);

×

UNCOV

1370

_exit:

×

UNCOV

1371

  tDeleteSVCreateTbBatchReq(&req);

×

UNCOV

1372

  taosArrayDestroyEx(rsp.pArray, tFreeSVCreateTbRsp);

×

UNCOV

1373

  taosArrayDestroy(tbUids);

×

UNCOV

1374

  tDecoderClear(&decoder);

×

UNCOV

1375

  tEncoderClear(&encoder);

×

UNCOV

1376

  taosArrayDestroyP(tbNames, NULL);

×

UNCOV

1377

  return rcode;

×

UNCOV

1380

static int32_t vnodeProcessAlterStbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

UNCOV

1381

  int32_t        code = 0;

×

UNCOV

1382

  SVCreateStbReq req = {0};

×

UNCOV

1383

  SDecoder       dc = {0};

×

UNCOV

1385

  pRsp->msgType = TDMT_VND_ALTER_STB_RSP;

×

UNCOV

1386

  pRsp->code = TSDB_CODE_SUCCESS;

×

UNCOV

1387

  pRsp->pCont = NULL;

×

UNCOV

1388

  pRsp->contLen = 0;

×

UNCOV

1390

  tDecoderInit(&dc, pReq, len);

×

UNCOV

1393

  code = tDecodeSVCreateStbReq(&dc, &req);

×

UNCOV

1394

  if (code) {

×

1395

    tDecoderClear(&dc);

×

1396

    return code;

×

UNCOV

1399

  code = metaAlterSuperTable(pVnode->pMeta, ver, &req);

×

UNCOV

1400

  if (code) {

×

1401

    pRsp->code = code;

×

1402

    tDecoderClear(&dc);

×

1403

    return code;

×

UNCOV

1406

  tDecoderClear(&dc);

×

UNCOV

1408

  return 0;

×

UNCOV

1411

static int32_t vnodeProcessDropStbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

UNCOV

1412

  SVDropStbReq req = {0};

×

UNCOV

1413

  int32_t      rcode = TSDB_CODE_SUCCESS;

×

UNCOV

1414

  SDecoder     decoder = {0};

×

UNCOV

1415

  SArray      *tbUidList = NULL;

×

UNCOV

1417

  pRsp->msgType = TDMT_VND_CREATE_STB_RSP;

×

UNCOV

1418

  pRsp->pCont = NULL;

×

UNCOV

1419

  pRsp->contLen = 0;

×

UNCOV

1422

  tDecoderInit(&decoder, pReq, len);

×

UNCOV

1423

  if (tDecodeSVDropStbReq(&decoder, &req) < 0) {

×

1424

    rcode = TSDB_CODE_INVALID_MSG;

×

1425

    goto _exit;

×

UNCOV

1429

  tbUidList = taosArrayInit(8, sizeof(int64_t));

×

UNCOV

1430

  if (tbUidList == NULL) goto _exit;

×

UNCOV

1431

  if (metaDropSuperTable(pVnode->pMeta, ver, &req) < 0) {

×

UNCOV

1432

    rcode = terrno;

×

UNCOV

1433

    goto _exit;

×

UNCOV

1436

  if (tqUpdateTbUidList(pVnode->pTq, tbUidList, false) < 0) {

×

1437

    rcode = terrno;

×

1438

    goto _exit;

×

UNCOV

1441

  if (tdProcessRSmaDrop(pVnode->pSma, &req) < 0) {

×

1442

    rcode = terrno;

×

1443

    goto _exit;

×

UNCOV

1447

_exit:

×

UNCOV

1448

  if (tbUidList) taosArrayDestroy(tbUidList);

×

UNCOV

1449

  pRsp->code = rcode;

×

UNCOV

1450

  tDecoderClear(&decoder);

×

UNCOV

1451

  return 0;

×

UNCOV

1454

static int32_t vnodeProcessAlterTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

UNCOV

1455

  SVAlterTbReq  vAlterTbReq = {0};

×

UNCOV

1456

  SVAlterTbRsp  vAlterTbRsp = {0};

×

UNCOV

1457

  SDecoder      dc = {0};

×

UNCOV

1458

  int32_t       code = 0;

×

UNCOV

1459

  int32_t       lino = 0;

×

UNCOV

1461

  SEncoder      ec = {0};

×

UNCOV

1462

  STableMetaRsp vMetaRsp = {0};

×

UNCOV

1464

  pRsp->msgType = TDMT_VND_ALTER_TABLE_RSP;

×

UNCOV

1465

  pRsp->pCont = NULL;

×

UNCOV

1466

  pRsp->contLen = 0;

×

UNCOV

1467

  pRsp->code = TSDB_CODE_SUCCESS;

×

UNCOV

1469

  tDecoderInit(&dc, pReq, len);

×

UNCOV

1472

  if (tDecodeSVAlterTbReq(&dc, &vAlterTbReq) < 0) {

×

1473

    vAlterTbRsp.code = TSDB_CODE_INVALID_MSG;

×

1474

    tDecoderClear(&dc);

×

1475

    goto _exit;

×

UNCOV

1479

  if (metaAlterTable(pVnode->pMeta, ver, &vAlterTbReq, &vMetaRsp) < 0) {

×

UNCOV

1480

    vAlterTbRsp.code = terrno;

×

UNCOV

1481

    tDecoderClear(&dc);

×

UNCOV

1482

    goto _exit;

×

UNCOV

1484

  tDecoderClear(&dc);

×

UNCOV

1486

  if (NULL != vMetaRsp.pSchemas) {

×

UNCOV

1487

    vnodeUpdateMetaRsp(pVnode, &vMetaRsp);

×

UNCOV

1488

    vAlterTbRsp.pMeta = &vMetaRsp;

×

UNCOV

1491

  if (vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_TAG_VAL || vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_MULTI_TAG_VAL) {

×

UNCOV

1492

    int64_t uid = metaGetTableEntryUidByName(pVnode->pMeta, vAlterTbReq.tbName);

×

UNCOV

1493

    if (uid == 0) {

×

1494

      vError("vgId:%d, %s failed at %s:%d since table %s not found", TD_VID(pVnode), __func__, __FILE__, __LINE__,

×

1496

      goto _exit;

×

UNCOV

1499

    SArray* tbUids = taosArrayInit(4, sizeof(int64_t));

×

UNCOV

1500

    void* p = taosArrayPush(tbUids, &uid);

×

UNCOV

1501

    TSDB_CHECK_NULL(p, code, lino, _exit, terrno);

×

UNCOV

1503

    vDebug("vgId:%d, remove tags value altered table:%s from query table list", TD_VID(pVnode), vAlterTbReq.tbName);

×

UNCOV

1504

    if ((code = tqUpdateTbUidList(pVnode->pTq, tbUids, false)) < 0) {

×

1505

      vError("vgId:%d, failed to remove tbUid list since %s", TD_VID(pVnode), tstrerror(code));

×

UNCOV

1508

    vDebug("vgId:%d, try to add table:%s in query table list", TD_VID(pVnode), vAlterTbReq.tbName);

×

UNCOV

1509

    if ((code = tqUpdateTbUidList(pVnode->pTq, tbUids, true)) < 0) {

×

1510

      vError("vgId:%d, failed to add tbUid list since %s", TD_VID(pVnode), tstrerror(code));

×

UNCOV

1513

    taosArrayDestroy(tbUids);

×

UNCOV

1516

_exit:

×

UNCOV

1517

  taosArrayDestroy(vAlterTbReq.pMultiTag);

×

UNCOV

1518

  tEncodeSize(tEncodeSVAlterTbRsp, &vAlterTbRsp, pRsp->contLen, ret);

×

UNCOV

1519

  pRsp->pCont = rpcMallocCont(pRsp->contLen);

×

UNCOV

1520

  tEncoderInit(&ec, pRsp->pCont, pRsp->contLen);

×

UNCOV

1521

  if (tEncodeSVAlterTbRsp(&ec, &vAlterTbRsp) != 0) {

×

1522

    vError("vgId:%d, failed to encode alter table response", TD_VID(pVnode));

×

UNCOV

1525

  tEncoderClear(&ec);

×

UNCOV

1526

  if (vMetaRsp.pSchemas) {

×

UNCOV

1527

    taosMemoryFree(vMetaRsp.pSchemas);

×

UNCOV

1528

    taosMemoryFree(vMetaRsp.pSchemaExt);

×

UNCOV

1530

  return 0;

×

UNCOV

1533

static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,

×

UNCOV

1535

  SVDropTbBatchReq req = {0};

×

UNCOV

1536

  SVDropTbBatchRsp rsp = {0};

×

UNCOV

1537

  SDecoder         decoder = {0};

×

UNCOV

1538

  SEncoder         encoder = {0};

×

UNCOV

1540

  SArray          *tbUids = NULL;

×

UNCOV

1541

  STbUidStore     *pStore = NULL;

×

UNCOV

1542

  SArray          *tbNames = NULL;

×

UNCOV

1544

  pRsp->msgType = TDMT_VND_DROP_TABLE_RSP;

×

UNCOV

1545

  pRsp->pCont = NULL;

×

UNCOV

1546

  pRsp->contLen = 0;

×

UNCOV

1547

  pRsp->code = TSDB_CODE_SUCCESS;

×

UNCOV

1550

  tDecoderInit(&decoder, pReq, len);

×

UNCOV

1551

  ret = tDecodeSVDropTbBatchReq(&decoder, &req);

×

UNCOV

1552

  if (ret < 0) {

×

1553

    terrno = TSDB_CODE_INVALID_MSG;

×

1554

    pRsp->code = terrno;

×

1555

    goto _exit;

×

UNCOV

1559

  tbUids = taosArrayInit(req.nReqs, sizeof(int64_t));

×

UNCOV

1560

  rsp.pArray = taosArrayInit(req.nReqs, sizeof(SVDropTbRsp));

×

UNCOV

1561

  tbNames = taosArrayInit(req.nReqs, sizeof(char *));

×

UNCOV

1562

  if (tbUids == NULL || rsp.pArray == NULL || tbNames == NULL) goto _exit;

×

UNCOV

1564

  for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {

×

UNCOV

1565

    SVDropTbReq *pDropTbReq = req.pReqs + iReq;

×

UNCOV

1566

    SVDropTbRsp  dropTbRsp = {0};

×

UNCOV

1567

    tb_uid_t     tbUid = 0;

×

UNCOV

1570

    ret = metaDropTable2(pVnode->pMeta, ver, pDropTbReq);

×

UNCOV

1571

    if (ret < 0) {

×

1572

      if (pDropTbReq->igNotExists && terrno == TSDB_CODE_TDB_TABLE_NOT_EXIST) {

×

1573

        dropTbRsp.code = TSDB_CODE_SUCCESS;

×

1575

        dropTbRsp.code = terrno;

×

UNCOV

1578

      dropTbRsp.code = TSDB_CODE_SUCCESS;

×

UNCOV

1579

      if (tbUid > 0) {

×

1580

        if (tdFetchTbUidList(pVnode->pSma, &pStore, pDropTbReq->suid, tbUid) < 0) {

×

1581

          vError("vgId:%d, failed to fetch tbUid list", TD_VID(pVnode));

×

UNCOV

1586

    if (taosArrayPush(rsp.pArray, &dropTbRsp) == NULL) {

×

1587

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1588

      pRsp->code = terrno;

×

1589

      goto _exit;

×

UNCOV

1592

    if (tsEnableAuditCreateTable) {

×

UNCOV

1593

      char *str = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN);

×

UNCOV

1594

      if (str == NULL) {

×

1595

        pRsp->code = terrno;

×

1596

        goto _exit;

×

UNCOV

1598

      tstrncpy(str, pDropTbReq->name, TSDB_TABLE_FNAME_LEN);

×

UNCOV

1599

      if (taosArrayPush(tbNames, &str) == NULL) {

×

1600

        terrno = TSDB_CODE_OUT_OF_MEMORY;

×

1601

        pRsp->code = terrno;

×

1602

        goto _exit;

×

UNCOV

1607

  if (tqUpdateTbUidList(pVnode->pTq, tbUids, false) < 0) {

×

1608

    vError("vgId:%d, failed to update tbUid list since %s", TD_VID(pVnode), tstrerror(terrno));

×

UNCOV

1611

  if (tdUpdateTbUidList(pVnode->pSma, pStore, false) < 0) {

×

1612

    goto _exit;

×

UNCOV

1615

  if (tsEnableAuditCreateTable) {

×

UNCOV

1616

    int64_t clusterId = pVnode->config.syncCfg.nodeInfo[0].clusterId;

×

UNCOV

1618

    SName name = {0};

×

UNCOV

1619

    if (tNameFromString(&name, pVnode->config.dbname, T_NAME_ACCT | T_NAME_DB) != 0) {

×

1620

      vError("vgId:%d, failed to get name from string", TD_VID(pVnode));

×

UNCOV

1623

    SStringBuilder sb = {0};

×

UNCOV

1624

    for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {

×

UNCOV

1625

      char **key = (char **)taosArrayGet(tbNames, iReq);

×

UNCOV

1626

      taosStringBuilderAppendStringLen(&sb, *key, strlen(*key));

×

UNCOV

1627

      if (iReq < req.nReqs - 1) {

×

UNCOV

1628

        taosStringBuilderAppendChar(&sb, ',');

×

UNCOV

1630

      taosMemoryFreeClear(*key);

×

UNCOV

1633

    size_t len = 0;

×

UNCOV

1634

    char  *keyJoined = taosStringBuilderGetResult(&sb, &len);

×

UNCOV

1636

    if (pOriginRpc->info.conn.user != NULL && strlen(pOriginRpc->info.conn.user) > 0) {

×

UNCOV

1637

      auditAddRecord(pOriginRpc, clusterId, "dropTable", name.dbname, "", keyJoined, len);

×

UNCOV

1640

    taosStringBuilderDestroy(&sb);

×

1643

_exit:

×

UNCOV

1644

  taosArrayDestroy(tbUids);

×

UNCOV

1645

  pStore = tdUidStoreFree(pStore);

×

UNCOV

1646

  tDecoderClear(&decoder);

×

UNCOV

1647

  tEncodeSize(tEncodeSVDropTbBatchRsp, &rsp, pRsp->contLen, ret);

×

UNCOV

1648

  pRsp->pCont = rpcMallocCont(pRsp->contLen);

×

UNCOV

1649

  tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);

×

UNCOV

1650

  if (tEncodeSVDropTbBatchRsp(&encoder, &rsp) != 0) {

×

1651

    vError("vgId:%d, failed to encode drop table batch response", TD_VID(pVnode));

×

UNCOV

1653

  tEncoderClear(&encoder);

×

UNCOV

1654

  taosArrayDestroy(rsp.pArray);

×

UNCOV

1655

  taosArrayDestroy(tbNames);

×

UNCOV

1656

  return 0;

×

1703

static int32_t vnodeResetTableCxt(SMeta *pMeta, SSubmitReqConvertCxt *pCxt) {

×

1704

  taosMemoryFreeClear(pCxt->pTbSchema);

×

1705

  int32_t code = metaGetTbTSchemaNotNull(pMeta, pCxt->msgIter.suid > 0 ? pCxt->msgIter.suid : pCxt->msgIter.uid,

×

1707

  if (TSDB_CODE_SUCCESS != code) {

×

1708

    return code;

×

1710

  tdSTSRowIterInit(&pCxt->rowIter, pCxt->pTbSchema);

×

1712

  tDestroySubmitTbData(pCxt->pTbData, TSDB_MSG_FLG_ENCODE);

×

1713

  if (NULL == pCxt->pTbData) {

×

1714

    pCxt->pTbData = taosMemoryCalloc(1, sizeof(SSubmitTbData));

×

1715

    if (NULL == pCxt->pTbData) {

×

1716

      return terrno;

×

1719

  pCxt->pTbData->flags = 0;

×

1720

  pCxt->pTbData->suid = pCxt->msgIter.suid;

×

1721

  pCxt->pTbData->uid = pCxt->msgIter.uid;

×

1722

  pCxt->pTbData->sver = pCxt->msgIter.sversion;

×

1723

  pCxt->pTbData->pCreateTbReq = NULL;

×

1724

  pCxt->pTbData->aRowP = taosArrayInit(128, POINTER_BYTES);

×

1725

  if (NULL == pCxt->pTbData->aRowP) {

×

1726

    return terrno;

×

1729

  taosArrayDestroy(pCxt->pColValues);

×

1730

  pCxt->pColValues = taosArrayInit(pCxt->pTbSchema->numOfCols, sizeof(SColVal));

×

1731

  if (NULL == pCxt->pColValues) {

×

1732

    return terrno;

×

1734

  for (int32_t i = 0; i < pCxt->pTbSchema->numOfCols; ++i) {

×

1735

    SColVal val = COL_VAL_NONE(pCxt->pTbSchema->columns[i].colId, pCxt->pTbSchema->columns[i].type);

×

1736

    if (taosArrayPush(pCxt->pColValues, &val) == NULL) {

×

1737

      return terrno;

×

1741

  return TSDB_CODE_SUCCESS;

×

1744

static void vnodeDestroySubmitReqConvertCxt(SSubmitReqConvertCxt *pCxt) {

×

1745

  taosMemoryFreeClear(pCxt->pTbSchema);

×

1746

  tDestroySubmitTbData(pCxt->pTbData, TSDB_MSG_FLG_ENCODE);

×

1747

  taosMemoryFreeClear(pCxt->pTbData);

×

1748

  taosArrayDestroy(pCxt->pColValues);

×

1749

×

1751

static int32_t vnodeCellValConvertToColVal(STColumn *pCol, SCellVal *pCellVal, SColVal *pColVal) {

×

1752

  if (tdValTypeIsNone(pCellVal->valType)) {

×

1753

    pColVal->flag = CV_FLAG_NONE;

×

1754

    return TSDB_CODE_SUCCESS;

×

1757

  if (tdValTypeIsNull(pCellVal->valType)) {

×

1758

    pColVal->flag = CV_FLAG_NULL;

×

1759

    return TSDB_CODE_SUCCESS;

×

1762

  if (IS_VAR_DATA_TYPE(pCol->type)) {

×

1763

    pColVal->value.nData = varDataLen(pCellVal->val);

×

1764

    pColVal->value.pData = (uint8_t *)varDataVal(pCellVal->val);

×

1765

  } else if (TSDB_DATA_TYPE_FLOAT == pCol->type) {

×

1766

    float f = GET_FLOAT_VAL(pCellVal->val);

×

1767

    memcpy(&pColVal->value.val, &f, sizeof(f));

×

1768

  } else if (TSDB_DATA_TYPE_DOUBLE == pCol->type) {

×

1769

    pColVal->value.val = *(int64_t *)pCellVal->val;

×

1771

    GET_TYPED_DATA(pColVal->value.val, int64_t, pCol->type, pCellVal->val);

×

1774

  pColVal->flag = CV_FLAG_VALUE;

×

1775

  return TSDB_CODE_SUCCESS;

×

1778

static int32_t vnodeTSRowConvertToColValArray(SSubmitReqConvertCxt *pCxt) {

×

1779

  int32_t code = TSDB_CODE_SUCCESS;

×

1780

  tdSTSRowIterReset(&pCxt->rowIter, pCxt->pRow);

×

1781

  for (int32_t i = 0; TSDB_CODE_SUCCESS == code && i < pCxt->pTbSchema->numOfCols; ++i) {

×

1782

    STColumn *pCol = pCxt->pTbSchema->columns + i;

×

1783

    SCellVal  cellVal = {0};

×

1784

    if (!tdSTSRowIterFetch(&pCxt->rowIter, pCol->colId, pCol->type, &cellVal)) {

×

1785

      break;

×

1787

    code = vnodeCellValConvertToColVal(pCol, &cellVal, (SColVal *)taosArrayGet(pCxt->pColValues, i));

×

1789

  return code;

×

1792

static int32_t vnodeDecodeCreateTbReq(SSubmitReqConvertCxt *pCxt) {

×

1793

  if (pCxt->msgIter.schemaLen <= 0) {

×

1794

    return TSDB_CODE_SUCCESS;

×

1797

  pCxt->pTbData->pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));

×

1798

  if (NULL == pCxt->pTbData->pCreateTbReq) {

×

1799

    return terrno;

×

1802

  SDecoder decoder = {0};

×

1803

  tDecoderInit(&decoder, (uint8_t *)pCxt->pBlock->data, pCxt->msgIter.schemaLen);

×

1804

  int32_t code = tDecodeSVCreateTbReq(&decoder, pCxt->pTbData->pCreateTbReq);

×

1805

  tDecoderClear(&decoder);

×

1807

  return code;

×

1810

static int32_t vnodeSubmitReqConvertToSubmitReq2(SVnode *pVnode, SSubmitReq *pReq, SSubmitReq2 *pReq2) {

×

1811

  pReq2->aSubmitTbData = taosArrayInit(128, sizeof(SSubmitTbData));

×

1812

  if (NULL == pReq2->aSubmitTbData) {

×

1813

    return terrno;

×

1816

  SSubmitReqConvertCxt cxt = {0};

×

1818

  int32_t code = tInitSubmitMsgIter(pReq, &cxt.msgIter);

×

1819

  while (TSDB_CODE_SUCCESS == code) {

×

1820

    code = tGetSubmitMsgNext(&cxt.msgIter, &cxt.pBlock);

×

1821

    if (TSDB_CODE_SUCCESS == code) {

×

1822

      if (NULL == cxt.pBlock) {

×

1823

        break;

×

1825

      code = vnodeResetTableCxt(pVnode->pMeta, &cxt);

×

1827

    if (TSDB_CODE_SUCCESS == code) {

×

1828

      code = tInitSubmitBlkIter(&cxt.msgIter, cxt.pBlock, &cxt.blkIter);

×

1830

    if (TSDB_CODE_SUCCESS == code) {

×

1831

      code = vnodeDecodeCreateTbReq(&cxt);

×

1833

    while (TSDB_CODE_SUCCESS == code && (cxt.pRow = tGetSubmitBlkNext(&cxt.blkIter)) != NULL) {

×

1834

      code = vnodeTSRowConvertToColValArray(&cxt);

×

1835

      if (TSDB_CODE_SUCCESS == code) {

×

1836

        SRow **pNewRow = taosArrayReserve(cxt.pTbData->aRowP, 1);

×

1837

        code = tRowBuild(cxt.pColValues, cxt.pTbSchema, pNewRow);

×

1840

    if (TSDB_CODE_SUCCESS == code) {

×

1841

      code = (NULL == taosArrayPush(pReq2->aSubmitTbData, cxt.pTbData) ? terrno : TSDB_CODE_SUCCESS);

×

1843

    if (TSDB_CODE_SUCCESS == code) {

×

1844

      taosMemoryFreeClear(cxt.pTbData);

×

1848

  vnodeDestroySubmitReqConvertCxt(&cxt);

×

1849

  return code;

×

1852

static int32_t vnodeRebuildSubmitReqMsg(SSubmitReq2 *pSubmitReq, void **ppMsg) {

×

1853

  int32_t  code = TSDB_CODE_SUCCESS;

×

1854

  char    *pMsg = NULL;

×

1855

  uint32_t msglen = 0;

×

1856

  tEncodeSize(tEncodeSubmitReq, pSubmitReq, msglen, code);

×

1857

  if (TSDB_CODE_SUCCESS == code) {

×

1858

    pMsg = taosMemoryMalloc(msglen);

×

1859

    if (NULL == pMsg) {

×

1860

      code = terrno;

×

1863

  if (TSDB_CODE_SUCCESS == code) {

×

1865

    tEncoderInit(&encoder, (uint8_t *)pMsg, msglen);

×

1866

    code = tEncodeSubmitReq(&encoder, pSubmitReq);

×

1867

    tEncoderClear(&encoder);

×

1869

  if (TSDB_CODE_SUCCESS == code) {

×

1870

    *ppMsg = pMsg;

×

1872

  return code;

×

UNCOV

1875

static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,

×

UNCOV

1877

  int32_t code = 0;

×

UNCOV

1878

  terrno = 0;

×

UNCOV

1880

  SSubmitReq2 *pSubmitReq = &(SSubmitReq2){0};

×

UNCOV

1881

  SSubmitRsp2 *pSubmitRsp = &(SSubmitRsp2){0};

×

UNCOV

1882

  SArray      *newTbUids = NULL;

×

UNCOV

1884

  SEncoder     ec = {0};

×

UNCOV

1886

  pRsp->code = TSDB_CODE_SUCCESS;

×

UNCOV

1888

  void           *pAllocMsg = NULL;

×

UNCOV

1889

  SSubmitReq2Msg *pMsg = (SSubmitReq2Msg *)pReq;

×

UNCOV

1890

  if (0 == pMsg->version) {

×

1891

    code = vnodeSubmitReqConvertToSubmitReq2(pVnode, (SSubmitReq *)pMsg, pSubmitReq);

×

1892

    if (TSDB_CODE_SUCCESS == code) {

×

1893

      code = vnodeRebuildSubmitReqMsg(pSubmitReq, &pReq);

×

1895

    if (TSDB_CODE_SUCCESS == code) {

×

1896

      pAllocMsg = pReq;

×

1898

    if (TSDB_CODE_SUCCESS != code) {

×

1899

      goto _exit;

×

UNCOV

1903

    pReq = POINTER_SHIFT(pReq, sizeof(SSubmitReq2Msg));

×

UNCOV

1904

    len -= sizeof(SSubmitReq2Msg);

×

UNCOV

1905

    SDecoder dc = {0};

×

UNCOV

1906

    tDecoderInit(&dc, pReq, len);

×

UNCOV

1907

    if (tDecodeSubmitReq(&dc, pSubmitReq, NULL) < 0) {

×

1908

      code = TSDB_CODE_INVALID_MSG;

×

1909

      goto _exit;

×

UNCOV

1911

    tDecoderClear(&dc);

×

UNCOV

1915

  TSKEY now = taosGetTimestamp(pVnode->config.tsdbCfg.precision);

×

UNCOV

1916

  TSKEY minKey = now - tsTickPerMin[pVnode->config.tsdbCfg.precision] * pVnode->config.tsdbCfg.keep2;

×

UNCOV

1917

  TSKEY maxKey = tsMaxKeyByPrecision[pVnode->config.tsdbCfg.precision];

×

UNCOV

1918

  for (int32_t i = 0; i < TARRAY_SIZE(pSubmitReq->aSubmitTbData); ++i) {

×

UNCOV

1919

    SSubmitTbData *pSubmitTbData = taosArrayGet(pSubmitReq->aSubmitTbData, i);

×

UNCOV

1921

    if (pSubmitTbData->pCreateTbReq && pSubmitTbData->pCreateTbReq->uid == 0) {

×

1922

      code = TSDB_CODE_INVALID_MSG;

×

1923

      goto _exit;

×

UNCOV

1926

    if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {

×

UNCOV

1927

      if (TARRAY_SIZE(pSubmitTbData->aCol) <= 0) {

×

1928

        code = TSDB_CODE_INVALID_MSG;

×

1929

        goto _exit;

×

UNCOV

1932

      SColData *colDataArr = TARRAY_DATA(pSubmitTbData->aCol);

×

UNCOV

1934

      tColDataArrGetRowKey(colDataArr, TARRAY_SIZE(pSubmitTbData->aCol), 0, &lastKey);

×

UNCOV

1935

      for (int32_t iRow = 1; iRow < colDataArr[0].nVal; iRow++) {

×

UNCOV

1937

        tColDataArrGetRowKey(TARRAY_DATA(pSubmitTbData->aCol), TARRAY_SIZE(pSubmitTbData->aCol), iRow, &key);

×

UNCOV

1938

        if (tRowKeyCompare(&lastKey, &key) >= 0) {

×

1939

          code = TSDB_CODE_INVALID_MSG;

×

1940

          vError("vgId:%d %s failed 1 since %s, version:%" PRId64, TD_VID(pVnode), __func__, tstrerror(terrno), ver);

×

1941

          goto _exit;

×

UNCOV

1945

      int32_t nRow = TARRAY_SIZE(pSubmitTbData->aRowP);

×

UNCOV

1946

      SRow  **aRow = (SRow **)TARRAY_DATA(pSubmitTbData->aRowP);

×

UNCOV

1948

      for (int32_t iRow = 0; iRow < nRow; ++iRow) {

×

UNCOV

1949

        if (aRow[iRow]->ts < minKey || aRow[iRow]->ts > maxKey) {

×

UNCOV

1950

          code = TSDB_CODE_INVALID_MSG;

×

UNCOV

1951

          vError("vgId:%d %s failed 2 since %s, version:%" PRId64, TD_VID(pVnode), __func__, tstrerror(code), ver);

×

1952

          goto _exit;

×

UNCOV

1954

        if (iRow == 0) {

×

UNCOV

1955

          tRowGetKey(aRow[iRow], &lastRowKey);

×

UNCOV

1958

          tRowGetKey(aRow[iRow], &rowKey);

×

UNCOV

1960

          if (tRowKeyCompare(&lastRowKey, &rowKey) >= 0) {

×

1961

            code = TSDB_CODE_INVALID_MSG;

×

1962

            vError("vgId:%d %s failed 3 since %s, version:%" PRId64, TD_VID(pVnode), __func__, tstrerror(code), ver);

×

1963

            goto _exit;

×

UNCOV

1965

          lastRowKey = rowKey;

×

UNCOV

1971

  for (int32_t i = 0; i < TARRAY_SIZE(pSubmitReq->aSubmitTbData); ++i) {

×

UNCOV

1972

    SSubmitTbData *pSubmitTbData = taosArrayGet(pSubmitReq->aSubmitTbData, i);

×

UNCOV

1974

    if (pSubmitTbData->pCreateTbReq) {

×

UNCOV

1975

      pSubmitTbData->uid = pSubmitTbData->pCreateTbReq->uid;

×

UNCOV

1977

      SMetaInfo info = {0};

×

UNCOV

1979

      code = metaGetInfo(pVnode->pMeta, pSubmitTbData->uid, &info, NULL);

×

UNCOV

1980

      if (code) {

×

1981

        code = TSDB_CODE_TDB_TABLE_NOT_EXIST;

×

1982

        vWarn("vgId:%d, table uid:%" PRId64 " not exists", TD_VID(pVnode), pSubmitTbData->uid);

×

UNCOV

1983

        goto _exit;

×

UNCOV

1986

      if (info.suid != pSubmitTbData->suid) {

×

1987

        code = TSDB_CODE_INVALID_MSG;

×

1988

        goto _exit;

×

UNCOV

1991

      if (info.suid) {

×

UNCOV

1992

        if (metaGetInfo(pVnode->pMeta, info.suid, &info, NULL) != 0) {

×

1993

          vWarn("vgId:%d, table uid:%" PRId64 " not exists", TD_VID(pVnode), info.suid);

×

UNCOV

1997

      if (pSubmitTbData->sver != info.skmVer) {

×

1998

        code = TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER;

×

1999

        goto _exit;

×

UNCOV

2003

    if (pSubmitTbData->flags & SUBMIT_REQ_COLUMN_DATA_FORMAT) {

×

UNCOV

2004

      int32_t   nColData = TARRAY_SIZE(pSubmitTbData->aCol);

×

UNCOV

2005

      SColData *aColData = (SColData *)TARRAY_DATA(pSubmitTbData->aCol);

×

UNCOV

2007

      if (nColData <= 0) {

×

2008

        code = TSDB_CODE_INVALID_MSG;

×

2009

        goto _exit;

×

UNCOV

2012

      if (aColData[0].cid != PRIMARYKEY_TIMESTAMP_COL_ID || aColData[0].type != TSDB_DATA_TYPE_TIMESTAMP ||

×

UNCOV

2013

          aColData[0].nVal <= 0) {

×

2014

        code = TSDB_CODE_INVALID_MSG;

×

2015

        goto _exit;

×

UNCOV

2018

      for (int32_t j = 1; j < nColData; j++) {

×

UNCOV

2019

        if (aColData[j].nVal != aColData[0].nVal) {

×

2020

          code = TSDB_CODE_INVALID_MSG;

×

2021

          goto _exit;

×

UNCOV

2027

  vDebug("vgId:%d, submit block size %d", TD_VID(pVnode), (int32_t)taosArrayGetSize(pSubmitReq->aSubmitTbData));

×

UNCOV

2030

  for (int32_t i = 0; i < TARRAY_SIZE(pSubmitReq->aSubmitTbData); ++i) {

×

UNCOV

2031

    SSubmitTbData *pSubmitTbData = taosArrayGet(pSubmitReq->aSubmitTbData, i);

×

UNCOV

2034

    if (pSubmitTbData->pCreateTbReq) {

×

UNCOV

2036

      if (pSubmitRsp->aCreateTbRsp == NULL &&

×

UNCOV

2037

          (pSubmitRsp->aCreateTbRsp = taosArrayInit(TARRAY_SIZE(pSubmitReq->aSubmitTbData), sizeof(SVCreateTbRsp))) ==

×

2039

        code = terrno;

×

UNCOV

2040

        goto _exit;

×

UNCOV

2043

      SVCreateTbRsp *pCreateTbRsp = taosArrayReserve(pSubmitRsp->aCreateTbRsp, 1);

×

UNCOV

2046

      if (metaCreateTable2(pVnode->pMeta, ver, pSubmitTbData->pCreateTbReq, &pCreateTbRsp->pMeta) == 0) {

×

UNCOV

2049

        if (newTbUids == NULL &&

×

UNCOV

2050

            (newTbUids = taosArrayInit(TARRAY_SIZE(pSubmitReq->aSubmitTbData), sizeof(int64_t))) == NULL) {

×

2051

          code = terrno;

×

2052

          goto _exit;

×

UNCOV

2055

        if (taosArrayPush(newTbUids, &pSubmitTbData->uid) == NULL) {

×

2056

          code = terrno;

×

2057

          goto _exit;

×

UNCOV

2060

        if (pCreateTbRsp->pMeta) {

×

UNCOV

2061

          vnodeUpdateMetaRsp(pVnode, pCreateTbRsp->pMeta);

×

UNCOV

2064

        if (terrno != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {

×

UNCOV

2065

          code = terrno;

×

UNCOV

2066

          vError("vgId:%d failed to create table:%s, code:%s", TD_VID(pVnode), pSubmitTbData->pCreateTbReq->name,

×

UNCOV

2068

          goto _exit;

×

UNCOV

2070

        terrno = 0;

×

UNCOV

2071

        pSubmitTbData->uid = pSubmitTbData->pCreateTbReq->uid;  // update uid if table exist for using below

×

UNCOV

2077

    code = tsdbInsertTableData(pVnode->pTsdb, ver, pSubmitTbData, &affectedRows);

×

UNCOV

2078

    if (code) goto _exit;

×

UNCOV

2080

    code = metaUpdateChangeTimeWithLock(pVnode->pMeta, pSubmitTbData->uid, pSubmitTbData->ctimeMs);

×

UNCOV

2081

    if (code) goto _exit;

×

UNCOV

2083

    pSubmitRsp->affectedRows += affectedRows;

×

UNCOV

2087

  if (taosArrayGetSize(newTbUids) > 0) {

×

UNCOV

2088

    vDebug("vgId:%d, add %d table into query table list in handling submit", TD_VID(pVnode),

×

UNCOV

2090

    if (tqUpdateTbUidList(pVnode->pTq, newTbUids, true) != 0) {

×

2091

      vError("vgId:%d, failed to update tbUid list", TD_VID(pVnode));

×

UNCOV

2095

_exit:

×

UNCOV

2097

  pRsp->code = code;

×

UNCOV

2098

  tEncodeSize(tEncodeSSubmitRsp2, pSubmitRsp, pRsp->contLen, ret);

×

UNCOV

2099

  pRsp->pCont = rpcMallocCont(pRsp->contLen);

×

UNCOV

2100

  tEncoderInit(&ec, pRsp->pCont, pRsp->contLen);

×

UNCOV

2101

  if (tEncodeSSubmitRsp2(&ec, pSubmitRsp) < 0) {

×

2102

    vError("vgId:%d, failed to encode submit response", TD_VID(pVnode));

×

UNCOV

2104

  tEncoderClear(&ec);

×

UNCOV

2107

  (void)atomic_add_fetch_64(&pVnode->statis.nInsert, pSubmitRsp->affectedRows);

×

UNCOV

2108

  (void)atomic_add_fetch_64(&pVnode->statis.nInsertSuccess, pSubmitRsp->affectedRows);

×

UNCOV

2109

  (void)atomic_add_fetch_64(&pVnode->statis.nBatchInsert, 1);

×

UNCOV

2111

  if (tsEnableMonitor && tsMonitorFqdn[0] != 0 && tsMonitorPort != 0 && pSubmitRsp->affectedRows > 0 &&

×

2112

      strlen(pOriginalMsg->info.conn.user) > 0 && tsInsertCounter != NULL) {

×

2113

    const char *sample_labels[] = {VNODE_METRIC_TAG_VALUE_INSERT_AFFECTED_ROWS,

×

2114

                                   pVnode->monitor.strClusterId,

×

2115

                                   pVnode->monitor.strDnodeId,

×

2117

                                   pVnode->monitor.strVgId,

×

2118

                                   pOriginalMsg->info.conn.user,

×

2120

    int         tv = taos_counter_add(tsInsertCounter, pSubmitRsp->affectedRows, sample_labels);

×

UNCOV

2123

  if (code == 0) {

×

UNCOV

2124

    (void)atomic_add_fetch_64(&pVnode->statis.nBatchInsertSuccess, 1);

×

UNCOV

2125

    code = tdProcessRSmaSubmit(pVnode->pSma, ver, pSubmitReq, pReq, len);

×

UNCOV

2146

  taosArrayDestroy(newTbUids);

×

UNCOV

2147

  tDestroySubmitReq(pSubmitReq, 0 == pMsg->version ? TSDB_MSG_FLG_CMPT : TSDB_MSG_FLG_DECODE);

×

UNCOV

2148

  tDestroySSubmitRsp2(pSubmitRsp, TSDB_MSG_FLG_ENCODE);

×

UNCOV

2150

  if (code) terrno = code;

×

UNCOV

2152

  taosMemoryFree(pAllocMsg);

×

UNCOV

2154

  return code;

×

UNCOV

2157

static int32_t vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

UNCOV

2158

  SVCreateTSmaReq req = {0};

×

UNCOV

2159

  SDecoder        coder = {0};

×

UNCOV

2161

  if (pRsp) {

×

UNCOV

2162

    pRsp->msgType = TDMT_VND_CREATE_SMA_RSP;

×

UNCOV

2163

    pRsp->code = TSDB_CODE_SUCCESS;

×

UNCOV

2164

    pRsp->pCont = NULL;

×

UNCOV

2165

    pRsp->contLen = 0;

×

UNCOV

2169

  tDecoderInit(&coder, pReq, len);

×

UNCOV

2171

  if (tDecodeSVCreateTSmaReq(&coder, &req) < 0) {

×

2172

    terrno = TSDB_CODE_MSG_DECODE_ERROR;

×

2173

    if (pRsp) pRsp->code = terrno;

×

2174

    goto _err;

×

UNCOV

2177

  if (tdProcessTSmaCreate(pVnode->pSma, ver, (const char *)&req) < 0) {

×

2178

    if (pRsp) pRsp->code = terrno;

×

2179

    goto _err;

×

UNCOV

2182

  tDecoderClear(&coder);

×

UNCOV

2183

  vDebug("vgId:%d, success to create tsma %s:%" PRIi64 " version %" PRIi64 " for table %" PRIi64, TD_VID(pVnode),

×

UNCOV

2185

  return 0;

×

2187

_err:

×

2188

  tDecoderClear(&coder);

×

2189

  vError("vgId:%d, failed to create tsma %s:%" PRIi64 " version %" PRIi64 "for table %" PRIi64 " since %s",

×

2191

  return terrno;

×

2202

int32_t vnodeProcessCreateTSma(SVnode *pVnode, void *pCont, uint32_t contLen) {

×

2203

  return vnodeProcessCreateTSmaReq(pVnode, 1, pCont, contLen, NULL);

×

UNCOV

2206

static int32_t vnodeConsolidateAlterHashRange(SVnode *pVnode, int64_t ver) {

×

UNCOV

2207

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

2209

  vInfo("vgId:%d, trim meta of tables per hash range [%" PRIu32 ", %" PRIu32 "]. apply-index:%" PRId64, TD_VID(pVnode),

×

UNCOV

2213

  code = metaTrimTables(pVnode->pMeta, ver);

×

UNCOV

2215

  return code;

×

UNCOV

2218

static int32_t vnodeProcessAlterConfirmReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

UNCOV

2219

  vInfo("vgId:%d, vnode handle msgType:alter-confirm, alter confirm msg is processed", TD_VID(pVnode));

×

UNCOV

2220

  int32_t code = TSDB_CODE_SUCCESS;

×

UNCOV

2221

  if (!pVnode->config.hashChange) {

×

UNCOV

2222

    goto _exit;

×

UNCOV

2225

  code = vnodeConsolidateAlterHashRange(pVnode, ver);

×

UNCOV

2226

  if (code < 0) {

×

2227

    vError("vgId:%d, failed to consolidate alter hashrange since %s. version:%" PRId64, TD_VID(pVnode), terrstr(), ver);

×

2228

    goto _exit;

×

UNCOV

2230

  pVnode->config.hashChange = false;

×

UNCOV

2232

_exit:

×

UNCOV

2233

  pRsp->msgType = TDMT_VND_ALTER_CONFIRM_RSP;

×

UNCOV

2234

  pRsp->code = code;

×

UNCOV

2235

  pRsp->pCont = NULL;

×

UNCOV

2236

  pRsp->contLen = 0;

×

UNCOV

2238

  return code;

×

UNCOV

2244

static int32_t vnodeProcessAlterConfigReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

UNCOV

2245

  bool walChanged = false;

×

UNCOV

2246

  bool tsdbChanged = false;

×

UNCOV

2248

  SAlterVnodeConfigReq req = {0};

×

UNCOV

2249

  if (tDeserializeSAlterVnodeConfigReq(pReq, len, &req) != 0) {

×

2250

    terrno = TSDB_CODE_INVALID_MSG;

×

2251

    return TSDB_CODE_INVALID_MSG;

×

UNCOV

2254

  vInfo("vgId:%d, start to alter vnode config, page:%d pageSize:%d buffer:%d szPage:%d szBuf:%" PRIu64

×

UNCOV

2263

  if (pVnode->config.cacheLastSize != req.cacheLastSize) {

×

UNCOV

2264

    pVnode->config.cacheLastSize = req.cacheLastSize;

×

UNCOV

2265

    tsdbCacheSetCapacity(pVnode, (size_t)pVnode->config.cacheLastSize * 1024 * 1024);

×

UNCOV

2268

  if (pVnode->config.szBuf != req.buffer * 1024LL * 1024LL) {

×

2269

    vInfo("vgId:%d, vnode buffer is changed from %" PRId64 " to %" PRId64, TD_VID(pVnode), pVnode->config.szBuf,

×

2271

    pVnode->config.szBuf = req.buffer * 1024LL * 1024LL;

×

UNCOV

2274

  if (pVnode->config.szCache != req.pages) {

×

2275

    if (metaAlterCache(pVnode->pMeta, req.pages) < 0) {

×

2276

      vError("vgId:%d, failed to change vnode pages from %d to %d failed since %s", TD_VID(pVnode),

×

2278

      return errno;

×

2280

      vInfo("vgId:%d, vnode pages is changed from %d to %d", TD_VID(pVnode), pVnode->config.szCache, req.pages);

×

2281

      pVnode->config.szCache = req.pages;

×

UNCOV

2285

  if (pVnode->config.cacheLast != req.cacheLast) {

×

UNCOV

2286

    pVnode->config.cacheLast = req.cacheLast;

×

UNCOV

2289

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

×

UNCOV

2290

    pVnode->config.walCfg.fsyncPeriod = req.walFsyncPeriod;

×

UNCOV

2291

    walChanged = true;

×

UNCOV

2294

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

×

UNCOV

2295

    if (pVnode->config.walCfg.level == 0) {

×

2296

      pVnode->config.walCfg.clearFiles = 1;

×

UNCOV

2298

    pVnode->config.walCfg.level = req.walLevel;

×

UNCOV

2299

    walChanged = true;

×

UNCOV

2302

  if (pVnode->config.walCfg.retentionPeriod != req.walRetentionPeriod) {

×

UNCOV

2303

    pVnode->config.walCfg.retentionPeriod = req.walRetentionPeriod;

×

UNCOV

2304

    walChanged = true;

×

UNCOV

2307

  if (pVnode->config.walCfg.retentionSize != req.walRetentionSize) {

×

2308

    pVnode->config.walCfg.retentionSize = req.walRetentionSize;

×

2309

    walChanged = true;

×

UNCOV

2312

  if (pVnode->config.tsdbCfg.keep0 != req.daysToKeep0) {

×

UNCOV

2313

    pVnode->config.tsdbCfg.keep0 = req.daysToKeep0;

×

UNCOV

2314

    if (!VND_IS_RSMA(pVnode)) {

×

UNCOV

2315

      tsdbChanged = true;

×

UNCOV

2319

  if (pVnode->config.tsdbCfg.keep1 != req.daysToKeep1) {

×

UNCOV

2320

    pVnode->config.tsdbCfg.keep1 = req.daysToKeep1;

×

UNCOV

2321

    if (!VND_IS_RSMA(pVnode)) {

×

UNCOV

2322

      tsdbChanged = true;

×

UNCOV

2326

  if (pVnode->config.tsdbCfg.keep2 != req.daysToKeep2) {

×

UNCOV

2327

    pVnode->config.tsdbCfg.keep2 = req.daysToKeep2;

×

UNCOV

2328

    if (!VND_IS_RSMA(pVnode)) {

×

UNCOV

2329

      tsdbChanged = true;

×

UNCOV

2333

  if (pVnode->config.tsdbCfg.keepTimeOffset != req.keepTimeOffset) {

×

UNCOV

2334

    pVnode->config.tsdbCfg.keepTimeOffset = req.keepTimeOffset;

×

UNCOV

2335

    if (!VND_IS_RSMA(pVnode)) {

×

UNCOV

2336

      tsdbChanged = true;

×

UNCOV

2340

  if (req.sttTrigger != -1 && req.sttTrigger != pVnode->config.sttTrigger) {

×

2341

    if (req.sttTrigger > 1 && pVnode->config.sttTrigger > 1) {

×

2342

      pVnode->config.sttTrigger = req.sttTrigger;

×

2344

      vnodeAWait(&pVnode->commitTask);

×

2346

      int32_t ret = tsdbDisableAndCancelAllBgTask(pVnode->pTsdb);

×

2347

      if (ret != 0) {

×

2348

        vError("vgId:%d, failed to disable bg task since %s", TD_VID(pVnode), tstrerror(errno));

×

2351

      pVnode->config.sttTrigger = req.sttTrigger;

×

2352

      tsdbEnableBgTask(pVnode->pTsdb);

×

UNCOV

2356

  if (req.minRows != -1 && req.minRows != pVnode->config.tsdbCfg.minRows) {

×

2357

    pVnode->config.tsdbCfg.minRows = req.minRows;

×

UNCOV

2360

  if (req.s3KeepLocal != -1 && req.s3KeepLocal != pVnode->config.s3KeepLocal) {

×

2361

    pVnode->config.s3KeepLocal = req.s3KeepLocal;

×

UNCOV

2363

  if (req.s3Compact != -1 && req.s3Compact != pVnode->config.s3Compact) {

×

2364

    pVnode->config.s3Compact = req.s3Compact;

×

UNCOV

2367

  if (walChanged) {

×

UNCOV

2368

    if (walAlter(pVnode->pWal, &pVnode->config.walCfg) != 0) {

×

2369

      vError("vgId:%d, failed to alter wal config since %s", TD_VID(pVnode), tstrerror(errno));

×

UNCOV

2373

  if (tsdbChanged) {

×

UNCOV

2374

    tsdbSetKeepCfg(pVnode->pTsdb, &pVnode->config.tsdbCfg);

×

UNCOV

2377

  return 0;

×

UNCOV

2380

static int32_t vnodeProcessBatchDeleteReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

UNCOV

2383

  tDecoderInit(&decoder, pReq, len);

×

UNCOV

2384

  if (tDecodeSBatchDeleteReq(&decoder, &deleteReq) < 0) {

×

2385

    tDecoderClear(&decoder);

×

2386

    return terrno = TSDB_CODE_INVALID_MSG;

×

UNCOV

2389

  SMetaReader mr = {0};

×

UNCOV

2390

  metaReaderDoInit(&mr, pVnode->pMeta, META_READER_NOLOCK);

×

UNCOV

2391

  STsdb *pTsdb = pVnode->pTsdb;

×

UNCOV

2393

  if (deleteReq.level) {

×

UNCOV

2394

    pTsdb = deleteReq.level == 1 ? VND_RSMA1(pVnode) : VND_RSMA2(pVnode);

×

UNCOV

2397

  int32_t sz = taosArrayGetSize(deleteReq.deleteReqs);

×

UNCOV

2398

  for (int32_t i = 0; i < sz; i++) {

×

UNCOV

2399

    SSingleDeleteReq *pOneReq = taosArrayGet(deleteReq.deleteReqs, i);

×

UNCOV

2400

    char             *name = pOneReq->tbname;

×

UNCOV

2401

    if (metaGetTableEntryByName(&mr, name) < 0) {

×

UNCOV

2402

      vDebug("vgId:%d, stream delete msg, skip since no table: %s", pVnode->config.vgId, name);

×

UNCOV

2403

      continue;

×

UNCOV

2406

    int64_t uid = mr.me.uid;

×

UNCOV

2408

    int32_t code = tsdbDeleteTableData(pTsdb, ver, deleteReq.suid, uid, pOneReq->startTs, pOneReq->endTs);

×

UNCOV

2409

    if (code < 0) {

×

2410

      terrno = code;

×

2411

      vError("vgId:%d, delete error since %s, suid:%" PRId64 ", uid:%" PRId64 ", start ts:%" PRId64 ", end ts:%" PRId64,

×

UNCOV

2415

    if (deleteReq.level == 0) {

×

UNCOV

2416

      code = metaUpdateChangeTimeWithLock(pVnode->pMeta, uid, deleteReq.ctimeMs);

×

UNCOV

2417

      if (code < 0) {

×

2418

        terrno = code;

×

2419

        vError("vgId:%d, update change time error since %s, suid:%" PRId64 ", uid:%" PRId64 ", start ts:%" PRId64

×

UNCOV

2424

    tDecoderClear(&mr.coder);

×

UNCOV

2426

  metaReaderClear(&mr);

×

UNCOV

2427

  taosArrayDestroy(deleteReq.deleteReqs);

×

UNCOV

2428

  return 0;

×

UNCOV

2431

static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp,

×

UNCOV

2433

  int32_t     code = 0;

×

UNCOV

2434

  SDecoder   *pCoder = &(SDecoder){0};

×

UNCOV

2435

  SDeleteRes *pRes = &(SDeleteRes){0};

×

UNCOV

2437

  pRsp->msgType = TDMT_VND_DELETE_RSP;

×

UNCOV

2438

  pRsp->pCont = NULL;

×

UNCOV

2439

  pRsp->contLen = 0;

×

UNCOV

2440

  pRsp->code = TSDB_CODE_SUCCESS;

×

UNCOV

2442

  pRes->uidList = taosArrayInit(0, sizeof(tb_uid_t));

×

UNCOV

2443

  if (pRes->uidList == NULL) {

×

2444

    code = terrno;

×

2445

    goto _err;

×

UNCOV

2448

  tDecoderInit(pCoder, pReq, len);

×

UNCOV

2449

  code = tDecodeDeleteRes(pCoder, pRes);

×

UNCOV

2450

  if (code) goto _err;

×

UNCOV

2452

  if (pRes->affectedRows > 0) {

×

UNCOV

2453

    for (int32_t iUid = 0; iUid < taosArrayGetSize(pRes->uidList); iUid++) {

×

UNCOV

2454

      uint64_t uid = *(uint64_t *)taosArrayGet(pRes->uidList, iUid);

×

UNCOV

2455

      code = tsdbDeleteTableData(pVnode->pTsdb, ver, pRes->suid, uid, pRes->skey, pRes->ekey);

×

UNCOV

2456

      if (code) goto _err;

×

UNCOV

2457

      code = metaUpdateChangeTimeWithLock(pVnode->pMeta, uid, pRes->ctimeMs);

×

UNCOV

2458

      if (code) goto _err;

×

UNCOV

2462

  code = tdProcessRSmaDelete(pVnode->pSma, ver, pRes, pReq, len);

×

UNCOV

2464

  tDecoderClear(pCoder);

×

UNCOV

2465

  taosArrayDestroy(pRes->uidList);

×

UNCOV

2467

  SVDeleteRsp rsp = {.affectedRows = pRes->affectedRows};

×

UNCOV

2468

  int32_t     ret = 0;

×

UNCOV

2469

  tEncodeSize(tEncodeSVDeleteRsp, &rsp, pRsp->contLen, ret);

×

UNCOV

2470

  pRsp->pCont = rpcMallocCont(pRsp->contLen);

×

UNCOV

2471

  SEncoder ec = {0};

×

UNCOV

2472

  tEncoderInit(&ec, pRsp->pCont, pRsp->contLen);

×

UNCOV

2473

  code = tEncodeSVDeleteRsp(&ec, &rsp);

×

UNCOV

2474

  if (code) goto _err;

×

UNCOV

2475

  tEncoderClear(&ec);

×

UNCOV

2476

  return code;

×

2478

_err:

×

2494

  return code;

×

UNCOV

2496

static int32_t vnodeProcessCreateIndexReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

UNCOV

2497

  SVCreateStbReq req = {0};

×

UNCOV

2498

  SDecoder       dc = {0};

×

UNCOV

2499

  int32_t        code = 0;

×

UNCOV

2501

  pRsp->msgType = TDMT_VND_CREATE_INDEX_RSP;

×

UNCOV

2502

  pRsp->code = TSDB_CODE_SUCCESS;

×

UNCOV

2503

  pRsp->pCont = NULL;

×

UNCOV

2504

  pRsp->contLen = 0;

×

UNCOV

2506

  tDecoderInit(&dc, pReq, len);

×

UNCOV

2508

  if (tDecodeSVCreateStbReq(&dc, &req) < 0) {

×

2509

    tDecoderClear(&dc);

×

2510

    return terrno = TSDB_CODE_INVALID_MSG;

×

UNCOV

2513

  code = metaAddIndexToSuperTable(pVnode->pMeta, ver, &req);

×

UNCOV

2514

  if (code) {

×

2515

    pRsp->code = code;

×

2516

    goto _err;

×

UNCOV

2518

  tDecoderClear(&dc);

×

UNCOV

2519

  return 0;

×

2521

_err:

×

2522

  tDecoderClear(&dc);

×

2523

  return code;

×

UNCOV

2525

static int32_t vnodeProcessDropIndexReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

UNCOV

2526

  SDropIndexReq req = {0};

×

UNCOV

2527

  int32_t       code = 0;

×

UNCOV

2528

  pRsp->msgType = TDMT_VND_DROP_INDEX_RSP;

×

UNCOV

2529

  pRsp->code = TSDB_CODE_SUCCESS;

×

UNCOV

2530

  pRsp->pCont = NULL;

×

UNCOV

2531

  pRsp->contLen = 0;

×

UNCOV

2533

  if ((code = tDeserializeSDropIdxReq(pReq, len, &req))) {

×

2534

    pRsp->code = code;

×

2535

    return code;

×

UNCOV

2538

  code = metaDropIndexFromSuperTable(pVnode->pMeta, ver, &req);

×

UNCOV

2539

  if (code) {

×

2540

    pRsp->code = code;

×

2541

    return code;

×

UNCOV

2543

  return TSDB_CODE_SUCCESS;

×

UNCOV

2548

static int32_t vnodeProcessCompactVnodeReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

UNCOV

2549

  if (!pVnode->restored) {

×

UNCOV

2550

    vInfo("vgId:%d, ignore compact req during restoring. ver:%" PRId64, TD_VID(pVnode), ver);

×

UNCOV

2551

    return 0;

×

UNCOV

2553

  return vnodeAsyncCompact(pVnode, ver, pReq, len, pRsp);

×

2556

static int32_t vnodeProcessConfigChangeReq(SVnode *pVnode, int64_t ver, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

2557

  if (syncCheckMember(pVnode->sync) != 0) {

×

2558

    vError("vgId:%d, failed to check member", TD_VID(pVnode));

×

2561

  pRsp->msgType = TDMT_SYNC_CONFIG_CHANGE_RSP;

×

2562

  pRsp->code = TSDB_CODE_SUCCESS;

×

2563

  pRsp->pCont = NULL;

×

2564

  pRsp->contLen = 0;

×

2566

  return 0;

×

UNCOV

2569

static int32_t vnodeCheckToken(SVnode *pVnode, char *member0Token, char *member1Token) {

×

UNCOV

2570

  SSyncState syncState = syncGetState(pVnode->sync);

×

UNCOV

2571

  if (syncState.state != TAOS_SYNC_STATE_LEADER) {

×

UNCOV

2572

    return terrno = TSDB_CODE_SYN_NOT_LEADER;

×

UNCOV

2575

  char token[TSDB_ARB_TOKEN_SIZE] = {0};

×

UNCOV

2576

  if (vnodeGetArbToken(pVnode, token) != 0) {

×

2577

    return terrno = TSDB_CODE_NOT_FOUND;

×

UNCOV

2580

  if (strncmp(token, member0Token, TSDB_ARB_TOKEN_SIZE) != 0 &&

×

UNCOV

2581

      strncmp(token, member1Token, TSDB_ARB_TOKEN_SIZE) != 0) {

×

2582

    return terrno = TSDB_CODE_MND_ARB_TOKEN_MISMATCH;

×

UNCOV

2585

  terrno = TSDB_CODE_SUCCESS;

×

UNCOV

2586

  return 0;

×

UNCOV

2589

static int32_t vnodeCheckSyncd(SVnode *pVnode, char *member0Token, char *member1Token) {

×

UNCOV

2590

  int32_t code = vnodeCheckToken(pVnode, member0Token, member1Token);

×

UNCOV

2591

  if (code != 0) {

×

UNCOV

2592

    return code;

×

UNCOV

2595

  return syncCheckSynced(pVnode->sync);

×

UNCOV

2598

static int32_t vnodeProcessArbCheckSyncReq(SVnode *pVnode, void *pReq, int32_t len, SRpcMsg *pRsp) {

×

UNCOV

2599

  int32_t code = 0;

×

UNCOV

2601

  SVArbCheckSyncReq syncReq = {0};

×

UNCOV

2603

  code = tDeserializeSVArbCheckSyncReq(pReq, len, &syncReq);

×

UNCOV

2604

  if (code) {

×

2605

    return terrno = code;

×

UNCOV

2608

  pRsp->msgType = TDMT_VND_ARB_CHECK_SYNC_RSP;

×

UNCOV

2609

  pRsp->code = TSDB_CODE_SUCCESS;

×

UNCOV

2610

  pRsp->pCont = NULL;

×

UNCOV

2611

  pRsp->contLen = 0;

×

UNCOV

2613

  SVArbCheckSyncRsp syncRsp = {0};

×

UNCOV

2614

  syncRsp.arbToken = syncReq.arbToken;

×

UNCOV

2615

  syncRsp.member0Token = syncReq.member0Token;

×

UNCOV

2616

  syncRsp.member1Token = syncReq.member1Token;

×

UNCOV

2617

  syncRsp.vgId = TD_VID(pVnode);

×

UNCOV

2619

  if (vnodeCheckSyncd(pVnode, syncReq.member0Token, syncReq.member1Token) != 0) {

×

UNCOV

2620

    vError("vgId:%d, failed to check assigned log syncd", TD_VID(pVnode));

×

UNCOV

2622

  syncRsp.errCode = terrno;

×

UNCOV

2624

  if (vnodeUpdateArbTerm(pVnode, syncReq.arbTerm) != 0) {

×

2625

    vError("vgId:%d, failed to update arb term", TD_VID(pVnode));

×

2626

    code = -1;

×

2627

    goto _OVER;

×

UNCOV

2630

  int32_t contLen = tSerializeSVArbCheckSyncRsp(NULL, 0, &syncRsp);

×

UNCOV

2631

  if (contLen <= 0) {

×

2632

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

2633

    code = -1;

×

2634

    goto _OVER;

×

UNCOV

2636

  void *pHead = rpcMallocCont(contLen);

×

UNCOV

2637

  if (!pHead) {

×

2638

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

2639

    code = -1;

×

2640

    goto _OVER;

×

UNCOV

2643

  if (tSerializeSVArbCheckSyncRsp(pHead, contLen, &syncRsp) <= 0) {

×

2644

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

2645

    rpcFreeCont(pHead);

×

2646

    code = -1;

×

2647

    goto _OVER;

×

UNCOV

2650

  pRsp->pCont = pHead;

×

UNCOV

2651

  pRsp->contLen = contLen;

×

UNCOV

2653

  terrno = TSDB_CODE_SUCCESS;

×

UNCOV

2655

_OVER:

×

UNCOV

2656

  tFreeSVArbCheckSyncReq(&syncReq);

×

UNCOV

2657

  return code;

×

taosdata / TDengine / #3621

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

taosdata / TDengine / #3621

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

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