taosdata / TDengine / #3621

coverage: 2.037% (-61.5%) from 63.573%
#3621

push

travis-ci

web-flow 
Merge pull request #29874 from taosdata/merge/mainto3.0

merge: from main to 3.0 branch

4357 of 287032 branches covered (1.52%)

Branch coverage included in aggregate %.

0 of 174 new or added lines in 18 files covered. (0.0%)

213359 existing lines in 469 files now uncovered.

7260 of 283369 relevant lines covered (2.56%)

23737.72 hits per line

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

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









16



#include "tsdb.h"










17



#include "vnd.h"










18












19



#define VNODE_GET_LOAD_RESET_VALS(pVar, oVal, vType, tags)                                                    \










20



  do {                                                                                                        \










21



    int##vType##_t newVal = atomic_sub_fetch_##vType(&(pVar), (oVal));                                        \










22



    if (newVal < 0) {                                                                                         \










23



      vWarn("vgId:%d, %s, abnormal val:%" PRIi64 ", old val:%" PRIi64, TD_VID(pVnode), tags, newVal, (oVal)); \










24



    }                                                                                                         \










25



  } while (0)










26













UNCOV

27



int vnodeQueryOpen(SVnode *pVnode) {




×








UNCOV

28



  return qWorkerInit(NODE_TYPE_VNODE, TD_VID(pVnode), (void **)&pVnode->pQuery, &pVnode->msgCb);




×







29



}










30













UNCOV

31



void vnodeQueryPreClose(SVnode *pVnode) { qWorkerStopAllTasks((void *)pVnode->pQuery); }




×







32













UNCOV

33



void vnodeQueryClose(SVnode *pVnode) { qWorkerDestroy((void **)&pVnode->pQuery); }




×







34













UNCOV

35



int32_t fillTableColCmpr(SMetaReader *reader, SSchemaExt *pExt, int32_t numOfCol) {




×








UNCOV

36



  int8_t tblType = reader->me.type;




×








UNCOV

37



  if (useCompress(tblType)) {





×








UNCOV

38



    SColCmprWrapper *p = &(reader->me.colCmpr);




×








UNCOV

39



    if (numOfCol != p->nCols) {





×







40



      vError("fillTableColCmpr table type:%d, col num:%d, col cmpr num:%d mismatch", tblType, numOfCol, p->nCols);





×







41



      return TSDB_CODE_APP_ERROR;




×







42



    }











UNCOV

43



    for (int i = 0; i < p->nCols; i++) {





×








UNCOV

44



      SColCmpr *pCmpr = &p->pColCmpr[i];




×








UNCOV

45



      pExt[i].colId = pCmpr->id;




×








UNCOV

46



      pExt[i].compress = pCmpr->alg;




×







47



    }










48



  }











UNCOV

49



  return 0;




×







50



}










51













UNCOV

52



void vnodePrintTableMeta(STableMetaRsp* pMeta) {




×








UNCOV

53



  if (!(qDebugFlag & DEBUG_DEBUG)) {





×








UNCOV

54



    return;




×







55



  }










56













UNCOV

57



  qDebug("tbName:%s", pMeta->tbName);





×








UNCOV

58



  qDebug("stbName:%s", pMeta->stbName);





×








UNCOV

59



  qDebug("dbFName:%s", pMeta->dbFName);





×








UNCOV

60



  qDebug("dbId:%" PRId64, pMeta->dbId);





×








UNCOV

61



  qDebug("numOfTags:%d", pMeta->numOfTags);





×








UNCOV

62



  qDebug("numOfColumns:%d", pMeta->numOfColumns);





×








UNCOV

63



  qDebug("precision:%d", pMeta->precision);





×








UNCOV

64



  qDebug("tableType:%d", pMeta->tableType);





×








UNCOV

65



  qDebug("sversion:%d", pMeta->sversion);





×








UNCOV

66



  qDebug("tversion:%d", pMeta->tversion);





×








UNCOV

67



  qDebug("suid:%" PRIu64, pMeta->suid);





×








UNCOV

68



  qDebug("tuid:%" PRIu64, pMeta->tuid);





×








UNCOV

69



  qDebug("vgId:%d", pMeta->vgId);





×








UNCOV

70



  qDebug("sysInfo:%d", pMeta->sysInfo);





×








UNCOV

71



  if (pMeta->pSchemas) {





×








UNCOV

72



    for (int32_t i = 0; i < (pMeta->numOfColumns + pMeta->numOfTags); ++i) {





×








UNCOV

73



      SSchema* pSchema = pMeta->pSchemas + i;




×








UNCOV

74



      qDebug("%d col/tag: type:%d, flags:%d, colId:%d, bytes:%d, name:%s", i, pSchema->type, pSchema->flags, pSchema->colId, pSchema->bytes, pSchema->name);





×







75



    }










76



  }










77












78



}










79












80













UNCOV

81



int32_t vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {




×








UNCOV

82



  STableInfoReq  infoReq = {0};




×








UNCOV

83



  STableMetaRsp  metaRsp = {0};




×








UNCOV

84



  SMetaReader    mer1 = {0};




×








UNCOV

85



  SMetaReader    mer2 = {0};




×







86



  char           tableFName[TSDB_TABLE_FNAME_LEN];











UNCOV

87



  bool           reqTbUid = false;




×








UNCOV

88



  SRpcMsg        rpcMsg = {0};




×








UNCOV

89



  int32_t        code = 0;




×








UNCOV

90



  int32_t        rspLen = 0;




×








UNCOV

91



  void          *pRsp = NULL;




×








UNCOV

92



  SSchemaWrapper schema = {0};




×








UNCOV

93



  SSchemaWrapper schemaTag = {0};




×







94












95



  // decode req











UNCOV

96



  if (tDeserializeSTableInfoReq(pMsg->pCont, pMsg->contLen, &infoReq) != 0) {





×







97



    code = terrno;




×







98



    goto _exit4;




×







99



  }










100













UNCOV

101



  if (infoReq.option == REQ_OPT_TBUID) reqTbUid = true;





×








UNCOV

102



  metaRsp.dbId = pVnode->config.dbId;




×








UNCOV

103



  tstrncpy(metaRsp.tbName, infoReq.tbName, TSDB_TABLE_NAME_LEN);




×








UNCOV

104



  (void)memcpy(metaRsp.dbFName, infoReq.dbFName, sizeof(metaRsp.dbFName));




×







105













UNCOV

106



  if (!reqTbUid) {





×








UNCOV

107



    (void)tsnprintf(tableFName, TSDB_TABLE_FNAME_LEN, "%s.%s", infoReq.dbFName, infoReq.tbName);




×








UNCOV

108



    code = vnodeValidateTableHash(pVnode, tableFName);




×








UNCOV

109



    if (code) {





×







110



      goto _exit4;




×







111



    }










112



  }










113












114



  // query meta











UNCOV

115



  metaReaderDoInit(&mer1, pVnode->pMeta, META_READER_LOCK);




×








UNCOV

116



  if (reqTbUid) {





×








UNCOV

117



    errno = 0;




×








UNCOV

118



    uint64_t tbUid = taosStr2UInt64(infoReq.tbName, NULL, 10);




×








UNCOV

119



    if (errno == ERANGE || tbUid == 0) {





×







120



      code = TSDB_CODE_TDB_TABLE_NOT_EXIST;




×








UNCOV

121



      goto _exit3;




×







122



    }











UNCOV

123



    SMetaReader mr3 = {0};




×








UNCOV

124



    metaReaderDoInit(&mr3, ((SVnode *)pVnode)->pMeta, META_READER_NOLOCK);




×








UNCOV

125



    if ((code = metaReaderGetTableEntryByUid(&mr3, tbUid)) < 0) {





×








UNCOV

126



      metaReaderClear(&mr3);




×








UNCOV

127



      TAOS_CHECK_GOTO(code, NULL, _exit3);





×







128



    }











UNCOV

129



    tstrncpy(metaRsp.tbName, mr3.me.name, TSDB_TABLE_NAME_LEN);




×








UNCOV

130



    metaReaderClear(&mr3);




×








UNCOV

131



    TAOS_CHECK_GOTO(metaGetTableEntryByName(&mer1, metaRsp.tbName), NULL, _exit3);





×








UNCOV

132



  } else if (metaGetTableEntryByName(&mer1, infoReq.tbName) < 0) {





×








UNCOV

133



    code = terrno;




×








UNCOV

134



    goto _exit3;




×







135



  }










136













UNCOV

137



  metaRsp.tableType = mer1.me.type;




×








UNCOV

138



  metaRsp.vgId = TD_VID(pVnode);




×








UNCOV

139



  metaRsp.tuid = mer1.me.uid;




×







140













UNCOV

141



  if (mer1.me.type == TSDB_SUPER_TABLE) {





×








UNCOV

142



    tstrncpy(metaRsp.stbName, mer1.me.name, TSDB_TABLE_NAME_LEN);




×








UNCOV

143



    schema = mer1.me.stbEntry.schemaRow;




×








UNCOV

144



    schemaTag = mer1.me.stbEntry.schemaTag;




×








UNCOV

145



    metaRsp.suid = mer1.me.uid;




×








UNCOV

146



  } else if (mer1.me.type == TSDB_CHILD_TABLE) {





×








UNCOV

147



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




×








UNCOV

148



    if (metaReaderGetTableEntryByUid(&mer2, mer1.me.ctbEntry.suid) < 0) goto _exit2;





×







149













UNCOV

150



    tstrncpy(metaRsp.stbName, mer2.me.name, TSDB_TABLE_NAME_LEN);




×








UNCOV

151



    metaRsp.suid = mer2.me.uid;




×








UNCOV

152



    schema = mer2.me.stbEntry.schemaRow;




×








UNCOV

153



    schemaTag = mer2.me.stbEntry.schemaTag;




×








UNCOV

154



  } else if (mer1.me.type == TSDB_NORMAL_TABLE) {





×








UNCOV

155



    schema = mer1.me.ntbEntry.schemaRow;




×







156



  } else {










157



    vError("vnodeGetTableMeta get invalid table type:%d", mer1.me.type);





×







158



    goto _exit3;




×







159



  }










160













UNCOV

161



  metaRsp.numOfTags = schemaTag.nCols;




×








UNCOV

162



  metaRsp.numOfColumns = schema.nCols;




×








UNCOV

163



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




×








UNCOV

164



  metaRsp.sversion = schema.version;




×








UNCOV

165



  metaRsp.tversion = schemaTag.version;




×








UNCOV

166



  metaRsp.pSchemas = (SSchema *)taosMemoryMalloc(sizeof(SSchema) * (metaRsp.numOfColumns + metaRsp.numOfTags));





×








UNCOV

167



  metaRsp.pSchemaExt = (SSchemaExt *)taosMemoryCalloc(metaRsp.numOfColumns, sizeof(SSchemaExt));





×








UNCOV

168



  if (NULL == metaRsp.pSchemas || NULL == metaRsp.pSchemaExt) {





×







169



    code = terrno;




×







170



    goto _exit;




×







171



  }











UNCOV

172



  (void)memcpy(metaRsp.pSchemas, schema.pSchema, sizeof(SSchema) * schema.nCols);




×








UNCOV

173



  if (schemaTag.nCols) {





×








UNCOV

174



    (void)memcpy(metaRsp.pSchemas + schema.nCols, schemaTag.pSchema, sizeof(SSchema) * schemaTag.nCols);




×







175



  }











UNCOV

176



  if (metaRsp.pSchemaExt) {





×








UNCOV

177



    SMetaReader *pReader = mer1.me.type == TSDB_CHILD_TABLE ? &mer2 : &mer1;





×








UNCOV

178



    code = fillTableColCmpr(pReader, metaRsp.pSchemaExt, metaRsp.numOfColumns);




×








UNCOV

179



    if (code < 0) {





×







180



      goto _exit;




×







181



    }










182



  } else {











UNCOV

183



    code = TSDB_CODE_OUT_OF_MEMORY;




×








UNCOV

184



    goto _exit;




×







185



  }










186













UNCOV

187



  vnodePrintTableMeta(&metaRsp);




×







188












189



  // encode and send response











UNCOV

190



  rspLen = tSerializeSTableMetaRsp(NULL, 0, &metaRsp);




×








UNCOV

191



  if (rspLen < 0) {





×







192



    code = terrno;




×







193



    goto _exit;




×







194



  }










195













UNCOV

196



  if (direct) {





×








UNCOV

197



    pRsp = rpcMallocCont(rspLen);




×







198



  } else {











UNCOV

199



    pRsp = taosMemoryCalloc(1, rspLen);





×







200



  }










201













UNCOV

202



  if (pRsp == NULL) {





×







203



    code = terrno;




×







204



    goto _exit;




×







205



  }










206













UNCOV

207



  rspLen = tSerializeSTableMetaRsp(pRsp, rspLen, &metaRsp);




×








UNCOV

208



  if (rspLen < 0) {





×







209



    code = terrno;




×







210



    goto _exit;




×







211



  }










212













UNCOV

213



_exit:




×








UNCOV

214



  taosMemoryFree(metaRsp.pSchemas);





×








UNCOV

215



  taosMemoryFree(metaRsp.pSchemaExt);





×








UNCOV

216



_exit2:




×








UNCOV

217



  metaReaderClear(&mer2);




×








UNCOV

218



_exit3:




×








UNCOV

219



  metaReaderClear(&mer1);




×








UNCOV

220



_exit4:




×








UNCOV

221



  rpcMsg.info = pMsg->info;




×








UNCOV

222



  rpcMsg.pCont = pRsp;




×








UNCOV

223



  rpcMsg.contLen = rspLen;




×








UNCOV

224



  rpcMsg.code = code;




×








UNCOV

225



  rpcMsg.msgType = pMsg->msgType;




×







226













UNCOV

227



  if (code) {





×








UNCOV

228



    qError("get table %s meta with %" PRIu8 " failed cause of %s", infoReq.tbName, infoReq.option, tstrerror(code));





×







229



  }










230













UNCOV

231



  if (direct) {





×








UNCOV

232



    tmsgSendRsp(&rpcMsg);




×







233



  } else {











UNCOV

234



    *pMsg = rpcMsg;




×







235



  }










236













UNCOV

237



  return code;




×







238



}










239













UNCOV

240



int32_t vnodeGetTableCfg(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {




×








UNCOV

241



  STableCfgReq   cfgReq = {0};




×








UNCOV

242



  STableCfgRsp   cfgRsp = {0};




×








UNCOV

243



  SMetaReader    mer1 = {0};




×








UNCOV

244



  SMetaReader    mer2 = {0};




×







245



  char           tableFName[TSDB_TABLE_FNAME_LEN];











UNCOV

246



  SRpcMsg        rpcMsg = {0};




×








UNCOV

247



  int32_t        code = 0;




×








UNCOV

248



  int32_t        rspLen = 0;




×








UNCOV

249



  void          *pRsp = NULL;




×








UNCOV

250



  SSchemaWrapper schema = {0};




×








UNCOV

251



  SSchemaWrapper schemaTag = {0};




×







252












253



  // decode req











UNCOV

254



  if (tDeserializeSTableCfgReq(pMsg->pCont, pMsg->contLen, &cfgReq) != 0) {





×







255



    code = terrno;




×







256



    goto _exit;




×







257



  }










258













UNCOV

259



  tstrncpy(cfgRsp.tbName, cfgReq.tbName, TSDB_TABLE_NAME_LEN);




×








UNCOV

260



  (void)memcpy(cfgRsp.dbFName, cfgReq.dbFName, sizeof(cfgRsp.dbFName));




×







261













UNCOV

262



  (void)tsnprintf(tableFName, TSDB_TABLE_FNAME_LEN, "%s.%s", cfgReq.dbFName, cfgReq.tbName);




×








UNCOV

263



  code = vnodeValidateTableHash(pVnode, tableFName);




×








UNCOV

264



  if (code) {





×







265



    goto _exit;




×







266



  }










267












268



  // query meta











UNCOV

269



  metaReaderDoInit(&mer1, pVnode->pMeta, META_READER_LOCK);




×







270













UNCOV

271



  if (metaGetTableEntryByName(&mer1, cfgReq.tbName) < 0) {





×







272



    code = terrno;




×







273



    goto _exit;




×







274



  }










275













UNCOV

276



  cfgRsp.tableType = mer1.me.type;




×







277













UNCOV

278



  if (mer1.me.type == TSDB_SUPER_TABLE) {





×







279



    code = TSDB_CODE_VND_HASH_MISMATCH;




×







280



    goto _exit;




×








UNCOV

281



  } else if (mer1.me.type == TSDB_CHILD_TABLE) {





×








UNCOV

282



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




×








UNCOV

283



    if (metaReaderGetTableEntryByUid(&mer2, mer1.me.ctbEntry.suid) < 0) goto _exit;





×







284













UNCOV

285



    tstrncpy(cfgRsp.stbName, mer2.me.name, TSDB_TABLE_NAME_LEN);




×








UNCOV

286



    schema = mer2.me.stbEntry.schemaRow;




×








UNCOV

287



    schemaTag = mer2.me.stbEntry.schemaTag;




×








UNCOV

288



    cfgRsp.ttl = mer1.me.ctbEntry.ttlDays;




×








UNCOV

289



    cfgRsp.commentLen = mer1.me.ctbEntry.commentLen;




×








UNCOV

290



    if (mer1.me.ctbEntry.commentLen > 0) {





×







291



      cfgRsp.pComment = taosStrdup(mer1.me.ctbEntry.comment);





×







292



      if (NULL == cfgRsp.pComment) {





×







293



        code = terrno;




×







294



        goto _exit;




×







295



      }










296



    }











UNCOV

297



    STag *pTag = (STag *)mer1.me.ctbEntry.pTags;




×








UNCOV

298



    cfgRsp.tagsLen = pTag->len;




×








UNCOV

299



    cfgRsp.pTags = taosMemoryMalloc(cfgRsp.tagsLen);





×








UNCOV

300



    if (NULL == cfgRsp.pTags) {





×







301



      code = terrno;




×







302



      goto _exit;




×







303



    }











UNCOV

304



    (void)memcpy(cfgRsp.pTags, pTag, cfgRsp.tagsLen);




×








UNCOV

305



  } else if (mer1.me.type == TSDB_NORMAL_TABLE) {





×








UNCOV

306



    schema = mer1.me.ntbEntry.schemaRow;




×








UNCOV

307



    cfgRsp.ttl = mer1.me.ntbEntry.ttlDays;




×








UNCOV

308



    cfgRsp.commentLen = mer1.me.ntbEntry.commentLen;




×








UNCOV

309



    if (mer1.me.ntbEntry.commentLen > 0) {





×







310



      cfgRsp.pComment = taosStrdup(mer1.me.ntbEntry.comment);





×







311



      if (NULL == cfgRsp.pComment) {





×







312



        code = terrno;




×







313



        goto _exit;




×







314



      }










315



    }










316



  } else {










317



    vError("vnodeGetTableCfg get invalid table type:%d", mer1.me.type);





×







318



    code = TSDB_CODE_APP_ERROR;




×







319



    goto _exit;




×







320



  }










321













UNCOV

322



  cfgRsp.numOfTags = schemaTag.nCols;




×








UNCOV

323



  cfgRsp.numOfColumns = schema.nCols;




×








UNCOV

324



  cfgRsp.pSchemas = (SSchema *)taosMemoryMalloc(sizeof(SSchema) * (cfgRsp.numOfColumns + cfgRsp.numOfTags));





×








UNCOV

325



  cfgRsp.pSchemaExt = (SSchemaExt *)taosMemoryMalloc(cfgRsp.numOfColumns * sizeof(SSchemaExt));





×







326













UNCOV

327



  if (NULL == cfgRsp.pSchemas || NULL == cfgRsp.pSchemaExt) {





×







328



    code = terrno;




×







329



    goto _exit;




×







330



  }











UNCOV

331



  (void)memcpy(cfgRsp.pSchemas, schema.pSchema, sizeof(SSchema) * schema.nCols);




×








UNCOV

332



  if (schemaTag.nCols) {





×








UNCOV

333



    (void)memcpy(cfgRsp.pSchemas + schema.nCols, schemaTag.pSchema, sizeof(SSchema) * schemaTag.nCols);




×







334



  }










335












336



  // if (useCompress(cfgRsp.tableType)) {










337













UNCOV

338



  SMetaReader     *pReader = mer1.me.type == TSDB_CHILD_TABLE ? &mer2 : &mer1;





×








UNCOV

339



  SColCmprWrapper *pColCmpr = &pReader->me.colCmpr;




×







340













UNCOV

341



  for (int32_t i = 0; i < cfgRsp.numOfColumns; i++) {





×








UNCOV

342



    SColCmpr   *pCmpr = &pColCmpr->pColCmpr[i];




×








UNCOV

343



    SSchemaExt *pSchExt = cfgRsp.pSchemaExt + i;




×








UNCOV

344



    pSchExt->colId = pCmpr->id;




×








UNCOV

345



    pSchExt->compress = pCmpr->alg;




×







346



  }










347



  //}










348












349



  // encode and send response











UNCOV

350



  rspLen = tSerializeSTableCfgRsp(NULL, 0, &cfgRsp);




×








UNCOV

351



  if (rspLen < 0) {





×







352



    code = terrno;




×







353



    goto _exit;




×







354



  }










355













UNCOV

356



  if (direct) {





×







357



    pRsp = rpcMallocCont(rspLen);




×







358



  } else {











UNCOV

359



    pRsp = taosMemoryCalloc(1, rspLen);





×







360



  }










361













UNCOV

362



  if (pRsp == NULL) {





×







363



    code = terrno;




×







364



    goto _exit;




×







365



  }










366













UNCOV

367



  rspLen = tSerializeSTableCfgRsp(pRsp, rspLen, &cfgRsp);




×








UNCOV

368



  if (rspLen < 0) {





×







369



    code = terrno;




×







370



    goto _exit;




×







371



  }










372













UNCOV

373



_exit:




×








UNCOV

374



  rpcMsg.info = pMsg->info;




×








UNCOV

375



  rpcMsg.pCont = pRsp;




×








UNCOV

376



  rpcMsg.contLen = rspLen;




×








UNCOV

377



  rpcMsg.code = code;




×








UNCOV

378



  rpcMsg.msgType = pMsg->msgType;




×







379













UNCOV

380



  if (code) {





×







381



    qError("get table %s cfg failed cause of %s", cfgReq.tbName, tstrerror(code));





×







382



  }










383













UNCOV

384



  if (direct) {





×







385



    tmsgSendRsp(&rpcMsg);




×







386



  } else {











UNCOV

387



    *pMsg = rpcMsg;




×







388



  }










389













UNCOV

390



  tFreeSTableCfgRsp(&cfgRsp);




×








UNCOV

391



  metaReaderClear(&mer2);




×








UNCOV

392



  metaReaderClear(&mer1);




×








UNCOV

393



  return code;




×







394



}










395












396



static FORCE_INLINE void vnodeFreeSBatchRspMsg(void *p) {










397



  if (NULL == p) {










398



    return;










399



  }










400












401



  SBatchRspMsg *pRsp = (SBatchRspMsg *)p;










402



  rpcFreeCont(pRsp->msg);










403



}










404













UNCOV

405



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




×








UNCOV

406



  int32_t      code = 0;




×








UNCOV

407



  int32_t      rspSize = 0;




×








UNCOV

408



  SBatchReq    batchReq = {0};




×








UNCOV

409



  SBatchMsg   *req = NULL;




×








UNCOV

410



  SBatchRspMsg rsp = {0};




×








UNCOV

411



  SBatchRsp    batchRsp = {0};




×








UNCOV

412



  SRpcMsg      reqMsg = *pMsg;




×








UNCOV

413



  SRpcMsg      rspMsg = {0};




×








UNCOV

414



  void        *pRsp = NULL;




×







415













UNCOV

416



  if (tDeserializeSBatchReq(pMsg->pCont, pMsg->contLen, &batchReq)) {





×







417



    code = terrno;




×







418



    qError("tDeserializeSBatchReq failed");





×







419



    goto _exit;




×







420



  }










421













UNCOV

422



  int32_t msgNum = taosArrayGetSize(batchReq.pMsgs);




×








UNCOV

423



  if (msgNum >= MAX_META_MSG_IN_BATCH) {





×







424



    code = TSDB_CODE_INVALID_MSG;




×







425



    qError("too many msgs %d in vnode batch meta req", msgNum);





×







426



    goto _exit;




×







427



  }










428













UNCOV

429



  batchRsp.pRsps = taosArrayInit(msgNum, sizeof(SBatchRspMsg));




×








UNCOV

430



  if (NULL == batchRsp.pRsps) {





×







431



    code = terrno;




×







432



    qError("taosArrayInit %d SBatchRspMsg failed", msgNum);





×







433



    goto _exit;




×







434



  }










435













UNCOV

436



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





×








UNCOV

437



    req = taosArrayGet(batchReq.pMsgs, i);




×








UNCOV

438



    if (req == NULL) {





×








UNCOV

439



      code = terrno;




×







440



      goto _exit;




×







441



    }










442













UNCOV

443



    reqMsg.msgType = req->msgType;




×








UNCOV

444



    reqMsg.pCont = req->msg;




×








UNCOV

445



    reqMsg.contLen = req->msgLen;




×







446













UNCOV

447



    switch (req->msgType) {





×








UNCOV

448



      case TDMT_VND_TABLE_META:




×







449



        // error code has been set into reqMsg, no need to handle it here.











UNCOV

450



        if (TSDB_CODE_SUCCESS != vnodeGetTableMeta(pVnode, &reqMsg, false)) {





×








UNCOV

451



          qWarn("vnodeGetBatchMeta failed, msgType:%d", req->msgType);





×







452



        }











UNCOV

453



        break;




×








UNCOV

454



      case TDMT_VND_TABLE_NAME:




×







455



        // error code has been set into reqMsg, no need to handle it here.











UNCOV

456



        if (TSDB_CODE_SUCCESS != vnodeGetTableMeta(pVnode, &reqMsg, false)) {





×








UNCOV

457



          qWarn("vnodeGetBatchName failed, msgType:%d", req->msgType);





×







458



        }











UNCOV

459



        break;




×








UNCOV

460



      case TDMT_VND_TABLE_CFG:




×







461



        // error code has been set into reqMsg, no need to handle it here.











UNCOV

462



        if (TSDB_CODE_SUCCESS != vnodeGetTableCfg(pVnode, &reqMsg, false)) {





×







463



          qWarn("vnodeGetBatchMeta failed, msgType:%d", req->msgType);





×







464



        }











UNCOV

465



        break;




×








UNCOV

466



      case TDMT_VND_GET_STREAM_PROGRESS:




×







467



        // error code has been set into reqMsg, no need to handle it here.











UNCOV

468



        if (TSDB_CODE_SUCCESS != vnodeGetStreamProgress(pVnode, &reqMsg, false)) {





×







469



          qWarn("vnodeGetBatchMeta failed, msgType:%d", req->msgType);





×







470



        }











UNCOV

471



        break;




×







472



      default:




×







473



        qError("invalid req msgType %d", req->msgType);





×







474



        reqMsg.code = TSDB_CODE_INVALID_MSG;




×







475



        reqMsg.pCont = NULL;




×







476



        reqMsg.contLen = 0;




×







477



        break;




×







478



    }










479













UNCOV

480



    rsp.msgIdx = req->msgIdx;




×








UNCOV

481



    rsp.reqType = reqMsg.msgType;




×








UNCOV

482



    rsp.msgLen = reqMsg.contLen;




×








UNCOV

483



    rsp.rspCode = reqMsg.code;




×








UNCOV

484



    rsp.msg = reqMsg.pCont;




×







485













UNCOV

486



    if (NULL == taosArrayPush(batchRsp.pRsps, &rsp)) {





×







487



      qError("taosArrayPush failed");





×







488



      code = terrno;




×







489



      goto _exit;




×







490



    }










491



  }










492













UNCOV

493



  rspSize = tSerializeSBatchRsp(NULL, 0, &batchRsp);




×








UNCOV

494



  if (rspSize < 0) {





×







495



    qError("tSerializeSBatchRsp failed");





×







496



    code = terrno;




×







497



    goto _exit;




×







498



  }











UNCOV

499



  pRsp = rpcMallocCont(rspSize);




×








UNCOV

500



  if (pRsp == NULL) {





×







501



    qError("rpcMallocCont %d failed", rspSize);





×







502



    code = terrno;




×







503



    goto _exit;




×







504



  }











UNCOV

505



  if (tSerializeSBatchRsp(pRsp, rspSize, &batchRsp) < 0) {





×







506



    qError("tSerializeSBatchRsp %d failed", rspSize);





×







507



    code = terrno;




×







508



    goto _exit;




×







509



  }










510













UNCOV

511



_exit:




×







512













UNCOV

513



  rspMsg.info = pMsg->info;




×








UNCOV

514



  rspMsg.pCont = pRsp;




×








UNCOV

515



  rspMsg.contLen = rspSize;




×








UNCOV

516



  rspMsg.code = code;




×








UNCOV

517



  rspMsg.msgType = pMsg->msgType;




×







518













UNCOV

519



  if (code) {





×







520



    qError("vnd get batch meta failed cause of %s", tstrerror(code));





×







521



  }










522













UNCOV

523



  taosArrayDestroyEx(batchReq.pMsgs, tFreeSBatchReqMsg);




×








UNCOV

524



  taosArrayDestroyEx(batchRsp.pRsps, tFreeSBatchRspMsg);




×







525













UNCOV

526



  tmsgSendRsp(&rspMsg);




×







527













UNCOV

528



  return code;




×







529



}










530













UNCOV

531



int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad) {




×








UNCOV

532



  SSyncState state = syncGetState(pVnode->sync);




×








UNCOV

533



  pLoad->syncAppliedIndex = pVnode->state.applied;




×








UNCOV

534



  syncGetCommitIndex(pVnode->sync, &pLoad->syncCommitIndex);




×







535













UNCOV

536



  pLoad->vgId = TD_VID(pVnode);




×








UNCOV

537



  pLoad->syncState = state.state;




×








UNCOV

538



  pLoad->syncRestore = state.restored;




×








UNCOV

539



  pLoad->syncTerm = state.term;




×








UNCOV

540



  pLoad->roleTimeMs = state.roleTimeMs;




×








UNCOV

541



  pLoad->startTimeMs = state.startTimeMs;




×








UNCOV

542



  pLoad->syncCanRead = state.canRead;




×








UNCOV

543



  pLoad->learnerProgress = state.progress;




×








UNCOV

544



  pLoad->cacheUsage = tsdbCacheGetUsage(pVnode);




×








UNCOV

545



  pLoad->numOfCachedTables = tsdbCacheGetElems(pVnode);




×








UNCOV

546



  pLoad->numOfTables = metaGetTbNum(pVnode->pMeta);




×








UNCOV

547



  pLoad->numOfTimeSeries = metaGetTimeSeriesNum(pVnode->pMeta, 1);




×








UNCOV

548



  pLoad->totalStorage = (int64_t)3 * 1073741824;




×








UNCOV

549



  pLoad->compStorage = (int64_t)2 * 1073741824;




×








UNCOV

550



  pLoad->pointsWritten = 100;




×








UNCOV

551



  pLoad->numOfSelectReqs = 1;




×








UNCOV

552



  pLoad->numOfInsertReqs = atomic_load_64(&pVnode->statis.nInsert);




×








UNCOV

553



  pLoad->numOfInsertSuccessReqs = atomic_load_64(&pVnode->statis.nInsertSuccess);




×








UNCOV

554



  pLoad->numOfBatchInsertReqs = atomic_load_64(&pVnode->statis.nBatchInsert);




×








UNCOV

555



  pLoad->numOfBatchInsertSuccessReqs = atomic_load_64(&pVnode->statis.nBatchInsertSuccess);




×








UNCOV

556



  return 0;




×







557



}










558












559



int32_t vnodeGetLoadLite(SVnode *pVnode, SVnodeLoadLite *pLoad) {




×







560



  SSyncState syncState = syncGetState(pVnode->sync);




×







561



  if (syncState.state == TAOS_SYNC_STATE_LEADER || syncState.state == TAOS_SYNC_STATE_ASSIGNED_LEADER) {





×







562



    pLoad->vgId = TD_VID(pVnode);




×







563



    pLoad->nTimeSeries = metaGetTimeSeriesNum(pVnode->pMeta, 1);




×







564



    return 0;




×







565



  }










566



  return -1;




×







567



}










568



/**










569



 * @brief Reset the statistics value by monitor interval










570



 *










571



 * @param pVnode










572



 * @param pLoad










573



 */











UNCOV

574



void vnodeResetLoad(SVnode *pVnode, SVnodeLoad *pLoad) {




×








UNCOV

575



  VNODE_GET_LOAD_RESET_VALS(pVnode->statis.nInsert, pLoad->numOfInsertReqs, 64, "nInsert");





×








UNCOV

576



  VNODE_GET_LOAD_RESET_VALS(pVnode->statis.nInsertSuccess, pLoad->numOfInsertSuccessReqs, 64, "nInsertSuccess");





×








UNCOV

577



  VNODE_GET_LOAD_RESET_VALS(pVnode->statis.nBatchInsert, pLoad->numOfBatchInsertReqs, 64, "nBatchInsert");





×








UNCOV

578



  VNODE_GET_LOAD_RESET_VALS(pVnode->statis.nBatchInsertSuccess, pLoad->numOfBatchInsertSuccessReqs, 64,





×







579



                            "nBatchInsertSuccess");











UNCOV

580



}




×







581













UNCOV

582



void vnodeGetInfo(void *pVnode, const char **dbname, int32_t *vgId, int64_t *numOfTables, int64_t *numOfNormalTables) {




×








UNCOV

583



  SVnode    *pVnodeObj = pVnode;




×








UNCOV

584



  SVnodeCfg *pConf = &pVnodeObj->config;




×







585













UNCOV

586



  if (dbname) {





×








UNCOV

587



    *dbname = pConf->dbname;




×







588



  }










589













UNCOV

590



  if (vgId) {





×








UNCOV

591



    *vgId = TD_VID(pVnodeObj);




×







592



  }










593













UNCOV

594



  if (numOfTables) {





×







595



    *numOfTables = pConf->vndStats.numOfNTables + pConf->vndStats.numOfCTables;




×







596



  }










597













UNCOV

598



  if (numOfNormalTables) {





×







599



    *numOfNormalTables = pConf->vndStats.numOfNTables;




×







600



  }











UNCOV

601



}




×







602












603



int32_t vnodeGetTableList(void *pVnode, int8_t type, SArray *pList) {




×







604



  if (type == TSDB_SUPER_TABLE) {





×







605



    return vnodeGetStbIdList(pVnode, 0, pList);




×







606



  } else {










607



    return TSDB_CODE_INVALID_PARA;




×







608



  }










609



}










610












611



int32_t vnodeGetAllTableList(SVnode *pVnode, uint64_t uid, SArray *list) {




×







612



  int32_t      code = TSDB_CODE_SUCCESS;




×







613



  SMCtbCursor *pCur = metaOpenCtbCursor(pVnode, uid, 1);




×







614



  if (NULL == pCur) {





×







615



    qError("vnode get all table list failed");





×







616



    return terrno;




×







617



  }










618












619



  while (1) {




×







620



    tb_uid_t id = metaCtbCursorNext(pCur);




×







621



    if (id == 0) {





×







622



      break;




×







623



    }










624












625



    STableKeyInfo info = {uid = id};




×







626



    if (NULL == taosArrayPush(list, &info)) {





×







627



      qError("taosArrayPush failed");





×







628



      code = terrno;




×







629



      goto _exit;




×







630



    }










631



  }










632



_exit:




×







633



  metaCloseCtbCursor(pCur);




×







634



  return code;




×







635



}










636












637



int32_t vnodeGetCtbIdListByFilter(SVnode *pVnode, int64_t suid, SArray *list, bool (*filter)(void *arg), void *arg) {




×







638



  return 0;




×







639



}










640













UNCOV

641



int32_t vnodeGetCtbIdList(void *pVnode, int64_t suid, SArray *list) {




×








UNCOV

642



  int32_t      code = TSDB_CODE_SUCCESS;




×








UNCOV

643



  SVnode      *pVnodeObj = pVnode;




×








UNCOV

644



  SMCtbCursor *pCur = metaOpenCtbCursor(pVnodeObj, suid, 1);




×








UNCOV

645



  if (NULL == pCur) {





×







646



    qError("vnode get all table list failed");





×







647



    return terrno;




×







648



  }










649













UNCOV

650



  while (1) {




×








UNCOV

651



    tb_uid_t id = metaCtbCursorNext(pCur);




×








UNCOV

652



    if (id == 0) {





×








UNCOV

653



      break;




×







654



    }










655













UNCOV

656



    if (NULL == taosArrayPush(list, &id)) {





×







657



      qError("taosArrayPush failed");





×







658



      code = terrno;




×







659



      goto _exit;




×







660



    }










661



  }










662













UNCOV

663



_exit:




×








UNCOV

664



  metaCloseCtbCursor(pCur);




×








UNCOV

665



  return code;




×







666



}










667













UNCOV

668



int32_t vnodeGetStbIdList(SVnode *pVnode, int64_t suid, SArray *list) {




×








UNCOV

669



  int32_t      code = TSDB_CODE_SUCCESS;




×








UNCOV

670



  SMStbCursor *pCur = metaOpenStbCursor(pVnode->pMeta, suid);




×








UNCOV

671



  if (!pCur) {





×







672



    return TSDB_CODE_OUT_OF_MEMORY;




×







673



  }










674













UNCOV

675



  while (1) {




×








UNCOV

676



    tb_uid_t id = metaStbCursorNext(pCur);




×








UNCOV

677



    if (id == 0) {





×








UNCOV

678



      break;




×







679



    }










680













UNCOV

681



    if (NULL == taosArrayPush(list, &id)) {





×







682



      qError("taosArrayPush failed");





×







683



      code = terrno;




×







684



      goto _exit;




×







685



    }










686



  }










687













UNCOV

688



_exit:




×








UNCOV

689



  metaCloseStbCursor(pCur);




×








UNCOV

690



  return code;




×







691



}










692












693



int32_t vnodeGetStbIdListByFilter(SVnode *pVnode, int64_t suid, SArray *list, bool (*filter)(void *arg, void *arg1),




×







694



                                  void *arg) {










695



  int32_t      code = TSDB_CODE_SUCCESS;




×







696



  SMStbCursor *pCur = metaOpenStbCursor(pVnode->pMeta, suid);




×







697



  if (!pCur) {





×







698



    return terrno;




×







699



  }










700












701



  while (1) {




×







702



    tb_uid_t id = metaStbCursorNext(pCur);




×







703



    if (id == 0) {





×







704



      break;




×







705



    }










706












707



    if ((*filter) && (*filter)(arg, &id)) {





×







708



      continue;




×







709



    }










710












711



    if (NULL == taosArrayPush(list, &id)) {





×







712



      qError("taosArrayPush failed");





×







713



      code = terrno;




×







714



      goto _exit;




×







715



    }










716



  }










717












718



_exit:




×







719



  metaCloseStbCursor(pCur);




×







720



  return code;




×







721



}










722













UNCOV

723



int32_t vnodeGetCtbNum(SVnode *pVnode, int64_t suid, int64_t *num) {




×








UNCOV

724



  SMCtbCursor *pCur = metaOpenCtbCursor(pVnode, suid, 0);




×








UNCOV

725



  if (!pCur) {





×







726



    return terrno;




×







727



  }










728













UNCOV

729



  *num = 0;




×








UNCOV

730



  while (1) {




×








UNCOV

731



    tb_uid_t id = metaCtbCursorNext(pCur);




×








UNCOV

732



    if (id == 0) {





×








UNCOV

733



      break;




×







734



    }










735













UNCOV

736



    ++(*num);




×







737



  }










738













UNCOV

739



  metaCloseCtbCursor(pCur);




×








UNCOV

740



  return TSDB_CODE_SUCCESS;




×







741



}










742













UNCOV

743



int32_t vnodeGetStbColumnNum(SVnode *pVnode, tb_uid_t suid, int *num) {




×








UNCOV

744



  SSchemaWrapper *pSW = metaGetTableSchema(pVnode->pMeta, suid, -1, 0);




×








UNCOV

745



  if (pSW) {





×








UNCOV

746



    *num = pSW->nCols;





×







747



    tDeleteSchemaWrapper(pSW);










748



  } else {











UNCOV

749



    *num = 2;




×







750



  }










751













UNCOV

752



  return TSDB_CODE_SUCCESS;




×







753



}










754












755



#ifdef TD_ENTERPRISE










756



const char *tkLogStb[] = {"cluster_info",










757



                          "data_dir",










758



                          "dnodes_info",










759



                          "d_info",










760



                          "grants_info",










761



                          "keeper_monitor",










762



                          "logs",










763



                          "log_dir",










764



                          "log_summary",










765



                          "m_info",










766



                          "taosadapter_restful_http_request_fail",










767



                          "taosadapter_restful_http_request_in_flight",










768



                          "taosadapter_restful_http_request_summary_milliseconds",










769



                          "taosadapter_restful_http_request_total",










770



                          "taosadapter_system_cpu_percent",










771



                          "taosadapter_system_mem_percent",










772



                          "temp_dir",










773



                          "vgroups_info",










774



                          "vnodes_role"};










775



const char *tkAuditStb[] = {"operations"};










776



const int   tkLogStbNum = ARRAY_SIZE(tkLogStb);










777



const int   tkAuditStbNum = ARRAY_SIZE(tkAuditStb);










778












779



// exclude stbs of taoskeeper log











UNCOV

780



static int32_t vnodeGetTimeSeriesBlackList(SVnode *pVnode, int32_t *tbSize) {




×








UNCOV

781



  int32_t      code = TSDB_CODE_SUCCESS;




×








UNCOV

782



  int32_t      tbNum = 0;




×








UNCOV

783



  const char **pTbArr = NULL;




×








UNCOV

784



  const char  *dbName = NULL;




×








UNCOV

785



  *tbSize = 0;




×







786













UNCOV

787



  if (!(dbName = strchr(pVnode->config.dbname, '.'))) return 0;





×








UNCOV

788



  if (0 == strncmp(++dbName, "log", TSDB_DB_NAME_LEN)) {





×







789



    tbNum = tkLogStbNum;




×







790



    pTbArr = (const char **)&tkLogStb;




×








UNCOV

791



  } else if (0 == strncmp(dbName, "audit", TSDB_DB_NAME_LEN)) {





×







792



    tbNum = tkAuditStbNum;




×







793



    pTbArr = (const char **)&tkAuditStb;




×







794



  }











UNCOV

795



  if (tbNum && pTbArr) {





×







796



    *tbSize = metaSizeOfTbFilterCache(pVnode->pMeta, 0);




×







797



    if (*tbSize < tbNum) {





×







798



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





×







799



        tb_uid_t suid = metaGetTableEntryUidByName(pVnode->pMeta, pTbArr[i]);




×







800



        if (suid != 0) {





×







801



          code = metaPutTbToFilterCache(pVnode->pMeta, &suid, 0);




×







802



          if (TSDB_CODE_SUCCESS != code) {





×







803



            return code;




×







804



          }










805



        }










806



      }










807



      *tbSize = metaSizeOfTbFilterCache(pVnode->pMeta, 0);




×







808



    }










809



  }










810













UNCOV

811



  return code;




×







812



}










813



#endif










814












815



static bool vnodeTimeSeriesFilter(void *arg1, void *arg2) {




×







816



  SVnode *pVnode = (SVnode *)arg1;




×







817












818



  if (metaTbInFilterCache(pVnode->pMeta, arg2, 0)) {





×







819



    return true;




×







820



  }










821



  return false;




×







822



}










823













UNCOV

824



int32_t vnodeGetTimeSeriesNum(SVnode *pVnode, int64_t *num) {




×








UNCOV

825



  SArray *suidList = NULL;




×







826













UNCOV

827



  if (!(suidList = taosArrayInit(1, sizeof(tb_uid_t)))) {





×







828



    return terrno;




×







829



  }










830













UNCOV

831



  int32_t tbFilterSize = 0;




×








UNCOV

832



  int32_t code = TSDB_CODE_SUCCESS;




×







833



#ifdef TD_ENTERPRISE











UNCOV

834



  code = vnodeGetTimeSeriesBlackList(pVnode, &tbFilterSize);




×








UNCOV

835



  if (TSDB_CODE_SUCCESS != code) {





×







836



    goto _exit;




×







837



  }










838



#endif










839













UNCOV

840



  if ((!tbFilterSize && vnodeGetStbIdList(pVnode, 0, suidList) < 0) ||





×








UNCOV

841



      (tbFilterSize && vnodeGetStbIdListByFilter(pVnode, 0, suidList, vnodeTimeSeriesFilter, pVnode) < 0)) {





×







842



    qError("vgId:%d, failed to get stb id list error: %s", TD_VID(pVnode), terrstr());





×







843



    taosArrayDestroy(suidList);




×







844



    return terrno;




×







845



  }










846













UNCOV

847



  *num = 0;




×








UNCOV

848



  int64_t arrSize = taosArrayGetSize(suidList);




×








UNCOV

849



  for (int64_t i = 0; i < arrSize; ++i) {





×








UNCOV

850



    tb_uid_t suid = *(tb_uid_t *)taosArrayGet(suidList, i);




×







851













UNCOV

852



    int64_t ctbNum = 0;




×








UNCOV

853



    int32_t numOfCols = 0;




×








UNCOV

854



    code = metaGetStbStats(pVnode, suid, &ctbNum, &numOfCols);




×








UNCOV

855



    if (TSDB_CODE_SUCCESS != code) {





×







856



      goto _exit;




×







857



    }











UNCOV

858



    *num += ctbNum * (numOfCols - 1);




×







859



  }










860













UNCOV

861



_exit:




×








UNCOV

862



  taosArrayDestroy(suidList);




×








UNCOV

863



  return TSDB_CODE_SUCCESS;




×







864



}










865












866



int32_t vnodeGetAllCtbNum(SVnode *pVnode, int64_t *num) {




×







867



  SMStbCursor *pCur = metaOpenStbCursor(pVnode->pMeta, 0);




×







868



  if (!pCur) {





×







869



    return terrno;




×







870



  }










871












872



  *num = 0;




×







873



  while (1) {




×







874



    tb_uid_t id = metaStbCursorNext(pCur);




×







875



    if (id == 0) {





×







876



      break;




×







877



    }










878












879



    int64_t ctbNum = 0;




×







880



    int32_t code = vnodeGetCtbNum(pVnode, id, &ctbNum);




×







881



    if (TSDB_CODE_SUCCESS != code) {





×







882



      metaCloseStbCursor(pCur);




×







883



      return code;




×







884



    }










885












886



    *num += ctbNum;




×







887



  }










888












889



  metaCloseStbCursor(pCur);




×







890



  return TSDB_CODE_SUCCESS;




×







891



}










892













UNCOV

893



void *vnodeGetIdx(void *pVnode) {




×








UNCOV

894



  if (pVnode == NULL) {





×







895



    return NULL;




×







896



  }










897













UNCOV

898



  return metaGetIdx(((SVnode *)pVnode)->pMeta);




×







899



}










900













UNCOV

901



void *vnodeGetIvtIdx(void *pVnode) {




×








UNCOV

902



  if (pVnode == NULL) {





×







903



    return NULL;




×







904



  }











UNCOV

905



  return metaGetIvtIdx(((SVnode *)pVnode)->pMeta);




×







906



}










907













UNCOV

908



int32_t vnodeGetTableSchema(void *pVnode, int64_t uid, STSchema **pSchema, int64_t *suid) {




×








UNCOV

909



  return tsdbGetTableSchema(((SVnode *)pVnode)->pMeta, uid, pSchema, suid);




×







910



}










911













UNCOV

912



int32_t vnodeGetDBSize(void *pVnode, SDbSizeStatisInfo *pInfo) {




×








UNCOV

913



  SVnode *pVnodeObj = pVnode;




×








UNCOV

914



  if (pVnodeObj == NULL) {





×







915



    return TSDB_CODE_VND_NOT_EXIST;




×







916



  }











UNCOV

917



  int32_t code = 0;




×








UNCOV

918



  char    path[TSDB_FILENAME_LEN] = {0};




×







919













UNCOV

920



  char   *dirName[] = {VNODE_TSDB_DIR, VNODE_WAL_DIR, VNODE_META_DIR, VNODE_TSDB_CACHE_DIR};




×







921



  int64_t dirSize[4];










922













UNCOV

923



  vnodeGetPrimaryDir(pVnodeObj->path, pVnodeObj->diskPrimary, pVnodeObj->pTfs, path, TSDB_FILENAME_LEN);




×








UNCOV

924



  int32_t offset = strlen(path);




×







925













UNCOV

926



  for (int i = 0; i < sizeof(dirName) / sizeof(dirName[0]); i++) {





×








UNCOV

927



    int64_t size = {0};




×








UNCOV

928



    (void)snprintf(path + offset, TSDB_FILENAME_LEN, "%s%s", TD_DIRSEP, dirName[i]);




×








UNCOV

929



    code = taosGetDirSize(path, &size);




×








UNCOV

930



    if (code != 0) {





×







931



      return code;




×







932



    }











UNCOV

933



    path[offset] = 0;




×








UNCOV

934



    dirSize[i] = size;




×







935



  }










936













UNCOV

937



  pInfo->l1Size = dirSize[0] - dirSize[3];




×








UNCOV

938



  pInfo->walSize = dirSize[1];




×








UNCOV

939



  pInfo->metaSize = dirSize[2];




×








UNCOV

940



  pInfo->cacheSize = dirSize[3];




×







941













UNCOV

942



  code = tsdbGetS3Size(pVnodeObj->pTsdb, &pInfo->s3Size);




×







943













UNCOV

944



  return code;




×







945



}










946













UNCOV

947



int32_t vnodeGetStreamProgress(SVnode *pVnode, SRpcMsg *pMsg, bool direct) {




×








UNCOV

948



  int32_t            code = 0;




×







949



  SStreamProgressReq req;











UNCOV

950



  SStreamProgressRsp rsp = {0};




×








UNCOV

951



  SRpcMsg            rpcMsg = {.info = pMsg->info, .code = 0};




×








UNCOV

952



  char              *buf = NULL;




×








UNCOV

953



  int32_t            rspLen = 0;




×








UNCOV

954



  code = tDeserializeStreamProgressReq(pMsg->pCont, pMsg->contLen, &req);




×







955













UNCOV

956



  if (code == TSDB_CODE_SUCCESS) {





×








UNCOV

957



    rsp.fetchIdx = req.fetchIdx;




×








UNCOV

958



    rsp.subFetchIdx = req.subFetchIdx;




×








UNCOV

959



    rsp.vgId = req.vgId;




×








UNCOV

960



    rsp.streamId = req.streamId;




×








UNCOV

961



    rspLen = tSerializeStreamProgressRsp(0, 0, &rsp);




×








UNCOV

962



    if (rspLen < 0) {





×







963



      code = terrno;




×







964



      goto _OVER;




×







965



    }











UNCOV

966



    if (direct) {





×







967



      buf = rpcMallocCont(rspLen);




×







968



    } else {











UNCOV

969



      buf = taosMemoryCalloc(1, rspLen);





×







970



    }











UNCOV

971



    if (!buf) {





×







972



      code = terrno;




×







973



      goto _OVER;




×







974



    }










975



  }










976













UNCOV

977



  if (code == TSDB_CODE_SUCCESS) {





×








UNCOV

978



    code = tqGetStreamExecInfo(pVnode, req.streamId, &rsp.progressDelay, &rsp.fillHisFinished);




×







979



  }











UNCOV

980



  if (code == TSDB_CODE_SUCCESS) {





×








UNCOV

981



    rspLen = tSerializeStreamProgressRsp(buf, rspLen, &rsp);




×








UNCOV

982



    if (rspLen < 0) {





×







983



      code = terrno;




×







984



      goto _OVER;




×







985



    }











UNCOV

986



    rpcMsg.pCont = buf;




×








UNCOV

987



    buf = NULL;




×








UNCOV

988



    rpcMsg.contLen = rspLen;




×








UNCOV

989



    rpcMsg.code = code;




×








UNCOV

990



    rpcMsg.msgType = pMsg->msgType;




×








UNCOV

991



    if (direct) {





×







992



      tmsgSendRsp(&rpcMsg);




×







993



    } else {











UNCOV

994



      *pMsg = rpcMsg;




×







995



    }










996



  }










997












998



_OVER:




×








UNCOV

999



  if (buf) {





×







1000



    taosMemoryFree(buf);





×







1001



  }











UNCOV

1002



  return code;




×







1003



}
























    

  • 

    •