taosdata / TDengine / #3562

coverage: 26.655% (-32.2%) from 58.812%
#3562

push

travis-ci

web-flow 
Merge pull request #29229 from taosdata/enh/TS-5749-3.0

enh: seperate tsdb async tasks to different thread pools

21498 of 109421 branches covered (19.65%)

Branch coverage included in aggregate %.

66 of 96 new or added lines in 7 files covered. (68.75%)

39441 existing lines in 157 files now uncovered.

35007 of 102566 relevant lines covered (34.13%)

53922.97 hits per line

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

0.0
/source/client/src/clientRawBlockWrite.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 "cJSON.h"










17



#include "clientInt.h"










18



#include "parser.h"










19



#include "tcol.h"










20



#include "tcompression.h"










21



#include "tdatablock.h"










22



#include "tdef.h"










23



#include "tglobal.h"










24



#include "tmsgtype.h"










25












26



#define RAW_NULL_CHECK(c) \










27



  do {                    \










28



    if (c == NULL) {      \










29



      code = terrno;      \










30



      goto end;           \










31



    }                     \










32



  } while (0)










33












34



#define RAW_FALSE_CHECK(c)           \










35



  do {                               \










36



    if (!c) {                        \










37



      code = TSDB_CODE_INVALID_PARA; \










38



      goto end;                      \










39



    }                                \










40



  } while (0)










41












42



#define RAW_RETURN_CHECK(c) \










43



  do {                      \










44



    code = c;               \










45



    if (code != 0) {        \










46



      goto end;             \










47



    }                       \










48



  } while (0)










49












50



#define LOG_ID_TAG   "connId:0x%" PRIx64 ",QID:0x%" PRIx64










51



#define LOG_ID_VALUE *(int64_t*)taos, pRequest->requestId










52












53



#define TMQ_META_VERSION "1.0"










54













UNCOV

55



static bool  tmqAddJsonObjectItem(cJSON *object, const char *string, cJSON *item){




×








UNCOV

56



  bool ret = cJSON_AddItemToObject(object, string, item);




×








UNCOV

57



  if (!ret){





×







58



    cJSON_Delete(item);




×







59



  }











UNCOV

60



  return ret;




×







61



}











UNCOV

62



static bool  tmqAddJsonArrayItem(cJSON *array, cJSON *item){




×








UNCOV

63



  bool ret = cJSON_AddItemToArray(array, item);




×








UNCOV

64



  if (!ret){





×







65



    cJSON_Delete(item);




×







66



  }











UNCOV

67



  return ret;




×







68



}










69












70












71



static int32_t  tmqWriteBatchMetaDataImpl(TAOS* taos, void* meta, uint32_t metaLen);











UNCOV

72



static tb_uid_t processSuid(tb_uid_t suid, char* db) {




×








UNCOV

73



  if (db == NULL) {





×







74



    return suid;




×







75



  }











UNCOV

76



  return suid + MurmurHash3_32(db, strlen(db));




×







77



}











UNCOV

78



static void buildCreateTableJson(SSchemaWrapper* schemaRow, SSchemaWrapper* schemaTag, char* name, int64_t id, int8_t t,




×







79



                                 SColCmprWrapper* pColCmprRow, cJSON** pJson) {











UNCOV

80



  if (schemaRow == NULL || name == NULL || pColCmprRow == NULL || pJson == NULL) {





×







81



    uError("invalid parameter, schemaRow:%p, name:%p, pColCmprRow:%p, pJson:%p", schemaRow, name, pColCmprRow, pJson);





×







82



    return;




×







83



  }











UNCOV

84



  int32_t code = TSDB_CODE_SUCCESS;




×








UNCOV

85



  int8_t  buildDefaultCompress = 0;




×








UNCOV

86



  if (pColCmprRow->nCols <= 0) {





×







87



    buildDefaultCompress = 1;




×







88



  }










89













UNCOV

90



  char*  string = NULL;




×








UNCOV

91



  cJSON* json = cJSON_CreateObject();




×








UNCOV

92



  RAW_NULL_CHECK(json);





×








UNCOV

93



  cJSON* type = cJSON_CreateString("create");




×








UNCOV

94



  RAW_NULL_CHECK(type);





×







95













UNCOV

96



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "type", type));





×








UNCOV

97



  cJSON* tableType = cJSON_CreateString(t == TSDB_NORMAL_TABLE ? "normal" : "super");





×








UNCOV

98



  RAW_NULL_CHECK(tableType);





×








UNCOV

99



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableType", tableType));





×








UNCOV

100



  cJSON* tableName = cJSON_CreateString(name);




×








UNCOV

101



  RAW_NULL_CHECK(tableName);





×








UNCOV

102



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableName", tableName));





×







103













UNCOV

104



  cJSON* columns = cJSON_CreateArray();




×








UNCOV

105



  RAW_NULL_CHECK(columns);





×








UNCOV

106



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "columns", columns));





×







107













UNCOV

108



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





×








UNCOV

109



    cJSON* column = cJSON_CreateObject();




×








UNCOV

110



    RAW_NULL_CHECK(column);





×








UNCOV

111



    RAW_FALSE_CHECK(tmqAddJsonArrayItem(columns, column));





×








UNCOV

112



    SSchema* s = schemaRow->pSchema + i;




×








UNCOV

113



    cJSON*   cname = cJSON_CreateString(s->name);




×








UNCOV

114



    RAW_NULL_CHECK(cname);





×








UNCOV

115



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "name", cname));





×








UNCOV

116



    cJSON* ctype = cJSON_CreateNumber(s->type);




×








UNCOV

117



    RAW_NULL_CHECK(ctype);





×








UNCOV

118



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "type", ctype));





×








UNCOV

119



    if (s->type == TSDB_DATA_TYPE_BINARY || s->type == TSDB_DATA_TYPE_VARBINARY || s->type == TSDB_DATA_TYPE_GEOMETRY) {





×








UNCOV

120



      int32_t length = s->bytes - VARSTR_HEADER_SIZE;




×








UNCOV

121



      cJSON*  cbytes = cJSON_CreateNumber(length);




×








UNCOV

122



      RAW_NULL_CHECK(cbytes);





×








UNCOV

123



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "length", cbytes));





×








UNCOV

124



    } else if (s->type == TSDB_DATA_TYPE_NCHAR) {





×








UNCOV

125



      int32_t length = (s->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;




×








UNCOV

126



      cJSON*  cbytes = cJSON_CreateNumber(length);




×








UNCOV

127



      RAW_NULL_CHECK(cbytes);





×








UNCOV

128



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "length", cbytes));





×







129



    }











UNCOV

130



    cJSON* isPk = cJSON_CreateBool(s->flags & COL_IS_KEY);




×








UNCOV

131



    RAW_NULL_CHECK(isPk);





×








UNCOV

132



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "isPrimarykey", isPk));





×







133













UNCOV

134



    if (pColCmprRow == NULL) {





×







135



      continue;




×







136



    }










137













UNCOV

138



    uint32_t alg = 0;




×








UNCOV

139



    if (buildDefaultCompress) {





×







140



      alg = createDefaultColCmprByType(s->type);




×







141



    } else {











UNCOV

142



      SColCmpr* pColCmpr = pColCmprRow->pColCmpr + i;




×








UNCOV

143



      alg = pColCmpr->alg;




×







144



    }











UNCOV

145



    const char* encode = columnEncodeStr(COMPRESS_L1_TYPE_U32(alg));




×








UNCOV

146



    RAW_NULL_CHECK(encode);





×








UNCOV

147



    const char* compress = columnCompressStr(COMPRESS_L2_TYPE_U32(alg));




×








UNCOV

148



    RAW_NULL_CHECK(compress);





×








UNCOV

149



    const char* level = columnLevelStr(COMPRESS_L2_TYPE_LEVEL_U32(alg));




×








UNCOV

150



    RAW_NULL_CHECK(level);





×







151













UNCOV

152



    cJSON* encodeJson = cJSON_CreateString(encode);




×








UNCOV

153



    RAW_NULL_CHECK(encodeJson);





×








UNCOV

154



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "encode", encodeJson));





×







155













UNCOV

156



    cJSON* compressJson = cJSON_CreateString(compress);




×








UNCOV

157



    RAW_NULL_CHECK(compressJson);





×








UNCOV

158



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "compress", compressJson));





×







159













UNCOV

160



    cJSON* levelJson = cJSON_CreateString(level);




×








UNCOV

161



    RAW_NULL_CHECK(levelJson);





×








UNCOV

162



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(column, "level", levelJson));





×







163



  }










164













UNCOV

165



  cJSON* tags = cJSON_CreateArray();




×








UNCOV

166



  RAW_NULL_CHECK(tags);





×








UNCOV

167



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tags", tags));





×







168













UNCOV

169



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





×








UNCOV

170



    cJSON* tag = cJSON_CreateObject();




×








UNCOV

171



    RAW_NULL_CHECK(tag);





×








UNCOV

172



    RAW_FALSE_CHECK(tmqAddJsonArrayItem(tags, tag));





×








UNCOV

173



    SSchema* s = schemaTag->pSchema + i;




×








UNCOV

174



    cJSON*   tname = cJSON_CreateString(s->name);




×








UNCOV

175



    RAW_NULL_CHECK(tname);





×








UNCOV

176



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "name", tname));





×








UNCOV

177



    cJSON* ttype = cJSON_CreateNumber(s->type);




×








UNCOV

178



    RAW_NULL_CHECK(ttype);





×








UNCOV

179



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "type", ttype));





×








UNCOV

180



    if (s->type == TSDB_DATA_TYPE_BINARY || s->type == TSDB_DATA_TYPE_VARBINARY || s->type == TSDB_DATA_TYPE_GEOMETRY) {





×








UNCOV

181



      int32_t length = s->bytes - VARSTR_HEADER_SIZE;




×








UNCOV

182



      cJSON*  cbytes = cJSON_CreateNumber(length);




×








UNCOV

183



      RAW_NULL_CHECK(cbytes);





×








UNCOV

184



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "length", cbytes));





×








UNCOV

185



    } else if (s->type == TSDB_DATA_TYPE_NCHAR) {





×








UNCOV

186



      int32_t length = (s->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;




×








UNCOV

187



      cJSON*  cbytes = cJSON_CreateNumber(length);




×








UNCOV

188



      RAW_NULL_CHECK(cbytes);





×








UNCOV

189



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "length", cbytes));





×







190



    }










191



  }










192













UNCOV

193



end:




×








UNCOV

194



  *pJson = json;




×







195



}










196












197



static int32_t setCompressOption(cJSON* json, uint32_t para) {




×







198



  if (json == NULL) {





×







199



    return TSDB_CODE_INVALID_PARA;




×







200



  }










201



  uint8_t encode = COMPRESS_L1_TYPE_U32(para);




×







202



  int32_t code = 0;




×







203



  if (encode != 0) {





×







204



    const char* encodeStr = columnEncodeStr(encode);




×







205



    RAW_NULL_CHECK(encodeStr);





×







206



    cJSON* encodeJson = cJSON_CreateString(encodeStr);




×







207



    RAW_NULL_CHECK(encodeJson);





×







208



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "encode", encodeJson));





×







209



    return code;




×







210



  }










211



  uint8_t compress = COMPRESS_L2_TYPE_U32(para);




×







212



  if (compress != 0) {





×







213



    const char* compressStr = columnCompressStr(compress);




×







214



    RAW_NULL_CHECK(compressStr);





×







215



    cJSON* compressJson = cJSON_CreateString(compressStr);




×







216



    RAW_NULL_CHECK(compressJson);





×







217



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "compress", compressJson));





×







218



    return code;




×







219



  }










220



  uint8_t level = COMPRESS_L2_TYPE_LEVEL_U32(para);




×







221



  if (level != 0) {





×







222



    const char* levelStr = columnLevelStr(level);




×







223



    RAW_NULL_CHECK(levelStr);





×







224



    cJSON* levelJson = cJSON_CreateString(levelStr);




×







225



    RAW_NULL_CHECK(levelJson);





×







226



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "level", levelJson));





×







227



    return code;




×







228



  }










229












230



end:




×







231



  return code;




×







232



}











UNCOV

233



static void buildAlterSTableJson(void* alterData, int32_t alterDataLen, cJSON** pJson) {




×








UNCOV

234



  if (alterData == NULL || pJson == NULL) {





×







235



    uError("invalid parameter in %s", __func__);





×







236



    return;




×







237



  }











UNCOV

238



  SMAlterStbReq req = {0};




×








UNCOV

239



  cJSON*        json = NULL;




×








UNCOV

240



  char*         string = NULL;




×








UNCOV

241



  int32_t       code = 0;




×







242













UNCOV

243



  if (tDeserializeSMAlterStbReq(alterData, alterDataLen, &req) != 0) {





×







244



    goto end;




×







245



  }










246













UNCOV

247



  json = cJSON_CreateObject();




×








UNCOV

248



  RAW_NULL_CHECK(json);





×








UNCOV

249



  cJSON* type = cJSON_CreateString("alter");




×








UNCOV

250



  RAW_NULL_CHECK(type);





×








UNCOV

251



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "type", type));





×








UNCOV

252



  SName name = {0};




×








UNCOV

253



  RAW_RETURN_CHECK(tNameFromString(&name, req.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE));





×








UNCOV

254



  cJSON* tableType = cJSON_CreateString("super");




×








UNCOV

255



  RAW_NULL_CHECK(tableType);





×








UNCOV

256



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableType", tableType));





×








UNCOV

257



  cJSON* tableName = cJSON_CreateString(name.tname);




×








UNCOV

258



  RAW_NULL_CHECK(tableName);





×








UNCOV

259



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableName", tableName));





×







260













UNCOV

261



  cJSON* alterType = cJSON_CreateNumber(req.alterType);




×








UNCOV

262



  RAW_NULL_CHECK(alterType);





×








UNCOV

263



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "alterType", alterType));





×








UNCOV

264



  switch (req.alterType) {





×








UNCOV

265



    case TSDB_ALTER_TABLE_ADD_TAG:




×







266



    case TSDB_ALTER_TABLE_ADD_COLUMN: {











UNCOV

267



      TAOS_FIELD* field = taosArrayGet(req.pFields, 0);




×








UNCOV

268



      RAW_NULL_CHECK(field);





×








UNCOV

269



      cJSON* colName = cJSON_CreateString(field->name);




×








UNCOV

270



      RAW_NULL_CHECK(colName);





×








UNCOV

271



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));





×








UNCOV

272



      cJSON* colType = cJSON_CreateNumber(field->type);




×








UNCOV

273



      RAW_NULL_CHECK(colType);





×








UNCOV

274



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colType", colType));





×







275













UNCOV

276



      if (field->type == TSDB_DATA_TYPE_BINARY || field->type == TSDB_DATA_TYPE_VARBINARY ||





×








UNCOV

277



          field->type == TSDB_DATA_TYPE_GEOMETRY) {





×








UNCOV

278



        int32_t length = field->bytes - VARSTR_HEADER_SIZE;




×








UNCOV

279



        cJSON*  cbytes = cJSON_CreateNumber(length);




×








UNCOV

280



        RAW_NULL_CHECK(cbytes);





×








UNCOV

281



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));





×








UNCOV

282



      } else if (field->type == TSDB_DATA_TYPE_NCHAR) {





×







283



        int32_t length = (field->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;




×







284



        cJSON*  cbytes = cJSON_CreateNumber(length);




×







285



        RAW_NULL_CHECK(cbytes);





×







286



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));





×







287



      }











UNCOV

288



      break;




×







289



    }










290



    case TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COMPRESS_OPTION: {




×







291



      SFieldWithOptions* field = taosArrayGet(req.pFields, 0);




×







292



      RAW_NULL_CHECK(field);





×







293



      cJSON* colName = cJSON_CreateString(field->name);




×







294



      RAW_NULL_CHECK(colName);





×







295



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));





×







296



      cJSON* colType = cJSON_CreateNumber(field->type);




×







297



      RAW_NULL_CHECK(colType);





×







298



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colType", colType));





×







299












300



      if (field->type == TSDB_DATA_TYPE_BINARY || field->type == TSDB_DATA_TYPE_VARBINARY ||





×







301



          field->type == TSDB_DATA_TYPE_GEOMETRY) {





×







302



        int32_t length = field->bytes - VARSTR_HEADER_SIZE;




×







303



        cJSON*  cbytes = cJSON_CreateNumber(length);




×







304



        RAW_NULL_CHECK(cbytes);





×







305



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));





×







306



      } else if (field->type == TSDB_DATA_TYPE_NCHAR) {





×







307



        int32_t length = (field->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;




×







308



        cJSON*  cbytes = cJSON_CreateNumber(length);




×







309



        RAW_NULL_CHECK(cbytes);





×







310



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));





×







311



      }










312



      RAW_RETURN_CHECK(setCompressOption(json, field->compress));





×







313



      break;




×







314



    }











UNCOV

315



    case TSDB_ALTER_TABLE_DROP_TAG:




×







316



    case TSDB_ALTER_TABLE_DROP_COLUMN: {











UNCOV

317



      TAOS_FIELD* field = taosArrayGet(req.pFields, 0);




×








UNCOV

318



      RAW_NULL_CHECK(field);





×








UNCOV

319



      cJSON* colName = cJSON_CreateString(field->name);




×








UNCOV

320



      RAW_NULL_CHECK(colName);





×








UNCOV

321



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));





×








UNCOV

322



      break;




×







323



    }











UNCOV

324



    case TSDB_ALTER_TABLE_UPDATE_TAG_BYTES:




×







325



    case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES: {











UNCOV

326



      TAOS_FIELD* field = taosArrayGet(req.pFields, 0);




×








UNCOV

327



      RAW_NULL_CHECK(field);





×








UNCOV

328



      cJSON* colName = cJSON_CreateString(field->name);




×








UNCOV

329



      RAW_NULL_CHECK(colName);





×








UNCOV

330



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));





×








UNCOV

331



      cJSON* colType = cJSON_CreateNumber(field->type);




×








UNCOV

332



      RAW_NULL_CHECK(colType);





×








UNCOV

333



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colType", colType));





×








UNCOV

334



      if (field->type == TSDB_DATA_TYPE_BINARY || field->type == TSDB_DATA_TYPE_VARBINARY ||





×








UNCOV

335



          field->type == TSDB_DATA_TYPE_GEOMETRY) {





×








UNCOV

336



        int32_t length = field->bytes - VARSTR_HEADER_SIZE;




×








UNCOV

337



        cJSON*  cbytes = cJSON_CreateNumber(length);




×








UNCOV

338



        RAW_NULL_CHECK(cbytes);





×








UNCOV

339



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));





×







340



      } else if (field->type == TSDB_DATA_TYPE_NCHAR) {





×







341



        int32_t length = (field->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;




×







342



        cJSON*  cbytes = cJSON_CreateNumber(length);




×







343



        RAW_NULL_CHECK(cbytes);





×







344



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));





×







345



      }











UNCOV

346



      break;




×







347



    }










348



    case TSDB_ALTER_TABLE_UPDATE_TAG_NAME:




×







349



    case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME: {










350



      TAOS_FIELD* oldField = taosArrayGet(req.pFields, 0);




×







351



      RAW_NULL_CHECK(oldField);





×







352



      TAOS_FIELD* newField = taosArrayGet(req.pFields, 1);




×







353



      RAW_NULL_CHECK(newField);





×







354



      cJSON* colName = cJSON_CreateString(oldField->name);




×







355



      RAW_NULL_CHECK(colName);





×







356



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));





×







357



      cJSON* colNewName = cJSON_CreateString(newField->name);




×







358



      RAW_NULL_CHECK(colNewName);





×







359



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colNewName", colNewName));





×







360



      break;




×







361



    }










362



    case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS: {




×







363



      TAOS_FIELD* field = taosArrayGet(req.pFields, 0);




×







364



      RAW_NULL_CHECK(field);





×







365



      cJSON* colName = cJSON_CreateString(field->name);




×







366



      RAW_NULL_CHECK(colName);





×







367



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));





×







368



      RAW_RETURN_CHECK(setCompressOption(json, field->bytes));





×







369



      break;




×







370



    }










371



    default:




×







372



      break;




×







373



  }










374













UNCOV

375



end:




×








UNCOV

376



  tFreeSMAltertbReq(&req);




×








UNCOV

377



  *pJson = json;




×







378



}










379













UNCOV

380



static void processCreateStb(SMqMetaRsp* metaRsp, cJSON** pJson) {




×








UNCOV

381



  if (metaRsp == NULL || pJson == NULL) {





×







382



    uError("invalid parameter in %s", __func__);





×







383



    return;




×







384



  }











UNCOV

385



  SVCreateStbReq req = {0};




×







386



  SDecoder       coder;










387













UNCOV

388



  uDebug("create stable data:%p", metaRsp);





×







389



  // decode and process req











UNCOV

390



  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));




×








UNCOV

391



  int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);




×








UNCOV

392



  tDecoderInit(&coder, data, len);




×







393













UNCOV

394



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





×







395



    goto end;




×







396



  }











UNCOV

397



  buildCreateTableJson(&req.schemaRow, &req.schemaTag, req.name, req.suid, TSDB_SUPER_TABLE, &req.colCmpr, pJson);




×







398













UNCOV

399



end:




×








UNCOV

400



  uDebug("create stable return");





×








UNCOV

401



  tDecoderClear(&coder);




×







402



}










403













UNCOV

404



static void processAlterStb(SMqMetaRsp* metaRsp, cJSON** pJson) {




×








UNCOV

405



  if (metaRsp == NULL || pJson == NULL) {





×







406



    uError("invalid parameter in %s", __func__);





×







407



    return;




×







408



  }











UNCOV

409



  SVCreateStbReq req = {0};




×








UNCOV

410



  SDecoder       coder = {0};




×








UNCOV

411



  uDebug("alter stable data:%p", metaRsp);





×







412












413



  // decode and process req











UNCOV

414



  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));




×








UNCOV

415



  int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);




×








UNCOV

416



  tDecoderInit(&coder, data, len);




×







417













UNCOV

418



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





×







419



    goto end;




×







420



  }











UNCOV

421



  buildAlterSTableJson(req.alterOriData, req.alterOriDataLen, pJson);




×







422













UNCOV

423



end:




×








UNCOV

424



  uDebug("alter stable return");





×








UNCOV

425



  tDecoderClear(&coder);




×







426



}










427













UNCOV

428



static void buildChildElement(cJSON* json, SVCreateTbReq* pCreateReq) {




×








UNCOV

429



  if (json == NULL || pCreateReq == NULL) {





×







430



    uError("invalid parameter in %s", __func__);





×







431



    return;




×







432



  }











UNCOV

433



  STag*   pTag = (STag*)pCreateReq->ctb.pTag;




×








UNCOV

434



  char*   sname = pCreateReq->ctb.stbName;




×








UNCOV

435



  char*   name = pCreateReq->name;




×








UNCOV

436



  SArray* tagName = pCreateReq->ctb.tagName;




×








UNCOV

437



  int64_t id = pCreateReq->uid;




×








UNCOV

438



  uint8_t tagNum = pCreateReq->ctb.tagNum;




×








UNCOV

439



  int32_t code = 0;




×








UNCOV

440



  SArray* pTagVals = NULL;




×








UNCOV

441



  char*   pJson = NULL;




×







442













UNCOV

443



  cJSON*  tableName = cJSON_CreateString(name);




×








UNCOV

444



  RAW_NULL_CHECK(tableName);





×








UNCOV

445



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableName", tableName));





×








UNCOV

446



  cJSON* using = cJSON_CreateString(sname);




×








UNCOV

447



  RAW_NULL_CHECK(using);





×








UNCOV

448



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "using", using));





×








UNCOV

449



  cJSON* tagNumJson = cJSON_CreateNumber(tagNum);




×








UNCOV

450



  RAW_NULL_CHECK(tagNumJson);





×








UNCOV

451



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tagNum", tagNumJson));





×







452













UNCOV

453



  cJSON* tags = cJSON_CreateArray();




×








UNCOV

454



  RAW_NULL_CHECK(tags);





×








UNCOV

455



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tags", tags));





×








UNCOV

456



  RAW_RETURN_CHECK(tTagToValArray(pTag, &pTagVals));





×








UNCOV

457



  if (tTagIsJson(pTag)) {





×








UNCOV

458



    STag* p = (STag*)pTag;




×








UNCOV

459



    if (p->nTag == 0) {





×








UNCOV

460



      uError("p->nTag == 0");





×








UNCOV

461



      goto end;




×







462



    }











UNCOV

463



    parseTagDatatoJson(pTag, &pJson, NULL);




×








UNCOV

464



    RAW_NULL_CHECK(pJson);





×








UNCOV

465



    cJSON* tag = cJSON_CreateObject();




×








UNCOV

466



    RAW_NULL_CHECK(tag);





×








UNCOV

467



    RAW_FALSE_CHECK(tmqAddJsonArrayItem(tags, tag));





×








UNCOV

468



    STagVal* pTagVal = taosArrayGet(pTagVals, 0);




×








UNCOV

469



    RAW_NULL_CHECK(pTagVal);





×








UNCOV

470



    char* ptname = taosArrayGet(tagName, 0);




×








UNCOV

471



    RAW_NULL_CHECK(ptname);





×








UNCOV

472



    cJSON* tname = cJSON_CreateString(ptname);




×








UNCOV

473



    RAW_NULL_CHECK(tname);





×








UNCOV

474



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "name", tname));





×








UNCOV

475



    cJSON* ttype = cJSON_CreateNumber(TSDB_DATA_TYPE_JSON);




×








UNCOV

476



    RAW_NULL_CHECK(ttype);





×








UNCOV

477



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "type", ttype));





×








UNCOV

478



    cJSON* tvalue = cJSON_CreateString(pJson);




×








UNCOV

479



    RAW_NULL_CHECK(tvalue);





×








UNCOV

480



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "value", tvalue));





×








UNCOV

481



    goto end;




×







482



  }










483













UNCOV

484



  for (int i = 0; i < taosArrayGetSize(pTagVals); i++) {





×








UNCOV

485



    STagVal* pTagVal = (STagVal*)taosArrayGet(pTagVals, i);




×








UNCOV

486



    RAW_NULL_CHECK(pTagVal);





×








UNCOV

487



    cJSON* tag = cJSON_CreateObject();




×








UNCOV

488



    RAW_NULL_CHECK(tag);





×








UNCOV

489



    RAW_FALSE_CHECK(tmqAddJsonArrayItem(tags, tag));





×








UNCOV

490



    char* ptname = taosArrayGet(tagName, i);




×








UNCOV

491



    RAW_NULL_CHECK(ptname);





×








UNCOV

492



    cJSON* tname = cJSON_CreateString(ptname);




×








UNCOV

493



    RAW_NULL_CHECK(tname);





×








UNCOV

494



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "name", tname));





×








UNCOV

495



    cJSON* ttype = cJSON_CreateNumber(pTagVal->type);




×








UNCOV

496



    RAW_NULL_CHECK(ttype);





×








UNCOV

497



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "type", ttype));





×







498













UNCOV

499



    cJSON* tvalue = NULL;




×








UNCOV

500



    if (IS_VAR_DATA_TYPE(pTagVal->type)) {





×








UNCOV

501



      int64_t bufSize = 0;




×








UNCOV

502



      if (pTagVal->type == TSDB_DATA_TYPE_VARBINARY) {





×







503



        bufSize = pTagVal->nData * 2 + 2 + 3;




×







504



      } else {











UNCOV

505



        bufSize = pTagVal->nData + 3;




×







506



      }











UNCOV

507



      char* buf = taosMemoryCalloc(bufSize, 1);





×








UNCOV

508



      RAW_NULL_CHECK(buf);





×








UNCOV

509



      if (dataConverToStr(buf, bufSize, pTagVal->type, pTagVal->pData, pTagVal->nData, NULL) != TSDB_CODE_SUCCESS) {





×







510



        taosMemoryFree(buf);





×







511



        goto end;




×







512



      }










513













UNCOV

514



      tvalue = cJSON_CreateString(buf);




×








UNCOV

515



      taosMemoryFree(buf);





×








UNCOV

516



      RAW_NULL_CHECK(tvalue);





×







517



    } else {











UNCOV

518



      double val = 0;




×








UNCOV

519



      GET_TYPED_DATA(val, double, pTagVal->type, &pTagVal->i64);





×








UNCOV

520



      tvalue = cJSON_CreateNumber(val);




×








UNCOV

521



      RAW_NULL_CHECK(tvalue);





×







522



    }










523













UNCOV

524



    RAW_FALSE_CHECK(tmqAddJsonObjectItem(tag, "value", tvalue));





×







525



  }










526













UNCOV

527



end:




×








UNCOV

528



  taosMemoryFree(pJson);





×








UNCOV

529



  taosArrayDestroy(pTagVals);




×







530



}










531













UNCOV

532



static void buildCreateCTableJson(SVCreateTbReq* pCreateReq, int32_t nReqs, cJSON** pJson) {




×








UNCOV

533



  if (pJson == NULL || pCreateReq == NULL) {





×







534



    uError("invalid parameter in %s", __func__);





×







535



    return;




×







536



  }











UNCOV

537



  int32_t code = 0;




×








UNCOV

538



  char*   string = NULL;




×








UNCOV

539



  cJSON*  json = cJSON_CreateObject();




×








UNCOV

540



  RAW_NULL_CHECK(json);





×








UNCOV

541



  cJSON* type = cJSON_CreateString("create");




×








UNCOV

542



  RAW_NULL_CHECK(type);





×








UNCOV

543



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "type", type));





×







544













UNCOV

545



  cJSON* tableType = cJSON_CreateString("child");




×








UNCOV

546



  RAW_NULL_CHECK(tableType);





×








UNCOV

547



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableType", tableType));





×







548













UNCOV

549



  buildChildElement(json, pCreateReq);




×








UNCOV

550



  cJSON* createList = cJSON_CreateArray();




×








UNCOV

551



  RAW_NULL_CHECK(createList);





×








UNCOV

552



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "createList", createList));





×







553













UNCOV

554



  for (int i = 0; nReqs > 1 && i < nReqs; i++) {





×








UNCOV

555



    cJSON* create = cJSON_CreateObject();




×








UNCOV

556



    RAW_NULL_CHECK(create);





×








UNCOV

557



    buildChildElement(create, pCreateReq + i);




×








UNCOV

558



    RAW_FALSE_CHECK(tmqAddJsonArrayItem(createList, create));





×







559



  }










560













UNCOV

561



end:




×








UNCOV

562



  *pJson = json;




×







563



}










564













UNCOV

565



static void processCreateTable(SMqMetaRsp* metaRsp, cJSON** pJson) {




×








UNCOV

566



  if (pJson == NULL || metaRsp == NULL) {





×







567



    uError("invalid parameter in %s", __func__);





×







568



    return;




×







569



  }











UNCOV

570



  SDecoder           decoder = {0};




×








UNCOV

571



  SVCreateTbBatchReq req = {0};




×







572



  SVCreateTbReq*     pCreateReq;










573



  // decode











UNCOV

574



  uDebug("create table data:%p", metaRsp);





×








UNCOV

575



  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));




×








UNCOV

576



  int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);




×








UNCOV

577



  tDecoderInit(&decoder, data, len);




×








UNCOV

578



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





×







579



    goto end;




×







580



  }










581












582



  // loop to create table











UNCOV

583



  if (req.nReqs > 0) {





×








UNCOV

584



    pCreateReq = req.pReqs;




×








UNCOV

585



    if (pCreateReq->type == TSDB_CHILD_TABLE) {





×








UNCOV

586



      buildCreateCTableJson(req.pReqs, req.nReqs, pJson);




×








UNCOV

587



    } else if (pCreateReq->type == TSDB_NORMAL_TABLE) {





×








UNCOV

588



      buildCreateTableJson(&pCreateReq->ntb.schemaRow, NULL, pCreateReq->name, pCreateReq->uid, TSDB_NORMAL_TABLE,




×







589



                           &pCreateReq->colCmpr, pJson);










590



    }










591



  }










592












593



end:




×








UNCOV

594



  uDebug("create table return");





×








UNCOV

595



  tDeleteSVCreateTbBatchReq(&req);




×








UNCOV

596



  tDecoderClear(&decoder);




×







597



}










598













UNCOV

599



static void processAutoCreateTable(SMqDataRsp* rsp, char** string) {




×








UNCOV

600



  if (rsp == NULL || string == NULL) {





×







601



    uError("invalid parameter in %s", __func__);





×







602



    return;




×







603



  }











UNCOV

604



  SDecoder*      decoder = NULL;




×








UNCOV

605



  SVCreateTbReq* pCreateReq = NULL;




×








UNCOV

606



  int32_t        code = 0;




×








UNCOV

607



  uDebug("auto create table data:%p", rsp);





×








UNCOV

608



  if (rsp->createTableNum <= 0) {





×







609



    uError("processAutoCreateTable rsp->createTableNum <= 0");





×







610



    goto end;




×







611



  }










612













UNCOV

613



  decoder = taosMemoryCalloc(rsp->createTableNum, sizeof(SDecoder));





×








UNCOV

614



  RAW_NULL_CHECK(decoder);





×








UNCOV

615



  pCreateReq = taosMemoryCalloc(rsp->createTableNum, sizeof(SVCreateTbReq));





×








UNCOV

616



  RAW_NULL_CHECK(pCreateReq);





×







617












618



  // loop to create table











UNCOV

619



  for (int32_t iReq = 0; iReq < rsp->createTableNum; iReq++) {





×







620



    // decode











UNCOV

621



    void** data = taosArrayGet(rsp->createTableReq, iReq);




×








UNCOV

622



    RAW_NULL_CHECK(data);





×








UNCOV

623



    int32_t* len = taosArrayGet(rsp->createTableLen, iReq);




×








UNCOV

624



    RAW_NULL_CHECK(len);





×








UNCOV

625



    tDecoderInit(&decoder[iReq], *data, *len);




×








UNCOV

626



    if (tDecodeSVCreateTbReq(&decoder[iReq], pCreateReq + iReq) < 0) {





×







627



      goto end;




×







628



    }










629













UNCOV

630



    if (pCreateReq[iReq].type != TSDB_CHILD_TABLE) {





×







631



      uError("processAutoCreateTable pCreateReq[iReq].type != TSDB_CHILD_TABLE");





×







632



      goto end;




×







633



    }










634



  }











UNCOV

635



  cJSON* pJson = NULL;




×








UNCOV

636



  buildCreateCTableJson(pCreateReq, rsp->createTableNum, &pJson);




×








UNCOV

637



  *string = cJSON_PrintUnformatted(pJson);




×








UNCOV

638



  cJSON_Delete(pJson);




×







639













UNCOV

640



end:




×








UNCOV

641



  uDebug("auto created table return, sql json:%s", *string);





×








UNCOV

642



  for (int i = 0; decoder && pCreateReq && i < rsp->createTableNum; i++) {





×








UNCOV

643



    tDecoderClear(&decoder[i]);




×








UNCOV

644



    taosMemoryFreeClear(pCreateReq[i].comment);





×








UNCOV

645



    if (pCreateReq[i].type == TSDB_CHILD_TABLE) {





×








UNCOV

646



      taosArrayDestroy(pCreateReq[i].ctb.tagName);




×







647



    }










648



  }











UNCOV

649



  taosMemoryFree(decoder);





×








UNCOV

650



  taosMemoryFree(pCreateReq);





×







651



}










652













UNCOV

653



static void processAlterTable(SMqMetaRsp* metaRsp, cJSON** pJson) {




×








UNCOV

654



  if (pJson == NULL || metaRsp == NULL) {





×







655



    uError("invalid parameter in %s", __func__);





×







656



    return;




×







657



  }











UNCOV

658



  SDecoder     decoder = {0};




×








UNCOV

659



  SVAlterTbReq vAlterTbReq = {0};




×








UNCOV

660



  char*        string = NULL;




×








UNCOV

661



  cJSON*       json = NULL;




×








UNCOV

662



  int32_t      code = 0;




×







663













UNCOV

664



  uDebug("alter table data:%p", metaRsp);





×







665



  // decode











UNCOV

666



  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));




×








UNCOV

667



  int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);




×








UNCOV

668



  tDecoderInit(&decoder, data, len);




×








UNCOV

669



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





×







670



    uError("tDecodeSVAlterTbReq error");





×







671



    goto end;




×







672



  }










673













UNCOV

674



  json = cJSON_CreateObject();




×








UNCOV

675



  RAW_NULL_CHECK(json);





×








UNCOV

676



  cJSON* type = cJSON_CreateString("alter");




×








UNCOV

677



  RAW_NULL_CHECK(type);





×








UNCOV

678



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "type", type));





×








UNCOV

679



  cJSON* tableType = cJSON_CreateString(vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_TAG_VAL ||





×








UNCOV

680



                                                vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_MULTI_TAG_VAL





×







681



                                            ? "child"










682



                                            : "normal");











UNCOV

683



  RAW_NULL_CHECK(tableType);





×








UNCOV

684



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableType", tableType));





×








UNCOV

685



  cJSON* tableName = cJSON_CreateString(vAlterTbReq.tbName);




×








UNCOV

686



  RAW_NULL_CHECK(tableName);





×








UNCOV

687



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableName", tableName));





×








UNCOV

688



  cJSON* alterType = cJSON_CreateNumber(vAlterTbReq.action);




×








UNCOV

689



  RAW_NULL_CHECK(alterType);





×








UNCOV

690



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "alterType", alterType));





×







691













UNCOV

692



  switch (vAlterTbReq.action) {





×








UNCOV

693



    case TSDB_ALTER_TABLE_ADD_COLUMN: {




×








UNCOV

694



      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);




×








UNCOV

695



      RAW_NULL_CHECK(colName);





×








UNCOV

696



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));





×








UNCOV

697



      cJSON* colType = cJSON_CreateNumber(vAlterTbReq.type);




×








UNCOV

698



      RAW_NULL_CHECK(colType);





×








UNCOV

699



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colType", colType));





×







700













UNCOV

701



      if (vAlterTbReq.type == TSDB_DATA_TYPE_BINARY || vAlterTbReq.type == TSDB_DATA_TYPE_VARBINARY ||





×








UNCOV

702



          vAlterTbReq.type == TSDB_DATA_TYPE_GEOMETRY) {





×







703



        int32_t length = vAlterTbReq.bytes - VARSTR_HEADER_SIZE;




×







704



        cJSON*  cbytes = cJSON_CreateNumber(length);




×







705



        RAW_NULL_CHECK(cbytes);





×







706



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));





×








UNCOV

707



      } else if (vAlterTbReq.type == TSDB_DATA_TYPE_NCHAR) {





×







708



        int32_t length = (vAlterTbReq.bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;




×







709



        cJSON*  cbytes = cJSON_CreateNumber(length);




×







710



        RAW_NULL_CHECK(cbytes);





×







711



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));





×







712



      }











UNCOV

713



      break;




×







714



    }










715



    case TSDB_ALTER_TABLE_ADD_COLUMN_WITH_COMPRESS_OPTION: {




×







716



      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);




×







717



      RAW_NULL_CHECK(colName);





×







718



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));





×







719



      cJSON* colType = cJSON_CreateNumber(vAlterTbReq.type);




×







720



      RAW_NULL_CHECK(colType);





×







721



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colType", colType));





×







722












723



      if (vAlterTbReq.type == TSDB_DATA_TYPE_BINARY || vAlterTbReq.type == TSDB_DATA_TYPE_VARBINARY ||





×







724



          vAlterTbReq.type == TSDB_DATA_TYPE_GEOMETRY) {





×







725



        int32_t length = vAlterTbReq.bytes - VARSTR_HEADER_SIZE;




×







726



        cJSON*  cbytes = cJSON_CreateNumber(length);




×







727



        RAW_NULL_CHECK(cbytes);





×







728



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));





×







729



      } else if (vAlterTbReq.type == TSDB_DATA_TYPE_NCHAR) {





×







730



        int32_t length = (vAlterTbReq.bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;




×







731



        cJSON*  cbytes = cJSON_CreateNumber(length);




×







732



        RAW_NULL_CHECK(cbytes);





×







733



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));





×







734



      }










735



      RAW_RETURN_CHECK(setCompressOption(json, vAlterTbReq.compress));





×







736



      break;




×







737



    }











UNCOV

738



    case TSDB_ALTER_TABLE_DROP_COLUMN: {




×








UNCOV

739



      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);




×








UNCOV

740



      RAW_NULL_CHECK(colName);





×








UNCOV

741



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));





×








UNCOV

742



      break;




×







743



    }











UNCOV

744



    case TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES: {




×








UNCOV

745



      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);




×








UNCOV

746



      RAW_NULL_CHECK(colName);





×








UNCOV

747



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));





×








UNCOV

748



      cJSON* colType = cJSON_CreateNumber(vAlterTbReq.colModType);




×








UNCOV

749



      RAW_NULL_CHECK(colType);





×








UNCOV

750



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colType", colType));





×








UNCOV

751



      if (vAlterTbReq.colModType == TSDB_DATA_TYPE_BINARY || vAlterTbReq.colModType == TSDB_DATA_TYPE_VARBINARY ||





×








UNCOV

752



          vAlterTbReq.colModType == TSDB_DATA_TYPE_GEOMETRY) {





×







753



        int32_t length = vAlterTbReq.colModBytes - VARSTR_HEADER_SIZE;




×







754



        cJSON*  cbytes = cJSON_CreateNumber(length);




×







755



        RAW_NULL_CHECK(cbytes);





×







756



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));





×








UNCOV

757



      } else if (vAlterTbReq.colModType == TSDB_DATA_TYPE_NCHAR) {





×








UNCOV

758



        int32_t length = (vAlterTbReq.colModBytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;




×








UNCOV

759



        cJSON*  cbytes = cJSON_CreateNumber(length);




×








UNCOV

760



        RAW_NULL_CHECK(cbytes);





×








UNCOV

761



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colLength", cbytes));





×







762



      }











UNCOV

763



      break;




×







764



    }











UNCOV

765



    case TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME: {




×








UNCOV

766



      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);




×








UNCOV

767



      RAW_NULL_CHECK(colName);





×








UNCOV

768



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));





×








UNCOV

769



      cJSON* colNewName = cJSON_CreateString(vAlterTbReq.colNewName);




×








UNCOV

770



      RAW_NULL_CHECK(colNewName);





×








UNCOV

771



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colNewName", colNewName));





×








UNCOV

772



      break;




×







773



    }











UNCOV

774



    case TSDB_ALTER_TABLE_UPDATE_TAG_VAL: {




×








UNCOV

775



      cJSON* tagName = cJSON_CreateString(vAlterTbReq.tagName);




×








UNCOV

776



      RAW_NULL_CHECK(tagName);





×








UNCOV

777



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", tagName));





×







778













UNCOV

779



      bool isNull = vAlterTbReq.isNull;




×








UNCOV

780



      if (vAlterTbReq.tagType == TSDB_DATA_TYPE_JSON) {





×







781



        STag* jsonTag = (STag*)vAlterTbReq.pTagVal;




×







782



        if (jsonTag->nTag == 0) isNull = true;





×







783



      }











UNCOV

784



      if (!isNull) {





×








UNCOV

785



        char* buf = NULL;




×







786













UNCOV

787



        if (vAlterTbReq.tagType == TSDB_DATA_TYPE_JSON) {





×







788



          if (!tTagIsJson(vAlterTbReq.pTagVal)) {





×







789



            uError("processAlterTable isJson false");





×







790



            goto end;




×







791



          }










792



          parseTagDatatoJson(vAlterTbReq.pTagVal, &buf, NULL);




×







793



          if (buf == NULL) {





×







794



            uError("parseTagDatatoJson failed, buf == NULL");





×







795



            goto end;




×







796



          }










797



        } else {











UNCOV

798



          int64_t bufSize = 0;




×








UNCOV

799



          if (vAlterTbReq.tagType == TSDB_DATA_TYPE_VARBINARY) {





×







800



            bufSize = vAlterTbReq.nTagVal * 2 + 2 + 3;




×







801



          } else {











UNCOV

802



            bufSize = vAlterTbReq.nTagVal + 3;




×







803



          }











UNCOV

804



          buf = taosMemoryCalloc(bufSize, 1);





×








UNCOV

805



          RAW_NULL_CHECK(buf);





×








UNCOV

806



          if (dataConverToStr(buf, bufSize, vAlterTbReq.tagType, vAlterTbReq.pTagVal, vAlterTbReq.nTagVal, NULL) !=





×







807



              TSDB_CODE_SUCCESS) {










808



            taosMemoryFree(buf);





×







809



            goto end;




×







810



          }










811



        }










812













UNCOV

813



        cJSON* colValue = cJSON_CreateString(buf);




×








UNCOV

814



        taosMemoryFree(buf);





×








UNCOV

815



        RAW_NULL_CHECK(colValue);





×








UNCOV

816



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colValue", colValue));





×







817



      }










818













UNCOV

819



      cJSON* isNullCJson = cJSON_CreateBool(isNull);




×








UNCOV

820



      RAW_NULL_CHECK(isNullCJson);





×








UNCOV

821



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colValueNull", isNullCJson));





×








UNCOV

822



      break;




×







823



    }










824



    case TSDB_ALTER_TABLE_UPDATE_MULTI_TAG_VAL: {




×







825



      int32_t nTags = taosArrayGetSize(vAlterTbReq.pMultiTag);




×







826



      if (nTags <= 0) {





×







827



        uError("processAlterTable parse multi tags error");





×







828



        goto end;




×







829



      }










830












831



      cJSON* tags = cJSON_CreateArray();




×







832



      RAW_NULL_CHECK(tags);





×







833



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tags", tags));





×







834












835



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





×







836



        cJSON* member = cJSON_CreateObject();




×







837



        RAW_NULL_CHECK(member);





×







838



        RAW_FALSE_CHECK(tmqAddJsonArrayItem(tags, member));





×







839












840



        SMultiTagUpateVal* pTagVal = taosArrayGet(vAlterTbReq.pMultiTag, i);




×







841



        cJSON*             tagName = cJSON_CreateString(pTagVal->tagName);




×







842



        RAW_NULL_CHECK(tagName);





×







843



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(member, "colName", tagName));





×







844












845



        if (pTagVal->tagType == TSDB_DATA_TYPE_JSON) {





×







846



          uError("processAlterTable isJson false");





×







847



          goto end;




×







848



        }










849



        bool isNull = pTagVal->isNull;




×







850



        if (!isNull) {





×







851



          int64_t bufSize = 0;




×







852



          if (pTagVal->tagType == TSDB_DATA_TYPE_VARBINARY) {





×







853



            bufSize = pTagVal->nTagVal * 2 + 2 + 3;




×







854



          } else {










855



            bufSize = pTagVal->nTagVal + 3;




×







856



          }










857



          char* buf = taosMemoryCalloc(bufSize, 1);





×







858



          RAW_NULL_CHECK(buf);





×







859



          if (dataConverToStr(buf, bufSize, pTagVal->tagType, pTagVal->pTagVal, pTagVal->nTagVal, NULL) !=





×







860



              TSDB_CODE_SUCCESS) {










861



            taosMemoryFree(buf);





×







862



            goto end;




×







863



          }










864



          cJSON* colValue = cJSON_CreateString(buf);




×







865



          taosMemoryFree(buf);





×







866



          RAW_NULL_CHECK(colValue);





×







867



          RAW_FALSE_CHECK(tmqAddJsonObjectItem(member, "colValue", colValue));





×







868



        }










869



        cJSON* isNullCJson = cJSON_CreateBool(isNull);




×







870



        RAW_NULL_CHECK(isNullCJson);





×







871



        RAW_FALSE_CHECK(tmqAddJsonObjectItem(member, "colValueNull", isNullCJson));





×







872



      }










873



      break;




×







874



    }










875












876



    case TSDB_ALTER_TABLE_UPDATE_COLUMN_COMPRESS: {




×







877



      cJSON* colName = cJSON_CreateString(vAlterTbReq.colName);




×







878



      RAW_NULL_CHECK(colName);





×







879



      RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "colName", colName));





×







880



      RAW_RETURN_CHECK(setCompressOption(json, vAlterTbReq.compress));





×







881



      break;




×







882



    }











UNCOV

883



    default:




×








UNCOV

884



      break;




×







885



  }










886













UNCOV

887



end:




×








UNCOV

888



  uDebug("alter table return");





×








UNCOV

889



  if (vAlterTbReq.action == TSDB_ALTER_TABLE_UPDATE_MULTI_TAG_VAL) {





×







890



    taosArrayDestroy(vAlterTbReq.pMultiTag);




×







891



  }











UNCOV

892



  tDecoderClear(&decoder);




×








UNCOV

893



  *pJson = json;




×







894



}










895













UNCOV

896



static void processDropSTable(SMqMetaRsp* metaRsp, cJSON** pJson) {




×








UNCOV

897



  if (pJson == NULL || metaRsp == NULL) {





×







898



    uError("invalid parameter in %s", __func__);





×







899



    return;




×







900



  }











UNCOV

901



  SDecoder     decoder = {0};




×








UNCOV

902



  SVDropStbReq req = {0};




×








UNCOV

903



  cJSON*       json = NULL;




×








UNCOV

904



  int32_t      code = 0;




×







905













UNCOV

906



  uDebug("processDropSTable data:%p", metaRsp);





×







907












908



  // decode











UNCOV

909



  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));




×








UNCOV

910



  int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);




×








UNCOV

911



  tDecoderInit(&decoder, data, len);




×








UNCOV

912



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





×







913



    uError("tDecodeSVDropStbReq failed");





×







914



    goto end;




×







915



  }










916













UNCOV

917



  json = cJSON_CreateObject();




×








UNCOV

918



  RAW_NULL_CHECK(json);





×








UNCOV

919



  cJSON* type = cJSON_CreateString("drop");




×








UNCOV

920



  RAW_NULL_CHECK(type);





×








UNCOV

921



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "type", type));





×








UNCOV

922



  cJSON* tableType = cJSON_CreateString("super");




×








UNCOV

923



  RAW_NULL_CHECK(tableType);





×








UNCOV

924



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableType", tableType));





×








UNCOV

925



  cJSON* tableName = cJSON_CreateString(req.name);




×








UNCOV

926



  RAW_NULL_CHECK(tableName);





×








UNCOV

927



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableName", tableName));





×







928













UNCOV

929



end:




×








UNCOV

930



  uDebug("processDropSTable return");





×








UNCOV

931



  tDecoderClear(&decoder);




×








UNCOV

932



  *pJson = json;




×







933



}











UNCOV

934



static void processDeleteTable(SMqMetaRsp* metaRsp, cJSON** pJson) {




×








UNCOV

935



  if (pJson == NULL || metaRsp == NULL) {





×







936



    uError("invalid parameter in %s", __func__);





×







937



    return;




×







938



  }











UNCOV

939



  SDeleteRes req = {0};




×








UNCOV

940



  SDecoder   coder = {0};




×








UNCOV

941



  cJSON*     json = NULL;




×








UNCOV

942



  int32_t    code = 0;




×







943













UNCOV

944



  uDebug("processDeleteTable data:%p", metaRsp);





×







945



  // decode and process req











UNCOV

946



  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));




×








UNCOV

947



  int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);




×







948













UNCOV

949



  tDecoderInit(&coder, data, len);




×








UNCOV

950



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





×







951



    uError("tDecodeDeleteRes failed");





×







952



    goto end;




×







953



  }










954












955



  //  getTbName(req.tableFName);











UNCOV

956



  char sql[256] = {0};




×








UNCOV

957



  (void)snprintf(sql, sizeof(sql), "delete from `%s` where `%s` >= %" PRId64 " and `%s` <= %" PRId64, req.tableFName,




×







958



                 req.tsColName, req.skey, req.tsColName, req.ekey);










959













UNCOV

960



  json = cJSON_CreateObject();




×








UNCOV

961



  RAW_NULL_CHECK(json);





×








UNCOV

962



  cJSON* type = cJSON_CreateString("delete");




×








UNCOV

963



  RAW_NULL_CHECK(type);





×








UNCOV

964



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "type", type));





×








UNCOV

965



  cJSON* sqlJson = cJSON_CreateString(sql);




×








UNCOV

966



  RAW_NULL_CHECK(sqlJson);





×








UNCOV

967



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "sql", sqlJson));





×







968













UNCOV

969



end:




×








UNCOV

970



  uDebug("processDeleteTable return");





×








UNCOV

971



  tDecoderClear(&coder);




×








UNCOV

972



  *pJson = json;




×







973



}










974













UNCOV

975



static void processDropTable(SMqMetaRsp* metaRsp, cJSON** pJson) {




×








UNCOV

976



  if (pJson == NULL || metaRsp == NULL) {





×







977



    uError("invalid parameter in %s", __func__);





×







978



    return;




×







979



  }











UNCOV

980



  SDecoder         decoder = {0};




×








UNCOV

981



  SVDropTbBatchReq req = {0};




×








UNCOV

982



  cJSON*           json = NULL;




×








UNCOV

983



  int32_t          code = 0;




×







984













UNCOV

985



  uDebug("processDropTable data:%p", metaRsp);





×







986



  // decode











UNCOV

987



  void*   data = POINTER_SHIFT(metaRsp->metaRsp, sizeof(SMsgHead));




×








UNCOV

988



  int32_t len = metaRsp->metaRspLen - sizeof(SMsgHead);




×








UNCOV

989



  tDecoderInit(&decoder, data, len);




×








UNCOV

990



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





×







991



    uError("tDecodeSVDropTbBatchReq failed");





×







992



    goto end;




×







993



  }










994













UNCOV

995



  json = cJSON_CreateObject();




×








UNCOV

996



  RAW_NULL_CHECK(json);





×








UNCOV

997



  cJSON* type = cJSON_CreateString("drop");




×








UNCOV

998



  RAW_NULL_CHECK(type);





×








UNCOV

999



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "type", type));





×








UNCOV

1000



  cJSON* tableNameList = cJSON_CreateArray();




×








UNCOV

1001



  RAW_NULL_CHECK(tableNameList);





×








UNCOV

1002



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(json, "tableNameList", tableNameList));





×







1003













UNCOV

1004



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





×








UNCOV

1005



    SVDropTbReq* pDropTbReq = req.pReqs + iReq;




×








UNCOV

1006



    cJSON*       tableName = cJSON_CreateString(pDropTbReq->name);




×








UNCOV

1007



    RAW_NULL_CHECK(tableName);





×








UNCOV

1008



    RAW_FALSE_CHECK(tmqAddJsonArrayItem(tableNameList, tableName));





×







1009



  }










1010













UNCOV

1011



end:




×








UNCOV

1012



  uDebug("processDropTable return");





×








UNCOV

1013



  tDecoderClear(&decoder);




×








UNCOV

1014



  *pJson = json;




×







1015



}










1016













UNCOV

1017



static int32_t taosCreateStb(TAOS* taos, void* meta, uint32_t metaLen) {




×








UNCOV

1018



  if (taos == NULL || meta == NULL) {





×







1019



    uError("invalid parameter in %s", __func__);





×







1020



    return TSDB_CODE_INVALID_PARA;




×







1021



  }











UNCOV

1022



  SVCreateStbReq req = {0};




×







1023



  SDecoder       coder;











UNCOV

1024



  SMCreateStbReq pReq = {0};




×








UNCOV

1025



  int32_t        code = TSDB_CODE_SUCCESS;




×








UNCOV

1026



  SRequestObj*   pRequest = NULL;




×







1027













UNCOV

1028



  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest, 0));





×








UNCOV

1029



  uDebug(LOG_ID_TAG " create stable, meta:%p, metaLen:%d", LOG_ID_VALUE, meta, metaLen);





×








UNCOV

1030



  pRequest->syncQuery = true;




×








UNCOV

1031



  if (!pRequest->pDb) {





×







1032



    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;




×







1033



    goto end;




×







1034



  }










1035



  // decode and process req











UNCOV

1036



  void*    data = POINTER_SHIFT(meta, sizeof(SMsgHead));




×








UNCOV

1037



  uint32_t len = metaLen - sizeof(SMsgHead);




×








UNCOV

1038



  tDecoderInit(&coder, data, len);




×








UNCOV

1039



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





×







1040



    code = TSDB_CODE_INVALID_PARA;




×







1041



    goto end;




×







1042



  }










1043













UNCOV

1044



  int8_t           createDefaultCompress = 0;




×








UNCOV

1045



  SColCmprWrapper* p = &req.colCmpr;




×








UNCOV

1046



  if (p->nCols == 0) {





×







1047



    createDefaultCompress = 1;




×







1048



  }










1049



  // build create stable











UNCOV

1050



  pReq.pColumns = taosArrayInit(req.schemaRow.nCols, sizeof(SFieldWithOptions));




×








UNCOV

1051



  RAW_NULL_CHECK(pReq.pColumns);





×








UNCOV

1052



  for (int32_t i = 0; i < req.schemaRow.nCols; i++) {





×








UNCOV

1053



    SSchema*          pSchema = req.schemaRow.pSchema + i;




×








UNCOV

1054



    SFieldWithOptions field = {.type = pSchema->type, .flags = pSchema->flags, .bytes = pSchema->bytes};




×








UNCOV

1055



    tstrncpy(field.name, pSchema->name, TSDB_COL_NAME_LEN);




×







1056













UNCOV

1057



    if (createDefaultCompress) {





×







1058



      field.compress = createDefaultColCmprByType(pSchema->type);




×







1059



    } else {











UNCOV

1060



      SColCmpr* pCmp = &req.colCmpr.pColCmpr[i];




×








UNCOV

1061



      field.compress = pCmp->alg;




×







1062



    }











UNCOV

1063



    RAW_NULL_CHECK(taosArrayPush(pReq.pColumns, &field));





×







1064



  }











UNCOV

1065



  pReq.pTags = taosArrayInit(req.schemaTag.nCols, sizeof(SField));




×








UNCOV

1066



  RAW_NULL_CHECK(pReq.pTags);





×








UNCOV

1067



  for (int32_t i = 0; i < req.schemaTag.nCols; i++) {





×








UNCOV

1068



    SSchema* pSchema = req.schemaTag.pSchema + i;




×








UNCOV

1069



    SField   field = {.type = pSchema->type, .flags = pSchema->flags, .bytes = pSchema->bytes};




×








UNCOV

1070



    tstrncpy(field.name, pSchema->name, TSDB_COL_NAME_LEN);




×








UNCOV

1071



    RAW_NULL_CHECK(taosArrayPush(pReq.pTags, &field));





×







1072



  }










1073













UNCOV

1074



  pReq.colVer = req.schemaRow.version;




×








UNCOV

1075



  pReq.tagVer = req.schemaTag.version;




×








UNCOV

1076



  pReq.numOfColumns = req.schemaRow.nCols;




×








UNCOV

1077



  pReq.numOfTags = req.schemaTag.nCols;




×








UNCOV

1078



  pReq.commentLen = -1;




×








UNCOV

1079



  pReq.suid = processSuid(req.suid, pRequest->pDb);




×








UNCOV

1080



  pReq.source = TD_REQ_FROM_TAOX;




×








UNCOV

1081



  pReq.igExists = true;




×







1082













UNCOV

1083



  uDebug(LOG_ID_TAG " create stable name:%s suid:%" PRId64 " processSuid:%" PRId64, LOG_ID_VALUE, req.name, req.suid,





×







1084



         pReq.suid);











UNCOV

1085



  STscObj* pTscObj = pRequest->pTscObj;




×








UNCOV

1086



  SName    tableName = {0};




×








UNCOV

1087



  toName(pTscObj->acctId, pRequest->pDb, req.name, &tableName);




×








UNCOV

1088



  RAW_RETURN_CHECK(tNameExtractFullName(&tableName, pReq.name));





×








UNCOV

1089



  SCmdMsgInfo pCmdMsg = {0};




×








UNCOV

1090



  pCmdMsg.epSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);




×








UNCOV

1091



  pCmdMsg.msgType = TDMT_MND_CREATE_STB;




×








UNCOV

1092



  pCmdMsg.msgLen = tSerializeSMCreateStbReq(NULL, 0, &pReq);




×








UNCOV

1093



  if (pCmdMsg.msgLen <= 0) {





×







1094



    code = TSDB_CODE_INVALID_PARA;




×







1095



    goto end;




×







1096



  }











UNCOV

1097



  pCmdMsg.pMsg = taosMemoryMalloc(pCmdMsg.msgLen);





×








UNCOV

1098



  RAW_NULL_CHECK(pCmdMsg.pMsg);





×








UNCOV

1099



  if (tSerializeSMCreateStbReq(pCmdMsg.pMsg, pCmdMsg.msgLen, &pReq) <= 0) {





×







1100



    code = TSDB_CODE_INVALID_PARA;




×







1101



    taosMemoryFree(pCmdMsg.pMsg);





×







1102



    goto end;




×







1103



  }










1104













UNCOV

1105



  SQuery pQuery = {0};




×








UNCOV

1106



  pQuery.execMode = QUERY_EXEC_MODE_RPC;




×








UNCOV

1107



  pQuery.pCmdMsg = &pCmdMsg;




×








UNCOV

1108



  pQuery.msgType = pQuery.pCmdMsg->msgType;




×








UNCOV

1109



  pQuery.stableQuery = true;




×







1110













UNCOV

1111



  launchQueryImpl(pRequest, &pQuery, true, NULL);  // ignore, because return value is pRequest




×







1112













UNCOV

1113



  taosMemoryFree(pCmdMsg.pMsg);





×







1114













UNCOV

1115



  if (pRequest->code == TSDB_CODE_SUCCESS) {





×








UNCOV

1116



    SCatalog* pCatalog = NULL;




×








UNCOV

1117



    RAW_RETURN_CHECK(catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog));





×








UNCOV

1118



    RAW_RETURN_CHECK(catalogRemoveTableMeta(pCatalog, &tableName));





×







1119



  }










1120













UNCOV

1121



  code = pRequest->code;




×







1122













UNCOV

1123



end:




×








UNCOV

1124



  uDebug(LOG_ID_TAG " create stable return, msg:%s", LOG_ID_VALUE, tstrerror(code));





×








UNCOV

1125



  destroyRequest(pRequest);




×








UNCOV

1126



  tFreeSMCreateStbReq(&pReq);




×








UNCOV

1127



  tDecoderClear(&coder);




×








UNCOV

1128



  return code;




×







1129



}










1130













UNCOV

1131



static int32_t taosDropStb(TAOS* taos, void* meta, uint32_t metaLen) {




×








UNCOV

1132



  if (taos == NULL || meta == NULL) {





×







1133



    uError("invalid parameter in %s", __func__);





×







1134



    return TSDB_CODE_INVALID_PARA;




×







1135



  }











UNCOV

1136



  SVDropStbReq req = {0};




×








UNCOV

1137



  SDecoder     coder = {0};




×








UNCOV

1138



  SMDropStbReq pReq = {0};




×








UNCOV

1139



  int32_t      code = TSDB_CODE_SUCCESS;




×








UNCOV

1140



  SRequestObj* pRequest = NULL;




×







1141













UNCOV

1142



  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest, 0));





×








UNCOV

1143



  uDebug(LOG_ID_TAG " drop stable, meta:%p, metaLen:%d", LOG_ID_VALUE, meta, metaLen);





×








UNCOV

1144



  pRequest->syncQuery = true;




×








UNCOV

1145



  if (!pRequest->pDb) {





×







1146



    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;




×







1147



    goto end;




×







1148



  }










1149



  // decode and process req











UNCOV

1150



  void*    data = POINTER_SHIFT(meta, sizeof(SMsgHead));




×








UNCOV

1151



  uint32_t len = metaLen - sizeof(SMsgHead);




×








UNCOV

1152



  tDecoderInit(&coder, data, len);




×








UNCOV

1153



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





×







1154



    code = TSDB_CODE_INVALID_PARA;




×







1155



    goto end;




×







1156



  }










1157













UNCOV

1158



  SCatalog* pCatalog = NULL;




×








UNCOV

1159



  RAW_RETURN_CHECK(catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog));





×








UNCOV

1160



  SRequestConnInfo conn = {.pTrans = pRequest->pTscObj->pAppInfo->pTransporter,




×








UNCOV

1161



                           .requestId = pRequest->requestId,




×








UNCOV

1162



                           .requestObjRefId = pRequest->self,




×








UNCOV

1163



                           .mgmtEps = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp)};




×








UNCOV

1164



  SName            pName = {0};




×








UNCOV

1165



  toName(pRequest->pTscObj->acctId, pRequest->pDb, req.name, &pName);




×








UNCOV

1166



  STableMeta* pTableMeta = NULL;




×








UNCOV

1167



  code = catalogGetTableMeta(pCatalog, &conn, &pName, &pTableMeta);




×








UNCOV

1168



  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {





×








UNCOV

1169



    code = TSDB_CODE_SUCCESS;




×








UNCOV

1170



    taosMemoryFreeClear(pTableMeta);





×








UNCOV

1171



    goto end;




×







1172



  }











UNCOV

1173



  if (code != TSDB_CODE_SUCCESS) {





×







1174



    goto end;




×







1175



  }











UNCOV

1176



  pReq.suid = pTableMeta->uid;




×








UNCOV

1177



  taosMemoryFreeClear(pTableMeta);





×







1178












1179



  // build drop stable











UNCOV

1180



  pReq.igNotExists = true;




×








UNCOV

1181



  pReq.source = TD_REQ_FROM_TAOX;




×







1182



  //  pReq.suid = processSuid(req.suid, pRequest->pDb);










1183













UNCOV

1184



  uDebug(LOG_ID_TAG " drop stable name:%s suid:%" PRId64 " new suid:%" PRId64, LOG_ID_VALUE, req.name, req.suid,





×







1185



         pReq.suid);











UNCOV

1186



  STscObj* pTscObj = pRequest->pTscObj;




×








UNCOV

1187



  SName    tableName = {0};




×








UNCOV

1188



  toName(pTscObj->acctId, pRequest->pDb, req.name, &tableName);




×








UNCOV

1189



  if (tNameExtractFullName(&tableName, pReq.name) != 0) {





×







1190



    code = TSDB_CODE_INVALID_PARA;




×







1191



    goto end;




×







1192



  }










1193













UNCOV

1194



  SCmdMsgInfo pCmdMsg = {0};




×








UNCOV

1195



  pCmdMsg.epSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);




×








UNCOV

1196



  pCmdMsg.msgType = TDMT_MND_DROP_STB;




×








UNCOV

1197



  pCmdMsg.msgLen = tSerializeSMDropStbReq(NULL, 0, &pReq);




×








UNCOV

1198



  if (pCmdMsg.msgLen <= 0) {





×







1199



    code = TSDB_CODE_INVALID_PARA;




×







1200



    goto end;




×







1201



  }











UNCOV

1202



  pCmdMsg.pMsg = taosMemoryMalloc(pCmdMsg.msgLen);





×








UNCOV

1203



  RAW_NULL_CHECK(pCmdMsg.pMsg);





×








UNCOV

1204



  if (tSerializeSMDropStbReq(pCmdMsg.pMsg, pCmdMsg.msgLen, &pReq) <= 0) {





×







1205



    code = TSDB_CODE_INVALID_PARA;




×







1206



    taosMemoryFree(pCmdMsg.pMsg);





×







1207



    goto end;




×







1208



  }










1209













UNCOV

1210



  SQuery pQuery = {0};




×








UNCOV

1211



  pQuery.execMode = QUERY_EXEC_MODE_RPC;




×








UNCOV

1212



  pQuery.pCmdMsg = &pCmdMsg;




×








UNCOV

1213



  pQuery.msgType = pQuery.pCmdMsg->msgType;




×








UNCOV

1214



  pQuery.stableQuery = true;




×







1215













UNCOV

1216



  launchQueryImpl(pRequest, &pQuery, true, NULL);  // ignore, because return value is pRequest




×








UNCOV

1217



  taosMemoryFree(pCmdMsg.pMsg);





×








UNCOV

1218



  if (pRequest->code == TSDB_CODE_SUCCESS) {





×







1219



    // ignore the error code











UNCOV

1220



    RAW_RETURN_CHECK(catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog));





×








UNCOV

1221



    RAW_RETURN_CHECK(catalogRemoveTableMeta(pCatalog, &tableName));





×







1222



  }










1223













UNCOV

1224



  code = pRequest->code;




×







1225













UNCOV

1226



end:




×








UNCOV

1227



  uDebug(LOG_ID_TAG " drop stable return, msg:%s", LOG_ID_VALUE, tstrerror(code));





×








UNCOV

1228



  destroyRequest(pRequest);




×








UNCOV

1229



  tDecoderClear(&coder);




×








UNCOV

1230



  return code;




×







1231



}










1232












1233



typedef struct SVgroupCreateTableBatch {










1234



  SVCreateTbBatchReq req;










1235



  SVgroupInfo        info;










1236



  char               dbName[TSDB_DB_NAME_LEN];










1237



} SVgroupCreateTableBatch;










1238













UNCOV

1239



static void destroyCreateTbReqBatch(void* data) {




×








UNCOV

1240



  if (data == NULL) {





×







1241



    uError("invalid parameter in %s", __func__);





×







1242



    return;




×







1243



  }











UNCOV

1244



  SVgroupCreateTableBatch* pTbBatch = (SVgroupCreateTableBatch*)data;




×








UNCOV

1245



  taosArrayDestroy(pTbBatch->req.pArray);




×







1246



}










1247













UNCOV

1248



static int32_t taosCreateTable(TAOS* taos, void* meta, uint32_t metaLen) {




×








UNCOV

1249



  if (taos == NULL || meta == NULL) {





×







1250



    uError("invalid parameter in %s", __func__);





×







1251



    return TSDB_CODE_INVALID_PARA;




×







1252



  }











UNCOV

1253



  SVCreateTbBatchReq req = {0};




×








UNCOV

1254



  SDecoder           coder = {0};




×








UNCOV

1255



  int32_t            code = TSDB_CODE_SUCCESS;




×








UNCOV

1256



  SRequestObj*       pRequest = NULL;




×








UNCOV

1257



  SQuery*            pQuery = NULL;




×








UNCOV

1258



  SHashObj*          pVgroupHashmap = NULL;




×








UNCOV

1259



  SArray*            pTagList = taosArrayInit(0, POINTER_BYTES);




×








UNCOV

1260



  RAW_NULL_CHECK(pTagList);





×








UNCOV

1261



  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest, 0));





×








UNCOV

1262



  uDebug(LOG_ID_TAG " create table, meta:%p, metaLen:%d", LOG_ID_VALUE, meta, metaLen);





×







1263













UNCOV

1264



  pRequest->syncQuery = true;




×








UNCOV

1265



  if (!pRequest->pDb) {





×







1266



    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;




×







1267



    goto end;




×







1268



  }










1269



  // decode and process req











UNCOV

1270



  void*    data = POINTER_SHIFT(meta, sizeof(SMsgHead));




×








UNCOV

1271



  uint32_t len = metaLen - sizeof(SMsgHead);




×








UNCOV

1272



  tDecoderInit(&coder, data, len);




×








UNCOV

1273



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





×







1274



    code = TSDB_CODE_INVALID_PARA;




×







1275



    goto end;




×







1276



  }










1277













UNCOV

1278



  STscObj* pTscObj = pRequest->pTscObj;




×







1279













UNCOV

1280



  SVCreateTbReq* pCreateReq = NULL;




×








UNCOV

1281



  SCatalog*      pCatalog = NULL;




×








UNCOV

1282



  RAW_RETURN_CHECK(catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog));





×








UNCOV

1283



  pVgroupHashmap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);




×








UNCOV

1284



  RAW_NULL_CHECK(pVgroupHashmap);





×








UNCOV

1285



  taosHashSetFreeFp(pVgroupHashmap, destroyCreateTbReqBatch);




×







1286













UNCOV

1287



  SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter,




×








UNCOV

1288



                           .requestId = pRequest->requestId,




×








UNCOV

1289



                           .requestObjRefId = pRequest->self,




×








UNCOV

1290



                           .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)};




×







1291













UNCOV

1292



  pRequest->tableList = taosArrayInit(req.nReqs, sizeof(SName));




×








UNCOV

1293



  RAW_NULL_CHECK(pRequest->tableList);





×







1294



  // loop to create table











UNCOV

1295



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





×








UNCOV

1296



    pCreateReq = req.pReqs + iReq;




×







1297













UNCOV

1298



    SVgroupInfo pInfo = {0};




×








UNCOV

1299



    SName       pName = {0};




×








UNCOV

1300



    toName(pTscObj->acctId, pRequest->pDb, pCreateReq->name, &pName);




×








UNCOV

1301



    code = catalogGetTableHashVgroup(pCatalog, &conn, &pName, &pInfo);




×








UNCOV

1302



    if (code != TSDB_CODE_SUCCESS) {





×







1303



      goto end;




×







1304



    }










1305













UNCOV

1306



    pCreateReq->flags |= TD_CREATE_IF_NOT_EXISTS;




×







1307



    // change tag cid to new cid











UNCOV

1308



    if (pCreateReq->type == TSDB_CHILD_TABLE) {





×








UNCOV

1309



      STableMeta* pTableMeta = NULL;




×








UNCOV

1310



      SName       sName = {0};




×








UNCOV

1311



      tb_uid_t    oldSuid = pCreateReq->ctb.suid;




×







1312



      //      pCreateReq->ctb.suid = processSuid(pCreateReq->ctb.suid, pRequest->pDb);











UNCOV

1313



      toName(pTscObj->acctId, pRequest->pDb, pCreateReq->ctb.stbName, &sName);




×








UNCOV

1314



      code = catalogGetTableMeta(pCatalog, &conn, &sName, &pTableMeta);




×








UNCOV

1315



      if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {





×







1316



        code = TSDB_CODE_SUCCESS;




×







1317



        taosMemoryFreeClear(pTableMeta);





×







1318



        continue;




×







1319



      }










1320













UNCOV

1321



      if (code != TSDB_CODE_SUCCESS) {





×







1322



        goto end;




×







1323



      }











UNCOV

1324



      pCreateReq->ctb.suid = pTableMeta->uid;




×







1325













UNCOV

1326



      SArray* pTagVals = NULL;




×








UNCOV

1327



      code = tTagToValArray((STag*)pCreateReq->ctb.pTag, &pTagVals);




×








UNCOV

1328



      if (code != TSDB_CODE_SUCCESS) {





×







1329



        taosMemoryFreeClear(pTableMeta);





×







1330



        goto end;




×







1331



      }










1332













UNCOV

1333



      bool rebuildTag = false;




×








UNCOV

1334



      for (int32_t i = 0; i < taosArrayGetSize(pCreateReq->ctb.tagName); i++) {





×








UNCOV

1335



        char* tName = taosArrayGet(pCreateReq->ctb.tagName, i);




×








UNCOV

1336



        if (tName == NULL) {





×







1337



          continue;




×







1338



        }











UNCOV

1339



        for (int32_t j = pTableMeta->tableInfo.numOfColumns;




×








UNCOV

1340



             j < pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags; j++) {





×








UNCOV

1341



          SSchema* tag = &pTableMeta->schema[j];




×








UNCOV

1342



          if (strcmp(tag->name, tName) == 0 && tag->type != TSDB_DATA_TYPE_JSON) {





×








UNCOV

1343



            STagVal* pTagVal = (STagVal*)taosArrayGet(pTagVals, i);




×








UNCOV

1344



            if (pTagVal) {





×








UNCOV

1345



              if (pTagVal->cid != tag->colId) {





×








UNCOV

1346



                pTagVal->cid = tag->colId;




×








UNCOV

1347



                rebuildTag = true;




×







1348



              }










1349



            } else {










1350



              uError("create tb invalid data %s, size:%d index:%d cid:%d", pCreateReq->name,





×







1351



                     (int)taosArrayGetSize(pTagVals), i, tag->colId);










1352



            }










1353



          }










1354



        }










1355



      }











UNCOV

1356



      taosMemoryFreeClear(pTableMeta);





×








UNCOV

1357



      if (rebuildTag) {





×








UNCOV

1358



        STag* ppTag = NULL;




×








UNCOV

1359



        code = tTagNew(pTagVals, 1, false, &ppTag);




×








UNCOV

1360



        taosArrayDestroy(pTagVals);




×








UNCOV

1361



        pTagVals = NULL;




×








UNCOV

1362



        if (code != TSDB_CODE_SUCCESS) {





×







1363



          goto end;




×







1364



        }











UNCOV

1365



        if (NULL == taosArrayPush(pTagList, &ppTag)) {





×







1366



          tTagFree(ppTag);




×







1367



          goto end;




×







1368



        }











UNCOV

1369



        pCreateReq->ctb.pTag = (uint8_t*)ppTag;




×







1370



      }











UNCOV

1371



      taosArrayDestroy(pTagVals);




×







1372



    }











UNCOV

1373



    RAW_NULL_CHECK(taosArrayPush(pRequest->tableList, &pName));





×







1374













UNCOV

1375



    SVgroupCreateTableBatch* pTableBatch = taosHashGet(pVgroupHashmap, &pInfo.vgId, sizeof(pInfo.vgId));




×








UNCOV

1376



    if (pTableBatch == NULL) {





×








UNCOV

1377



      SVgroupCreateTableBatch tBatch = {0};




×








UNCOV

1378



      tBatch.info = pInfo;




×








UNCOV

1379



      tstrncpy(tBatch.dbName, pRequest->pDb, TSDB_DB_NAME_LEN);




×







1380













UNCOV

1381



      tBatch.req.pArray = taosArrayInit(4, sizeof(struct SVCreateTbReq));




×








UNCOV

1382



      RAW_NULL_CHECK(tBatch.req.pArray);





×








UNCOV

1383



      RAW_NULL_CHECK(taosArrayPush(tBatch.req.pArray, pCreateReq));





×








UNCOV

1384



      tBatch.req.source = TD_REQ_FROM_TAOX;




×








UNCOV

1385



      RAW_RETURN_CHECK(taosHashPut(pVgroupHashmap, &pInfo.vgId, sizeof(pInfo.vgId), &tBatch, sizeof(tBatch)));





×







1386



    } else {  // add to the correct vgroup











UNCOV

1387



      RAW_NULL_CHECK(taosArrayPush(pTableBatch->req.pArray, pCreateReq));





×







1388



    }










1389



  }










1390













UNCOV

1391



  if (taosHashGetSize(pVgroupHashmap) == 0) {





×







1392



    goto end;




×







1393



  }











UNCOV

1394



  SArray* pBufArray = NULL;




×








UNCOV

1395



  RAW_RETURN_CHECK(serializeVgroupsCreateTableBatch(pVgroupHashmap, &pBufArray));





×








UNCOV

1396



  pQuery = NULL;




×








UNCOV

1397



  code = nodesMakeNode(QUERY_NODE_QUERY, (SNode**)&pQuery);




×








UNCOV

1398



  if (TSDB_CODE_SUCCESS != code) goto end;





×








UNCOV

1399



  pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;




×








UNCOV

1400



  pQuery->msgType = TDMT_VND_CREATE_TABLE;




×








UNCOV

1401



  pQuery->stableQuery = false;




×








UNCOV

1402



  code = nodesMakeNode(QUERY_NODE_CREATE_TABLE_STMT, &pQuery->pRoot);




×








UNCOV

1403



  if (TSDB_CODE_SUCCESS != code) goto end;





×








UNCOV

1404



  RAW_NULL_CHECK(pQuery->pRoot);





×







1405













UNCOV

1406



  RAW_RETURN_CHECK(rewriteToVnodeModifyOpStmt(pQuery, pBufArray));





×







1407













UNCOV

1408



  launchQueryImpl(pRequest, pQuery, true, NULL);




×








UNCOV

1409



  if (pRequest->code == TSDB_CODE_SUCCESS) {





×








UNCOV

1410



    RAW_RETURN_CHECK(removeMeta(pTscObj, pRequest->tableList, false));





×







1411



  }










1412













UNCOV

1413



  code = pRequest->code;




×







1414













UNCOV

1415



end:




×








UNCOV

1416



  uDebug(LOG_ID_TAG " create table return, msg:%s", LOG_ID_VALUE, tstrerror(code));





×








UNCOV

1417



  tDeleteSVCreateTbBatchReq(&req);




×







1418













UNCOV

1419



  taosHashCleanup(pVgroupHashmap);




×








UNCOV

1420



  destroyRequest(pRequest);




×








UNCOV

1421



  tDecoderClear(&coder);




×








UNCOV

1422



  qDestroyQuery(pQuery);




×








UNCOV

1423



  taosArrayDestroyP(pTagList, NULL);




×








UNCOV

1424



  return code;




×







1425



}










1426












1427



typedef struct SVgroupDropTableBatch {










1428



  SVDropTbBatchReq req;










1429



  SVgroupInfo      info;










1430



  char             dbName[TSDB_DB_NAME_LEN];










1431



} SVgroupDropTableBatch;










1432













UNCOV

1433



static void destroyDropTbReqBatch(void* data) {




×








UNCOV

1434



  if (data == NULL) {





×







1435



    uError("invalid parameter in %s", __func__);





×







1436



    return;




×







1437



  }











UNCOV

1438



  SVgroupDropTableBatch* pTbBatch = (SVgroupDropTableBatch*)data;




×








UNCOV

1439



  taosArrayDestroy(pTbBatch->req.pArray);




×







1440



}










1441













UNCOV

1442



static int32_t taosDropTable(TAOS* taos, void* meta, uint32_t metaLen) {




×








UNCOV

1443



  if (taos == NULL || meta == NULL) {





×







1444



    uError("invalid parameter in %s", __func__);





×







1445



    return TSDB_CODE_INVALID_PARA;




×







1446



  }











UNCOV

1447



  SVDropTbBatchReq req = {0};




×








UNCOV

1448



  SDecoder         coder = {0};




×








UNCOV

1449



  int32_t          code = TSDB_CODE_SUCCESS;




×








UNCOV

1450



  SRequestObj*     pRequest = NULL;




×








UNCOV

1451



  SQuery*          pQuery = NULL;




×








UNCOV

1452



  SHashObj*        pVgroupHashmap = NULL;




×







1453













UNCOV

1454



  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest, 0));





×








UNCOV

1455



  uDebug(LOG_ID_TAG " drop table, meta:%p, len:%d", LOG_ID_VALUE, meta, metaLen);





×







1456













UNCOV

1457



  pRequest->syncQuery = true;




×








UNCOV

1458



  if (!pRequest->pDb) {





×







1459



    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;




×







1460



    goto end;




×







1461



  }










1462



  // decode and process req











UNCOV

1463



  void*    data = POINTER_SHIFT(meta, sizeof(SMsgHead));




×








UNCOV

1464



  uint32_t len = metaLen - sizeof(SMsgHead);




×








UNCOV

1465



  tDecoderInit(&coder, data, len);




×








UNCOV

1466



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





×







1467



    code = TSDB_CODE_INVALID_PARA;




×







1468



    goto end;




×







1469



  }










1470













UNCOV

1471



  STscObj* pTscObj = pRequest->pTscObj;




×







1472













UNCOV

1473



  SVDropTbReq* pDropReq = NULL;




×








UNCOV

1474



  SCatalog*    pCatalog = NULL;




×








UNCOV

1475



  RAW_RETURN_CHECK(catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog));





×







1476













UNCOV

1477



  pVgroupHashmap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);




×








UNCOV

1478



  RAW_NULL_CHECK(pVgroupHashmap);





×








UNCOV

1479



  taosHashSetFreeFp(pVgroupHashmap, destroyDropTbReqBatch);




×







1480













UNCOV

1481



  SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter,




×








UNCOV

1482



                           .requestId = pRequest->requestId,




×








UNCOV

1483



                           .requestObjRefId = pRequest->self,




×








UNCOV

1484



                           .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)};




×








UNCOV

1485



  pRequest->tableList = taosArrayInit(req.nReqs, sizeof(SName));




×








UNCOV

1486



  RAW_NULL_CHECK(pRequest->tableList);





×







1487



  // loop to create table











UNCOV

1488



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





×








UNCOV

1489



    pDropReq = req.pReqs + iReq;




×








UNCOV

1490



    pDropReq->igNotExists = true;




×







1491



    //    pDropReq->suid = processSuid(pDropReq->suid, pRequest->pDb);










1492













UNCOV

1493



    SVgroupInfo pInfo = {0};




×








UNCOV

1494



    SName       pName = {0};




×








UNCOV

1495



    toName(pTscObj->acctId, pRequest->pDb, pDropReq->name, &pName);




×








UNCOV

1496



    RAW_RETURN_CHECK(catalogGetTableHashVgroup(pCatalog, &conn, &pName, &pInfo));





×







1497













UNCOV

1498



    STableMeta* pTableMeta = NULL;




×








UNCOV

1499



    code = catalogGetTableMeta(pCatalog, &conn, &pName, &pTableMeta);




×








UNCOV

1500



    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST) {





×








UNCOV

1501



      code = TSDB_CODE_SUCCESS;




×








UNCOV

1502



      taosMemoryFreeClear(pTableMeta);





×








UNCOV

1503



      continue;




×







1504



    }











UNCOV

1505



    if (code != TSDB_CODE_SUCCESS) {





×







1506



      goto end;




×







1507



    }











UNCOV

1508



    tb_uid_t oldSuid = pDropReq->suid;




×








UNCOV

1509



    pDropReq->suid = pTableMeta->suid;




×








UNCOV

1510



    taosMemoryFreeClear(pTableMeta);





×








UNCOV

1511



    uDebug(LOG_ID_TAG " drop table name:%s suid:%" PRId64 " new suid:%" PRId64, LOG_ID_VALUE, pDropReq->name, oldSuid,





×







1512



           pDropReq->suid);










1513













UNCOV

1514



    RAW_NULL_CHECK(taosArrayPush(pRequest->tableList, &pName));





×








UNCOV

1515



    SVgroupDropTableBatch* pTableBatch = taosHashGet(pVgroupHashmap, &pInfo.vgId, sizeof(pInfo.vgId));




×








UNCOV

1516



    if (pTableBatch == NULL) {





×








UNCOV

1517



      SVgroupDropTableBatch tBatch = {0};




×








UNCOV

1518



      tBatch.info = pInfo;




×








UNCOV

1519



      tBatch.req.pArray = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SVDropTbReq));




×








UNCOV

1520



      RAW_NULL_CHECK(tBatch.req.pArray);





×








UNCOV

1521



      RAW_NULL_CHECK(taosArrayPush(tBatch.req.pArray, pDropReq));





×








UNCOV

1522



      RAW_RETURN_CHECK(taosHashPut(pVgroupHashmap, &pInfo.vgId, sizeof(pInfo.vgId), &tBatch, sizeof(tBatch)));





×







1523



    } else {  // add to the correct vgroup











UNCOV

1524



      RAW_NULL_CHECK(taosArrayPush(pTableBatch->req.pArray, pDropReq));





×







1525



    }










1526



  }










1527













UNCOV

1528



  if (taosHashGetSize(pVgroupHashmap) == 0) {





×








UNCOV

1529



    goto end;




×







1530



  }











UNCOV

1531



  SArray* pBufArray = NULL;




×








UNCOV

1532



  RAW_RETURN_CHECK(serializeVgroupsDropTableBatch(pVgroupHashmap, &pBufArray));





×








UNCOV

1533



  code = nodesMakeNode(QUERY_NODE_QUERY, (SNode**)&pQuery);




×








UNCOV

1534



  if (TSDB_CODE_SUCCESS != code) goto end;





×








UNCOV

1535



  pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;




×








UNCOV

1536



  pQuery->msgType = TDMT_VND_DROP_TABLE;




×








UNCOV

1537



  pQuery->stableQuery = false;




×








UNCOV

1538



  pQuery->pRoot = NULL;




×








UNCOV

1539



  code = nodesMakeNode(QUERY_NODE_DROP_TABLE_STMT, &pQuery->pRoot);




×








UNCOV

1540



  if (TSDB_CODE_SUCCESS != code) goto end;





×








UNCOV

1541



  RAW_RETURN_CHECK(rewriteToVnodeModifyOpStmt(pQuery, pBufArray));





×







1542













UNCOV

1543



  launchQueryImpl(pRequest, pQuery, true, NULL);




×








UNCOV

1544



  if (pRequest->code == TSDB_CODE_SUCCESS) {





×








UNCOV

1545



    RAW_RETURN_CHECK(removeMeta(pTscObj, pRequest->tableList, false));





×







1546



  }











UNCOV

1547



  code = pRequest->code;




×







1548













UNCOV

1549



end:




×








UNCOV

1550



  uDebug(LOG_ID_TAG " drop table return, msg:%s", LOG_ID_VALUE, tstrerror(code));





×








UNCOV

1551



  taosHashCleanup(pVgroupHashmap);




×








UNCOV

1552



  destroyRequest(pRequest);




×








UNCOV

1553



  tDecoderClear(&coder);




×








UNCOV

1554



  qDestroyQuery(pQuery);




×








UNCOV

1555



  return code;




×







1556



}










1557













UNCOV

1558



static int32_t taosDeleteData(TAOS* taos, void* meta, uint32_t metaLen) {




×








UNCOV

1559



  if (taos == NULL || meta == NULL) {





×







1560



    uError("invalid parameter in %s", __func__);





×







1561



    return TSDB_CODE_INVALID_PARA;




×







1562



  }











UNCOV

1563



  SDeleteRes req = {0};




×








UNCOV

1564



  SDecoder   coder = {0};




×








UNCOV

1565



  char       sql[256] = {0};




×








UNCOV

1566



  int32_t    code = TSDB_CODE_SUCCESS;




×







1567













UNCOV

1568



  uDebug("connId:0x%" PRIx64 " delete data, meta:%p, len:%d", *(int64_t*)taos, meta, metaLen);





×







1569












1570



  // decode and process req











UNCOV

1571



  void*    data = POINTER_SHIFT(meta, sizeof(SMsgHead));




×








UNCOV

1572



  uint32_t len = metaLen - sizeof(SMsgHead);




×








UNCOV

1573



  tDecoderInit(&coder, data, len);




×








UNCOV

1574



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





×







1575



    code = TSDB_CODE_INVALID_PARA;




×







1576



    goto end;




×







1577



  }










1578













UNCOV

1579



  (void)snprintf(sql, sizeof(sql), "delete from `%s` where `%s` >= %" PRId64 " and `%s` <= %" PRId64, req.tableFName,




×







1580



                 req.tsColName, req.skey, req.tsColName, req.ekey);










1581













UNCOV

1582



  TAOS_RES* res = taosQueryImpl(taos, sql, false, TD_REQ_FROM_TAOX);




×








UNCOV

1583



  RAW_NULL_CHECK(res);





×








UNCOV

1584



  SRequestObj* pRequest = (SRequestObj*)res;




×








UNCOV

1585



  code = pRequest->code;




×








UNCOV

1586



  if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_PAR_GET_META_ERROR) {





×








UNCOV

1587



    code = TSDB_CODE_SUCCESS;




×







1588



  }











UNCOV

1589



  taos_free_result(res);




×







1590













UNCOV

1591



end:




×








UNCOV

1592



  uDebug("connId:0x%" PRIx64 " delete data sql:%s, code:%s", *(int64_t*)taos, sql, tstrerror(code));





×








UNCOV

1593



  tDecoderClear(&coder);




×








UNCOV

1594



  return code;




×







1595



}










1596













UNCOV

1597



static int32_t taosAlterTable(TAOS* taos, void* meta, uint32_t metaLen) {




×








UNCOV

1598



  if (taos == NULL || meta == NULL) {





×







1599



    uError("invalid parameter in %s", __func__);





×







1600



    return TSDB_CODE_INVALID_PARA;




×







1601



  }











UNCOV

1602



  SVAlterTbReq   req = {0};




×








UNCOV

1603



  SDecoder       dcoder = {0};




×








UNCOV

1604



  int32_t        code = TSDB_CODE_SUCCESS;




×








UNCOV

1605



  SRequestObj*   pRequest = NULL;




×








UNCOV

1606



  SQuery*        pQuery = NULL;




×








UNCOV

1607



  SArray*        pArray = NULL;




×








UNCOV

1608



  SVgDataBlocks* pVgData = NULL;




×







1609













UNCOV

1610



  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest, 0));





×








UNCOV

1611



  uDebug(LOG_ID_TAG " alter table, meta:%p, len:%d", LOG_ID_VALUE, meta, metaLen);





×








UNCOV

1612



  pRequest->syncQuery = true;




×








UNCOV

1613



  if (!pRequest->pDb) {





×







1614



    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;




×







1615



    goto end;




×







1616



  }










1617



  // decode and process req











UNCOV

1618



  void*    data = POINTER_SHIFT(meta, sizeof(SMsgHead));




×








UNCOV

1619



  uint32_t len = metaLen - sizeof(SMsgHead);




×








UNCOV

1620



  tDecoderInit(&dcoder, data, len);




×








UNCOV

1621



  if (tDecodeSVAlterTbReq(&dcoder, &req) < 0) {





×







1622



    code = TSDB_CODE_INVALID_PARA;




×







1623



    goto end;




×







1624



  }










1625












1626



  // do not deal TSDB_ALTER_TABLE_UPDATE_OPTIONS











UNCOV

1627



  if (req.action == TSDB_ALTER_TABLE_UPDATE_OPTIONS) {





×








UNCOV

1628



    goto end;




×







1629



  }










1630













UNCOV

1631



  STscObj*  pTscObj = pRequest->pTscObj;




×








UNCOV

1632



  SCatalog* pCatalog = NULL;




×








UNCOV

1633



  RAW_RETURN_CHECK(catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCatalog));





×








UNCOV

1634



  SRequestConnInfo conn = {.pTrans = pTscObj->pAppInfo->pTransporter,




×








UNCOV

1635



                           .requestId = pRequest->requestId,




×








UNCOV

1636



                           .requestObjRefId = pRequest->self,




×








UNCOV

1637



                           .mgmtEps = getEpSet_s(&pTscObj->pAppInfo->mgmtEp)};




×







1638













UNCOV

1639



  SVgroupInfo pInfo = {0};




×








UNCOV

1640



  SName       pName = {0};




×








UNCOV

1641



  toName(pTscObj->acctId, pRequest->pDb, req.tbName, &pName);




×








UNCOV

1642



  RAW_RETURN_CHECK(catalogGetTableHashVgroup(pCatalog, &conn, &pName, &pInfo));





×








UNCOV

1643



  pArray = taosArrayInit(1, sizeof(void*));




×








UNCOV

1644



  RAW_NULL_CHECK(pArray);





×







1645













UNCOV

1646



  pVgData = taosMemoryCalloc(1, sizeof(SVgDataBlocks));





×








UNCOV

1647



  RAW_NULL_CHECK(pVgData);





×








UNCOV

1648



  pVgData->vg = pInfo;




×







1649













UNCOV

1650



  int tlen = 0;




×








UNCOV

1651



  req.source = TD_REQ_FROM_TAOX;




×








UNCOV

1652



  tEncodeSize(tEncodeSVAlterTbReq, &req, tlen, code);





×








UNCOV

1653



  if (code != 0) {





×







1654



    code = terrno;




×







1655



    goto end;




×







1656



  }











UNCOV

1657



  tlen += sizeof(SMsgHead);




×








UNCOV

1658



  void* pMsg = taosMemoryMalloc(tlen);





×








UNCOV

1659



  RAW_NULL_CHECK(pMsg);





×








UNCOV

1660



  ((SMsgHead*)pMsg)->vgId = htonl(pInfo.vgId);




×








UNCOV

1661



  ((SMsgHead*)pMsg)->contLen = htonl(tlen);




×








UNCOV

1662



  void*    pBuf = POINTER_SHIFT(pMsg, sizeof(SMsgHead));




×








UNCOV

1663



  SEncoder coder = {0};




×








UNCOV

1664



  tEncoderInit(&coder, pBuf, tlen - sizeof(SMsgHead));




×








UNCOV

1665



  code = tEncodeSVAlterTbReq(&coder, &req);




×








UNCOV

1666



  if (code != 0) {





×







1667



    tEncoderClear(&coder);




×







1668



    code = terrno;




×







1669



    goto end;




×







1670



  }











UNCOV

1671



  tEncoderClear(&coder);




×







1672













UNCOV

1673



  pVgData->pData = pMsg;




×








UNCOV

1674



  pVgData->size = tlen;




×







1675













UNCOV

1676



  pVgData->numOfTables = 1;




×








UNCOV

1677



  RAW_NULL_CHECK(taosArrayPush(pArray, &pVgData));





×







1678













UNCOV

1679



  pQuery = NULL;




×








UNCOV

1680



  code = nodesMakeNode(QUERY_NODE_QUERY, (SNode**)&pQuery);




×








UNCOV

1681



  if (NULL == pQuery) goto end;





×








UNCOV

1682



  pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;




×








UNCOV

1683



  pQuery->msgType = TDMT_VND_ALTER_TABLE;




×








UNCOV

1684



  pQuery->stableQuery = false;




×








UNCOV

1685



  pQuery->pRoot = NULL;




×








UNCOV

1686



  code = nodesMakeNode(QUERY_NODE_ALTER_TABLE_STMT, &pQuery->pRoot);




×








UNCOV

1687



  if (TSDB_CODE_SUCCESS != code) goto end;





×








UNCOV

1688



  RAW_RETURN_CHECK(rewriteToVnodeModifyOpStmt(pQuery, pArray));





×







1689













UNCOV

1690



  launchQueryImpl(pRequest, pQuery, true, NULL);




×







1691













UNCOV

1692



  pVgData = NULL;




×








UNCOV

1693



  pArray = NULL;




×








UNCOV

1694



  code = pRequest->code;




×








UNCOV

1695



  if (code == TSDB_CODE_TDB_TABLE_NOT_EXIST) {





×








UNCOV

1696



    code = TSDB_CODE_SUCCESS;




×







1697



  }










1698













UNCOV

1699



  if (pRequest->code == TSDB_CODE_SUCCESS) {





×








UNCOV

1700



    SExecResult* pRes = &pRequest->body.resInfo.execRes;




×








UNCOV

1701



    if (pRes->res != NULL) {





×








UNCOV

1702



      code = handleAlterTbExecRes(pRes->res, pCatalog);




×







1703



    }










1704



  }











UNCOV

1705



end:




×








UNCOV

1706



  uDebug(LOG_ID_TAG " alter table return, meta:%p, len:%d, msg:%s", LOG_ID_VALUE, meta, metaLen, tstrerror(code));





×








UNCOV

1707



  taosArrayDestroy(pArray);




×








UNCOV

1708



  if (pVgData) taosMemoryFreeClear(pVgData->pData);





×








UNCOV

1709



  taosMemoryFreeClear(pVgData);





×








UNCOV

1710



  destroyRequest(pRequest);




×








UNCOV

1711



  tDecoderClear(&dcoder);




×








UNCOV

1712



  qDestroyQuery(pQuery);




×








UNCOV

1713



  return code;




×







1714



}










1715













UNCOV

1716



int taos_write_raw_block_with_fields(TAOS* taos, int rows, char* pData, const char* tbname, TAOS_FIELD* fields,




×







1717



                                     int numFields) {











UNCOV

1718



  return taos_write_raw_block_with_fields_with_reqid(taos, rows, pData, tbname, fields, numFields, 0);




×







1719



}










1720













UNCOV

1721



int taos_write_raw_block_with_fields_with_reqid(TAOS* taos, int rows, char* pData, const char* tbname,




×







1722



                                                TAOS_FIELD* fields, int numFields, int64_t reqid) {











UNCOV

1723



  if (taos == NULL || pData == NULL || tbname == NULL) {





×







1724



    uError("invalid parameter in %s", __func__);





×







1725



    return TSDB_CODE_INVALID_PARA;




×







1726



  }











UNCOV

1727



  int32_t     code = TSDB_CODE_SUCCESS;




×








UNCOV

1728



  STableMeta* pTableMeta = NULL;




×








UNCOV

1729



  SQuery*     pQuery = NULL;




×








UNCOV

1730



  SHashObj*   pVgHash = NULL;




×







1731













UNCOV

1732



  SRequestObj* pRequest = NULL;




×








UNCOV

1733



  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest, reqid));





×







1734













UNCOV

1735



  uDebug(LOG_ID_TAG " write raw block with field, rows:%d, pData:%p, tbname:%s, fields:%p, numFields:%d", LOG_ID_VALUE,





×







1736



         rows, pData, tbname, fields, numFields);










1737













UNCOV

1738



  pRequest->syncQuery = true;




×








UNCOV

1739



  if (!pRequest->pDb) {





×







1740



    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;




×







1741



    goto end;




×







1742



  }










1743













UNCOV

1744



  SName pName = {TSDB_TABLE_NAME_T, pRequest->pTscObj->acctId, {0}, {0}};




×








UNCOV

1745



  tstrncpy(pName.dbname, pRequest->pDb, sizeof(pName.dbname));




×








UNCOV

1746



  tstrncpy(pName.tname, tbname, sizeof(pName.tname));




×







1747













UNCOV

1748



  struct SCatalog* pCatalog = NULL;




×








UNCOV

1749



  RAW_RETURN_CHECK(catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog));





×







1750













UNCOV

1751



  SRequestConnInfo conn = {0};




×








UNCOV

1752



  conn.pTrans = pRequest->pTscObj->pAppInfo->pTransporter;




×








UNCOV

1753



  conn.requestId = pRequest->requestId;




×








UNCOV

1754



  conn.requestObjRefId = pRequest->self;




×








UNCOV

1755



  conn.mgmtEps = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp);




×







1756













UNCOV

1757



  SVgroupInfo vgData = {0};




×








UNCOV

1758



  RAW_RETURN_CHECK(catalogGetTableHashVgroup(pCatalog, &conn, &pName, &vgData));





×








UNCOV

1759



  RAW_RETURN_CHECK(catalogGetTableMeta(pCatalog, &conn, &pName, &pTableMeta));





×








UNCOV

1760



  RAW_RETURN_CHECK(smlInitHandle(&pQuery));





×








UNCOV

1761



  pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);




×








UNCOV

1762



  RAW_NULL_CHECK(pVgHash);





×








UNCOV

1763



  RAW_RETURN_CHECK(





×







1764



      taosHashPut(pVgHash, (const char*)&vgData.vgId, sizeof(vgData.vgId), (char*)&vgData, sizeof(vgData)));











UNCOV

1765



  RAW_RETURN_CHECK(rawBlockBindData(pQuery, pTableMeta, pData, NULL, fields, numFields, false, NULL, 0, false));





×








UNCOV

1766



  RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));





×







1767













UNCOV

1768



  launchQueryImpl(pRequest, pQuery, true, NULL);




×








UNCOV

1769



  code = pRequest->code;




×







1770













UNCOV

1771



end:




×








UNCOV

1772



  uDebug(LOG_ID_TAG " write raw block with field return, msg:%s", LOG_ID_VALUE, tstrerror(code));





×








UNCOV

1773



  taosMemoryFreeClear(pTableMeta);





×








UNCOV

1774



  qDestroyQuery(pQuery);




×








UNCOV

1775



  destroyRequest(pRequest);




×








UNCOV

1776



  taosHashCleanup(pVgHash);




×








UNCOV

1777



  return code;




×







1778



}










1779













UNCOV

1780



int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname) {




×








UNCOV

1781



  return taos_write_raw_block_with_reqid(taos, rows, pData, tbname, 0);




×







1782



}










1783













UNCOV

1784



int taos_write_raw_block_with_reqid(TAOS* taos, int rows, char* pData, const char* tbname, int64_t reqid) {




×








UNCOV

1785



  if (taos == NULL || pData == NULL || tbname == NULL) {





×







1786



    return TSDB_CODE_INVALID_PARA;




×







1787



  }











UNCOV

1788



  int32_t     code = TSDB_CODE_SUCCESS;




×








UNCOV

1789



  STableMeta* pTableMeta = NULL;




×








UNCOV

1790



  SQuery*     pQuery = NULL;




×








UNCOV

1791



  SHashObj*   pVgHash = NULL;




×







1792













UNCOV

1793



  SRequestObj* pRequest = NULL;




×








UNCOV

1794



  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, &pRequest, reqid));





×







1795













UNCOV

1796



  uDebug(LOG_ID_TAG " write raw block, rows:%d, pData:%p, tbname:%s", LOG_ID_VALUE, rows, pData, tbname);





×







1797













UNCOV

1798



  pRequest->syncQuery = true;




×








UNCOV

1799



  if (!pRequest->pDb) {





×







1800



    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;




×







1801



    goto end;




×







1802



  }










1803













UNCOV

1804



  SName pName = {TSDB_TABLE_NAME_T, pRequest->pTscObj->acctId, {0}, {0}};




×








UNCOV

1805



  tstrncpy(pName.dbname, pRequest->pDb, sizeof(pName.dbname));




×








UNCOV

1806



  tstrncpy(pName.tname, tbname, sizeof(pName.tname));




×







1807













UNCOV

1808



  struct SCatalog* pCatalog = NULL;




×








UNCOV

1809



  RAW_RETURN_CHECK(catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog));





×







1810













UNCOV

1811



  SRequestConnInfo conn = {0};




×








UNCOV

1812



  conn.pTrans = pRequest->pTscObj->pAppInfo->pTransporter;




×








UNCOV

1813



  conn.requestId = pRequest->requestId;




×








UNCOV

1814



  conn.requestObjRefId = pRequest->self;




×








UNCOV

1815



  conn.mgmtEps = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp);




×







1816













UNCOV

1817



  SVgroupInfo vgData = {0};




×








UNCOV

1818



  RAW_RETURN_CHECK(catalogGetTableHashVgroup(pCatalog, &conn, &pName, &vgData));





×








UNCOV

1819



  RAW_RETURN_CHECK(catalogGetTableMeta(pCatalog, &conn, &pName, &pTableMeta));





×








UNCOV

1820



  RAW_RETURN_CHECK(smlInitHandle(&pQuery));





×








UNCOV

1821



  pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);




×








UNCOV

1822



  RAW_NULL_CHECK(pVgHash);





×








UNCOV

1823



  RAW_RETURN_CHECK(





×







1824



      taosHashPut(pVgHash, (const char*)&vgData.vgId, sizeof(vgData.vgId), (char*)&vgData, sizeof(vgData)));











UNCOV

1825



  RAW_RETURN_CHECK(rawBlockBindData(pQuery, pTableMeta, pData, NULL, NULL, 0, false, NULL, 0, false));





×








UNCOV

1826



  RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));





×







1827













UNCOV

1828



  launchQueryImpl(pRequest, pQuery, true, NULL);




×








UNCOV

1829



  code = pRequest->code;




×







1830













UNCOV

1831



end:




×








UNCOV

1832



  uDebug(LOG_ID_TAG " write raw block return, msg:%s", LOG_ID_VALUE, tstrerror(code));





×








UNCOV

1833



  taosMemoryFreeClear(pTableMeta);





×








UNCOV

1834



  qDestroyQuery(pQuery);




×








UNCOV

1835



  destroyRequest(pRequest);




×








UNCOV

1836



  taosHashCleanup(pVgHash);




×








UNCOV

1837



  return code;




×







1838



}










1839













UNCOV

1840



static void* getRawDataFromRes(void* pRetrieve) {




×








UNCOV

1841



  if (pRetrieve == NULL) {





×







1842



    uError("invalid parameter in %s", __func__);





×







1843



    return NULL;




×







1844



  }











UNCOV

1845



  void* rawData = NULL;




×







1846



  // deal with compatibility











UNCOV

1847



  if (*(int64_t*)pRetrieve == 0) {





×







1848



    rawData = ((SRetrieveTableRsp*)pRetrieve)->data;




×








UNCOV

1849



  } else if (*(int64_t*)pRetrieve == 1) {





×








UNCOV

1850



    rawData = ((SRetrieveTableRspForTmq*)pRetrieve)->data;




×







1851



  }











UNCOV

1852



  return rawData;




×







1853



}










1854













UNCOV

1855



static int32_t buildCreateTbMap(SMqDataRsp* rsp, SHashObj* pHashObj) {




×








UNCOV

1856



  if (rsp == NULL || pHashObj == NULL) {





×







1857



    uError("invalid parameter in %s", __func__);





×







1858



    return TSDB_CODE_INVALID_PARA;




×







1859



  }










1860



  // find schema data info











UNCOV

1861



  int32_t       code = 0;




×








UNCOV

1862



  SVCreateTbReq pCreateReq = {0};




×








UNCOV

1863



  SDecoder      decoderTmp = {0};




×







1864













UNCOV

1865



  for (int j = 0; j < rsp->createTableNum; j++) {





×








UNCOV

1866



    void** dataTmp = taosArrayGet(rsp->createTableReq, j);




×








UNCOV

1867



    RAW_NULL_CHECK(dataTmp);





×








UNCOV

1868



    int32_t* lenTmp = taosArrayGet(rsp->createTableLen, j);




×








UNCOV

1869



    RAW_NULL_CHECK(lenTmp);





×







1870













UNCOV

1871



    tDecoderInit(&decoderTmp, *dataTmp, *lenTmp);




×








UNCOV

1872



    RAW_RETURN_CHECK(tDecodeSVCreateTbReq(&decoderTmp, &pCreateReq));





×







1873













UNCOV

1874



    if (pCreateReq.type != TSDB_CHILD_TABLE) {





×







1875



      code = TSDB_CODE_INVALID_MSG;




×







1876



      goto end;




×







1877



    }











UNCOV

1878



    if (taosHashGet(pHashObj, pCreateReq.name, strlen(pCreateReq.name)) == NULL) {





×








UNCOV

1879



      RAW_RETURN_CHECK(





×







1880



          taosHashPut(pHashObj, pCreateReq.name, strlen(pCreateReq.name), &pCreateReq, sizeof(SVCreateTbReq)));










1881



    } else {











UNCOV

1882



      tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE);




×








UNCOV

1883



      pCreateReq = (SVCreateTbReq){0};




×







1884



    }










1885













UNCOV

1886



    tDecoderClear(&decoderTmp);




×







1887



  }











UNCOV

1888



  return 0;




×







1889












1890



end:




×







1891



  tDecoderClear(&decoderTmp);




×







1892



  tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE);




×







1893



  return code;




×







1894



}










1895












1896



typedef enum {










1897



  WRITE_RAW_INIT_START = 0,










1898



  WRITE_RAW_INIT_OK,










1899



  WRITE_RAW_INIT_FAIL,










1900



} WRITE_RAW_INIT_STATUS;










1901












1902



static SHashObj* writeRawCache = NULL;










1903



static int8_t    initFlag = 0;










1904



static int8_t    initedFlag = WRITE_RAW_INIT_START;










1905












1906



typedef struct {










1907



  SHashObj* pVgHash;










1908



  SHashObj* pNameHash;










1909



  SHashObj* pMetaHash;










1910



} rawCacheInfo;










1911












1912



typedef struct {










1913



  SVgroupInfo vgInfo;










1914



  int64_t     uid;










1915



  int64_t     suid;










1916



} tbInfo;










1917













UNCOV

1918



static void tmqFreeMeta(void* data) {




×








UNCOV

1919



  if (data == NULL) {





×







1920



    uError("invalid parameter in %s", __func__);





×







1921



    return;




×







1922



  }











UNCOV

1923



  STableMeta* pTableMeta = *(STableMeta**)data;




×








UNCOV

1924



  taosMemoryFree(pTableMeta);





×







1925



}










1926












1927



static void freeRawCache(void* data) {




×







1928



  if (data == NULL) {





×







1929



    uError("invalid parameter in %s", __func__);





×







1930



    return;




×







1931



  }










1932



  rawCacheInfo* pRawCache = (rawCacheInfo*)data;




×







1933



  taosHashCleanup(pRawCache->pMetaHash);




×







1934



  taosHashCleanup(pRawCache->pNameHash);




×







1935



  taosHashCleanup(pRawCache->pVgHash);




×







1936



}










1937













UNCOV

1938



static int32_t initRawCacheHash() {




×








UNCOV

1939



  if (writeRawCache == NULL) {





×








UNCOV

1940



    writeRawCache = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);




×








UNCOV

1941



    if (writeRawCache == NULL) {





×







1942



      return terrno;




×







1943



    }











UNCOV

1944



    taosHashSetFreeFp(writeRawCache, freeRawCache);




×







1945



  }











UNCOV

1946



  return 0;




×







1947



}










1948













UNCOV

1949



static bool needRefreshMeta(void* rawData, STableMeta* pTableMeta, SSchemaWrapper* pSW) {




×








UNCOV

1950



  if (rawData == NULL || pTableMeta == NULL || pSW == NULL) {





×







1951



    uError("invalid parameter in %s", __func__);





×







1952



    return false;




×







1953



  }











UNCOV

1954



  char* p = (char*)rawData;




×







1955



  // | version | total length | total rows | blankFill | total columns | flag seg| block group id | column schema | each










1956



  // column length |











UNCOV

1957



  p += sizeof(int32_t);




×








UNCOV

1958



  p += sizeof(int32_t);




×








UNCOV

1959



  p += sizeof(int32_t);




×








UNCOV

1960



  p += sizeof(int32_t);




×








UNCOV

1961



  p += sizeof(int32_t);




×








UNCOV

1962



  p += sizeof(uint64_t);




×








UNCOV

1963



  int8_t* fields = p;




×







1964













UNCOV

1965



  if (pSW->nCols != pTableMeta->tableInfo.numOfColumns) {





×








UNCOV

1966



    return true;




×







1967



  }











UNCOV

1968



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





×








UNCOV

1969



    int j = 0;




×








UNCOV

1970



    for (; j < pTableMeta->tableInfo.numOfColumns; j++) {





×








UNCOV

1971



      SSchema* pColSchema = &pTableMeta->schema[j];




×








UNCOV

1972



      char*    fieldName = pSW->pSchema[i].name;




×







1973













UNCOV

1974



      if (strcmp(pColSchema->name, fieldName) == 0) {





×








UNCOV

1975



        if (*fields != pColSchema->type || *(int32_t*)(fields + sizeof(int8_t)) != pColSchema->bytes) {





×







1976



          return true;




×







1977



        }











UNCOV

1978



        break;




×







1979



      }










1980



    }











UNCOV

1981



    fields += sizeof(int8_t) + sizeof(int32_t);




×







1982













UNCOV

1983



    if (j == pTableMeta->tableInfo.numOfColumns) return true;





×







1984



  }











UNCOV

1985



  return false;




×







1986



}










1987













UNCOV

1988



static int32_t getRawCache(SHashObj** pVgHash, SHashObj** pNameHash, SHashObj** pMetaHash, void* key) {




×








UNCOV

1989



  if (pVgHash == NULL || pNameHash == NULL || pMetaHash == NULL || key == NULL) {





×







1990



    uError("invalid parameter in %s", __func__);





×







1991



    return TSDB_CODE_INVALID_PARA;




×







1992



  }











UNCOV

1993



  int32_t code = 0;




×








UNCOV

1994



  void*   cacheInfo = taosHashGet(writeRawCache, &key, POINTER_BYTES);




×








UNCOV

1995



  if (cacheInfo == NULL) {





×








UNCOV

1996



    *pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);




×








UNCOV

1997



    RAW_NULL_CHECK(*pVgHash);





×








UNCOV

1998



    *pNameHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);




×








UNCOV

1999



    RAW_NULL_CHECK(*pNameHash);





×








UNCOV

2000



    *pMetaHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);




×








UNCOV

2001



    RAW_NULL_CHECK(*pMetaHash);





×








UNCOV

2002



    taosHashSetFreeFp(*pMetaHash, tmqFreeMeta);




×








UNCOV

2003



    rawCacheInfo info = {*pVgHash, *pNameHash, *pMetaHash};




×








UNCOV

2004



    RAW_RETURN_CHECK(taosHashPut(writeRawCache, &key, POINTER_BYTES, &info, sizeof(rawCacheInfo)));





×







2005



  } else {










2006



    rawCacheInfo* info = (rawCacheInfo*)cacheInfo;




×







2007



    *pVgHash = info->pVgHash;




×







2008



    *pNameHash = info->pNameHash;




×







2009



    *pMetaHash = info->pMetaHash;




×







2010



  }










2011













UNCOV

2012



  return 0;




×







2013



end:




×







2014



  taosHashCleanup(*pMetaHash);




×







2015



  taosHashCleanup(*pNameHash);




×







2016



  taosHashCleanup(*pVgHash);




×







2017



  return code;




×







2018



}










2019













UNCOV

2020



static int32_t buildRawRequest(TAOS* taos, SRequestObj** pRequest, SCatalog** pCatalog, SRequestConnInfo* conn) {




×








UNCOV

2021



  if (taos == NULL || pRequest == NULL || pCatalog == NULL || conn == NULL) {





×







2022



    uError("invalid parameter in %s", __func__);





×







2023



    return TSDB_CODE_INVALID_PARA;




×







2024



  }











UNCOV

2025



  int32_t code = 0;




×








UNCOV

2026



  RAW_RETURN_CHECK(buildRequest(*(int64_t*)taos, "", 0, NULL, false, pRequest, 0));





×








UNCOV

2027



  (*pRequest)->syncQuery = true;




×








UNCOV

2028



  if (!(*pRequest)->pDb) {





×







2029



    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;




×







2030



    goto end;




×







2031



  }










2032













UNCOV

2033



  RAW_RETURN_CHECK(catalogGetHandle((*pRequest)->pTscObj->pAppInfo->clusterId, pCatalog));





×








UNCOV

2034



  conn->pTrans = (*pRequest)->pTscObj->pAppInfo->pTransporter;




×








UNCOV

2035



  conn->requestId = (*pRequest)->requestId;




×








UNCOV

2036



  conn->requestObjRefId = (*pRequest)->self;




×








UNCOV

2037



  conn->mgmtEps = getEpSet_s(&(*pRequest)->pTscObj->pAppInfo->mgmtEp);




×







2038













UNCOV

2039



end:




×








UNCOV

2040



  return code;




×







2041



}










2042












2043



typedef int32_t _raw_decode_func_(SDecoder* pDecoder, SMqDataRsp* pRsp);











UNCOV

2044



static int32_t  decodeRawData(SDecoder* decoder, void* data, uint32_t dataLen, _raw_decode_func_ func,




×







2045



                              SMqRspObj* rspObj) {











UNCOV

2046



  if (decoder == NULL || data == NULL || func == NULL || rspObj == NULL) {





×







2047



    uError("invalid parameter in %s", __func__);





×







2048



    return TSDB_CODE_INVALID_PARA;




×







2049



  }











UNCOV

2050



  int8_t dataVersion = *(int8_t*)data;




×








UNCOV

2051



  if (dataVersion >= MQ_DATA_RSP_VERSION) {





×








UNCOV

2052



    data = POINTER_SHIFT(data, sizeof(int8_t) + sizeof(int32_t));




×








UNCOV

2053



    if (dataLen < sizeof(int8_t) + sizeof(int32_t)) {





×







2054



      return TSDB_CODE_INVALID_PARA;




×







2055



    }











UNCOV

2056



    dataLen -= sizeof(int8_t) + sizeof(int32_t);




×







2057



  }










2058













UNCOV

2059



  rspObj->resIter = -1;




×








UNCOV

2060



  tDecoderInit(decoder, data, dataLen);




×








UNCOV

2061



  int32_t code = func(decoder, &rspObj->dataRsp);




×








UNCOV

2062



  if (code != 0) {





×







2063



    SET_ERROR_MSG("decode mq taosx data rsp failed");




×







2064



  }











UNCOV

2065



  return code;




×







2066



}










2067













UNCOV

2068



static int32_t processCacheMeta(SHashObj* pVgHash, SHashObj* pNameHash, SHashObj* pMetaHash,




×







2069



                                SVCreateTbReq* pCreateReqDst, SCatalog* pCatalog, SRequestConnInfo* conn, SName* pName,










2070



                                STableMeta** pMeta, SSchemaWrapper* pSW, void* rawData, int32_t retry) {











UNCOV

2071



  if (pVgHash == NULL || pNameHash == NULL || pMetaHash == NULL || pCatalog == NULL || conn == NULL || pName == NULL ||





×








UNCOV

2072



      pMeta == NULL || pSW == NULL || rawData == NULL) {





×







2073



    uError("invalid parameter in %s", __func__);





×







2074



    return TSDB_CODE_INVALID_PARA;




×







2075



  }











UNCOV

2076



  int32_t     code = 0;




×








UNCOV

2077



  STableMeta* pTableMeta = NULL;




×








UNCOV

2078



  tbInfo*     tmpInfo = (tbInfo*)taosHashGet(pNameHash, pName->tname, strlen(pName->tname));




×








UNCOV

2079



  if (tmpInfo == NULL || retry > 0) {





×








UNCOV

2080



    tbInfo info = {0};




×







2081













UNCOV

2082



    RAW_RETURN_CHECK(catalogGetTableHashVgroup(pCatalog, conn, pName, &info.vgInfo));





×








UNCOV

2083



    if (pCreateReqDst && tmpInfo == NULL) {  // change stable name to get meta





×








UNCOV

2084



      tstrncpy(pName->tname, pCreateReqDst->ctb.stbName, TSDB_TABLE_NAME_LEN);




×







2085



    }











UNCOV

2086



    RAW_RETURN_CHECK(catalogGetTableMeta(pCatalog, conn, pName, &pTableMeta));





×








UNCOV

2087



    info.uid = pTableMeta->uid;




×








UNCOV

2088



    if (pTableMeta->tableType == TSDB_CHILD_TABLE) {





×








UNCOV

2089



      info.suid = pTableMeta->suid;




×







2090



    } else {











UNCOV

2091



      info.suid = pTableMeta->uid;




×







2092



    }











UNCOV

2093



    code = taosHashPut(pMetaHash, &info.suid, LONG_BYTES, &pTableMeta, POINTER_BYTES);




×








UNCOV

2094



    if (code != 0) {





×







2095



      taosMemoryFree(pTableMeta);





×







2096



      goto end;




×







2097



    }











UNCOV

2098



    if (pCreateReqDst) {





×








UNCOV

2099



      pTableMeta->vgId = info.vgInfo.vgId;




×








UNCOV

2100



      pTableMeta->uid = pCreateReqDst->uid;




×








UNCOV

2101



      pCreateReqDst->ctb.suid = pTableMeta->suid;




×







2102



    }










2103













UNCOV

2104



    RAW_RETURN_CHECK(taosHashPut(pNameHash, pName->tname, strlen(pName->tname), &info, sizeof(tbInfo)));





×








UNCOV

2105



    tmpInfo = (tbInfo*)taosHashGet(pNameHash, pName->tname, strlen(pName->tname));




×








UNCOV

2106



    RAW_RETURN_CHECK(





×







2107



        taosHashPut(pVgHash, &info.vgInfo.vgId, sizeof(info.vgInfo.vgId), &info.vgInfo, sizeof(SVgroupInfo)));










2108



  }










2109













UNCOV

2110



  if (pTableMeta == NULL || retry > 0) {





×








UNCOV

2111



    STableMeta** pTableMetaTmp = (STableMeta**)taosHashGet(pMetaHash, &tmpInfo->suid, LONG_BYTES);




×








UNCOV

2112



    if (pTableMetaTmp == NULL || retry > 0 || needRefreshMeta(rawData, *pTableMetaTmp, pSW)) {





×








UNCOV

2113



      RAW_RETURN_CHECK(catalogGetTableMeta(pCatalog, conn, pName, &pTableMeta));





×








UNCOV

2114



      code = taosHashPut(pMetaHash, &tmpInfo->suid, LONG_BYTES, &pTableMeta, POINTER_BYTES);




×








UNCOV

2115



      if (code != 0) {





×







2116



        taosMemoryFree(pTableMeta);





×







2117



        goto end;




×







2118



      }










2119












2120



    } else {











UNCOV

2121



      pTableMeta = *pTableMetaTmp;




×








UNCOV

2122



      pTableMeta->uid = tmpInfo->uid;




×








UNCOV

2123



      pTableMeta->vgId = tmpInfo->vgInfo.vgId;




×







2124



    }










2125



  }











UNCOV

2126



  *pMeta = pTableMeta;




×







2127













UNCOV

2128



end:




×








UNCOV

2129



  return code;




×







2130



}










2131













UNCOV

2132



static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, uint32_t dataLen) {




×








UNCOV

2133



  if (taos == NULL || data == NULL) {





×







2134



    uError("invalid parameter in %s", __func__);





×







2135



    return TSDB_CODE_INVALID_PARA;




×







2136



  }











UNCOV

2137



  int32_t   code = TSDB_CODE_SUCCESS;




×








UNCOV

2138



  SQuery*   pQuery = NULL;




×








UNCOV

2139



  SMqRspObj rspObj = {0};




×








UNCOV

2140



  SDecoder  decoder = {0};




×







2141













UNCOV

2142



  SRequestObj*     pRequest = NULL;




×








UNCOV

2143



  SCatalog*        pCatalog = NULL;




×








UNCOV

2144



  SRequestConnInfo conn = {0};




×








UNCOV

2145



  RAW_RETURN_CHECK(buildRawRequest(taos, &pRequest, &pCatalog, &conn));





×








UNCOV

2146



  uDebug(LOG_ID_TAG " write raw data, data:%p, dataLen:%d", LOG_ID_VALUE, data, dataLen);





×








UNCOV

2147



  RAW_RETURN_CHECK(decodeRawData(&decoder, data, dataLen, tDecodeMqDataRsp, &rspObj));





×







2148













UNCOV

2149



  SHashObj* pVgHash = NULL;




×








UNCOV

2150



  SHashObj* pNameHash = NULL;




×








UNCOV

2151



  SHashObj* pMetaHash = NULL;




×








UNCOV

2152



  RAW_RETURN_CHECK(getRawCache(&pVgHash, &pNameHash, &pMetaHash, taos));





×








UNCOV

2153



  int retry = 0;




×







2154



  while (1) {











UNCOV

2155



    RAW_RETURN_CHECK(smlInitHandle(&pQuery));





×








UNCOV

2156



    uDebug(LOG_ID_TAG " write raw meta data block num:%d", LOG_ID_VALUE, rspObj.dataRsp.blockNum);





×








UNCOV

2157



    while (++rspObj.resIter < rspObj.dataRsp.blockNum) {





×








UNCOV

2158



      if (!rspObj.dataRsp.withSchema) {





×







2159



        goto end;




×







2160



      }










2161













UNCOV

2162



      const char* tbName = (const char*)taosArrayGetP(rspObj.dataRsp.blockTbName, rspObj.resIter);




×








UNCOV

2163



      RAW_NULL_CHECK(tbName);





×








UNCOV

2164



      SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(rspObj.dataRsp.blockSchema, rspObj.resIter);




×








UNCOV

2165



      RAW_NULL_CHECK(pSW);





×








UNCOV

2166



      void* pRetrieve = taosArrayGetP(rspObj.dataRsp.blockData, rspObj.resIter);




×








UNCOV

2167



      RAW_NULL_CHECK(pRetrieve);





×








UNCOV

2168



      void* rawData = getRawDataFromRes(pRetrieve);




×








UNCOV

2169



      RAW_NULL_CHECK(rawData);





×







2170













UNCOV

2171



      uDebug(LOG_ID_TAG " write raw data block tbname:%s", LOG_ID_VALUE, tbName);





×








UNCOV

2172



      SName pName = {TSDB_TABLE_NAME_T, pRequest->pTscObj->acctId, {0}, {0}};




×








UNCOV

2173



      tstrncpy(pName.dbname, pRequest->pDb, TSDB_DB_NAME_LEN);




×








UNCOV

2174



      tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN);




×







2175













UNCOV

2176



      STableMeta* pTableMeta = NULL;




×








UNCOV

2177



      RAW_RETURN_CHECK(processCacheMeta(pVgHash, pNameHash, pMetaHash, NULL, pCatalog, &conn, &pName, &pTableMeta, pSW,





×







2178



                                        rawData, retry));











UNCOV

2179



      char err[ERR_MSG_LEN] = {0};




×








UNCOV

2180



      code = rawBlockBindData(pQuery, pTableMeta, rawData, NULL, pSW, pSW->nCols, true, err, ERR_MSG_LEN, true);




×








UNCOV

2181



      if (code != TSDB_CODE_SUCCESS) {





×







2182



        SET_ERROR_MSG("table:%s, err:%s", pName.tname, err);




×







2183



        goto end;




×







2184



      }










2185



    }











UNCOV

2186



    RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));





×








UNCOV

2187



    launchQueryImpl(pRequest, pQuery, true, NULL);




×








UNCOV

2188



    code = pRequest->code;




×







2189













UNCOV

2190



    if (NEED_CLIENT_HANDLE_ERROR(code) && retry++ < 3) {





×







2191



      uInfo("write raw retry:%d/3 end code:%d, msg:%s", retry, code, tstrerror(code));





×







2192



      qDestroyQuery(pQuery);




×







2193



      pQuery = NULL;




×







2194



      rspObj.resIter = -1;




×







2195



      continue;




×







2196



    }











UNCOV

2197



    break;




×







2198



  }










2199













UNCOV

2200



end:




×








UNCOV

2201



  uDebug(LOG_ID_TAG " write raw data return, msg:%s", LOG_ID_VALUE, tstrerror(code));





×








UNCOV

2202



  tDeleteMqDataRsp(&rspObj.dataRsp);




×








UNCOV

2203



  tDecoderClear(&decoder);




×








UNCOV

2204



  qDestroyQuery(pQuery);




×








UNCOV

2205



  destroyRequest(pRequest);




×








UNCOV

2206



  return code;




×







2207



}










2208













UNCOV

2209



static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, uint32_t dataLen) {




×








UNCOV

2210



  if (taos == NULL || data == NULL) {





×







2211



    uError("invalid parameter in %s", __func__);





×







2212



    return TSDB_CODE_INVALID_PARA;




×







2213



  }











UNCOV

2214



  int32_t   code = TSDB_CODE_SUCCESS;




×








UNCOV

2215



  SQuery*   pQuery = NULL;




×








UNCOV

2216



  SMqRspObj rspObj = {0};




×








UNCOV

2217



  SDecoder  decoder = {0};




×








UNCOV

2218



  SHashObj* pCreateTbHash = NULL;




×







2219













UNCOV

2220



  SRequestObj*     pRequest = NULL;




×








UNCOV

2221



  SCatalog*        pCatalog = NULL;




×








UNCOV

2222



  SRequestConnInfo conn = {0};




×







2223













UNCOV

2224



  RAW_RETURN_CHECK(buildRawRequest(taos, &pRequest, &pCatalog, &conn));





×








UNCOV

2225



  uDebug(LOG_ID_TAG " write raw metadata, data:%p, dataLen:%d", LOG_ID_VALUE, data, dataLen);





×








UNCOV

2226



  RAW_RETURN_CHECK(decodeRawData(&decoder, data, dataLen, tDecodeSTaosxRsp, &rspObj));





×







2227













UNCOV

2228



  pCreateTbHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);




×








UNCOV

2229



  RAW_NULL_CHECK(pCreateTbHash);





×








UNCOV

2230



  RAW_RETURN_CHECK(buildCreateTbMap(&rspObj.dataRsp, pCreateTbHash));





×







2231













UNCOV

2232



  SHashObj* pVgHash = NULL;




×








UNCOV

2233



  SHashObj* pNameHash = NULL;




×








UNCOV

2234



  SHashObj* pMetaHash = NULL;




×








UNCOV

2235



  RAW_RETURN_CHECK(getRawCache(&pVgHash, &pNameHash, &pMetaHash, taos));





×








UNCOV

2236



  int retry = 0;




×







2237



  while (1) {











UNCOV

2238



    RAW_RETURN_CHECK(smlInitHandle(&pQuery));





×








UNCOV

2239



    uDebug(LOG_ID_TAG " write raw meta data block num:%d", LOG_ID_VALUE, rspObj.dataRsp.blockNum);





×








UNCOV

2240



    while (++rspObj.resIter < rspObj.dataRsp.blockNum) {





×








UNCOV

2241



      if (!rspObj.dataRsp.withSchema) {





×







2242



        goto end;




×







2243



      }










2244













UNCOV

2245



      const char* tbName = (const char*)taosArrayGetP(rspObj.dataRsp.blockTbName, rspObj.resIter);




×








UNCOV

2246



      RAW_NULL_CHECK(tbName);





×








UNCOV

2247



      SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(rspObj.dataRsp.blockSchema, rspObj.resIter);




×








UNCOV

2248



      RAW_NULL_CHECK(pSW);





×








UNCOV

2249



      void* pRetrieve = taosArrayGetP(rspObj.dataRsp.blockData, rspObj.resIter);




×








UNCOV

2250



      RAW_NULL_CHECK(pRetrieve);





×








UNCOV

2251



      void* rawData = getRawDataFromRes(pRetrieve);




×








UNCOV

2252



      RAW_NULL_CHECK(rawData);





×







2253













UNCOV

2254



      uDebug(LOG_ID_TAG " write raw data block tbname:%s", LOG_ID_VALUE, tbName);





×








UNCOV

2255



      SName pName = {TSDB_TABLE_NAME_T, pRequest->pTscObj->acctId, {0}, {0}};




×








UNCOV

2256



      tstrncpy(pName.dbname, pRequest->pDb, TSDB_DB_NAME_LEN);




×








UNCOV

2257



      tstrncpy(pName.tname, tbName, TSDB_TABLE_NAME_LEN);




×







2258












2259



      // find schema data info











UNCOV

2260



      SVCreateTbReq* pCreateReqDst = (SVCreateTbReq*)taosHashGet(pCreateTbHash, pName.tname, strlen(pName.tname));




×








UNCOV

2261



      STableMeta*    pTableMeta = NULL;




×








UNCOV

2262



      RAW_RETURN_CHECK(processCacheMeta(pVgHash, pNameHash, pMetaHash, pCreateReqDst, pCatalog, &conn, &pName,





×







2263



                                        &pTableMeta, pSW, rawData, retry));











UNCOV

2264



      char err[ERR_MSG_LEN] = {0};




×







2265



      code =











UNCOV

2266



          rawBlockBindData(pQuery, pTableMeta, rawData, pCreateReqDst, pSW, pSW->nCols, true, err, ERR_MSG_LEN, true);




×








UNCOV

2267



      if (code != TSDB_CODE_SUCCESS) {





×







2268



        SET_ERROR_MSG("table:%s, err:%s", pName.tname, err);




×







2269



        goto end;




×







2270



      }










2271



    }











UNCOV

2272



    RAW_RETURN_CHECK(smlBuildOutput(pQuery, pVgHash));





×








UNCOV

2273



    launchQueryImpl(pRequest, pQuery, true, NULL);




×








UNCOV

2274



    code = pRequest->code;




×







2275













UNCOV

2276



    if (NEED_CLIENT_HANDLE_ERROR(code) && retry++ < 3) {





×







2277



      uInfo("write raw retry:%d/3 end code:%d, msg:%s", retry, code, tstrerror(code));





×







2278



      qDestroyQuery(pQuery);




×







2279



      pQuery = NULL;




×







2280



      rspObj.resIter = -1;




×







2281



      continue;




×







2282



    }











UNCOV

2283



    break;




×







2284



  }










2285













UNCOV

2286



end:




×








UNCOV

2287



  uDebug(LOG_ID_TAG " write raw metadata return, msg:%s", LOG_ID_VALUE, tstrerror(code));





×








UNCOV

2288



  tDeleteSTaosxRsp(&rspObj.dataRsp);




×








UNCOV

2289



  void* pIter = taosHashIterate(pCreateTbHash, NULL);




×








UNCOV

2290



  while (pIter) {





×








UNCOV

2291



    tDestroySVCreateTbReq(pIter, TSDB_MSG_FLG_DECODE);




×








UNCOV

2292



    pIter = taosHashIterate(pCreateTbHash, pIter);




×







2293



  }











UNCOV

2294



  taosHashCleanup(pCreateTbHash);




×








UNCOV

2295



  tDecoderClear(&decoder);




×








UNCOV

2296



  qDestroyQuery(pQuery);




×








UNCOV

2297



  destroyRequest(pRequest);




×








UNCOV

2298



  return code;




×







2299



}










2300













UNCOV

2301



static void processSimpleMeta(SMqMetaRsp* pMetaRsp, cJSON** meta) {




×








UNCOV

2302



  if (pMetaRsp == NULL || meta == NULL) {





×







2303



    uError("invalid parameter in %s", __func__);





×







2304



    return;




×







2305



  }











UNCOV

2306



  if (pMetaRsp->resMsgType == TDMT_VND_CREATE_STB) {





×








UNCOV

2307



    processCreateStb(pMetaRsp, meta);




×








UNCOV

2308



  } else if (pMetaRsp->resMsgType == TDMT_VND_ALTER_STB) {





×








UNCOV

2309



    processAlterStb(pMetaRsp, meta);




×








UNCOV

2310



  } else if (pMetaRsp->resMsgType == TDMT_VND_DROP_STB) {





×








UNCOV

2311



    processDropSTable(pMetaRsp, meta);




×








UNCOV

2312



  } else if (pMetaRsp->resMsgType == TDMT_VND_CREATE_TABLE) {





×








UNCOV

2313



    processCreateTable(pMetaRsp, meta);




×








UNCOV

2314



  } else if (pMetaRsp->resMsgType == TDMT_VND_ALTER_TABLE) {





×








UNCOV

2315



    processAlterTable(pMetaRsp, meta);




×








UNCOV

2316



  } else if (pMetaRsp->resMsgType == TDMT_VND_DROP_TABLE) {





×








UNCOV

2317



    processDropTable(pMetaRsp, meta);




×








UNCOV

2318



  } else if (pMetaRsp->resMsgType == TDMT_VND_DROP_TABLE) {





×







2319



    processDropTable(pMetaRsp, meta);




×








UNCOV

2320



  } else if (pMetaRsp->resMsgType == TDMT_VND_DELETE) {





×








UNCOV

2321



    processDeleteTable(pMetaRsp, meta);




×







2322



  }










2323



}










2324













UNCOV

2325



static void processBatchMetaToJson(SMqBatchMetaRsp* pMsgRsp, char** string) {




×








UNCOV

2326



  if (pMsgRsp == NULL || string == NULL) {





×







2327



    uError("invalid parameter in %s", __func__);





×








UNCOV

2328



    return;




×







2329



  }










2330



  SDecoder        coder;











UNCOV

2331



  SMqBatchMetaRsp rsp = {0};




×








UNCOV

2332



  int32_t         code = 0;




×








UNCOV

2333



  cJSON*          pJson = NULL;




×








UNCOV

2334



  tDecoderInit(&coder, pMsgRsp->pMetaBuff, pMsgRsp->metaBuffLen);




×








UNCOV

2335



  if (tDecodeMqBatchMetaRsp(&coder, &rsp) < 0) {





×







2336



    goto end;




×







2337



  }










2338













UNCOV

2339



  pJson = cJSON_CreateObject();




×








UNCOV

2340



  RAW_NULL_CHECK(pJson);





×








UNCOV

2341



  RAW_FALSE_CHECK(cJSON_AddStringToObject(pJson, "tmq_meta_version", TMQ_META_VERSION));





×








UNCOV

2342



  cJSON* pMetaArr = cJSON_CreateArray();




×








UNCOV

2343



  RAW_NULL_CHECK(pMetaArr);





×








UNCOV

2344



  RAW_FALSE_CHECK(tmqAddJsonObjectItem(pJson, "metas", pMetaArr));





×







2345













UNCOV

2346



  int32_t num = taosArrayGetSize(rsp.batchMetaReq);




×








UNCOV

2347



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





×








UNCOV

2348



    int32_t* len = taosArrayGet(rsp.batchMetaLen, i);




×








UNCOV

2349



    RAW_NULL_CHECK(len);





×








UNCOV

2350



    void* tmpBuf = taosArrayGetP(rsp.batchMetaReq, i);




×








UNCOV

2351



    RAW_NULL_CHECK(tmpBuf);





×








UNCOV

2352



    SDecoder   metaCoder = {0};




×








UNCOV

2353



    SMqMetaRsp metaRsp = {0};




×








UNCOV

2354



    tDecoderInit(&metaCoder, POINTER_SHIFT(tmpBuf, sizeof(SMqRspHead)), *len - sizeof(SMqRspHead));




×








UNCOV

2355



    if (tDecodeMqMetaRsp(&metaCoder, &metaRsp) < 0) {





×







2356



      goto end;




×







2357



    }











UNCOV

2358



    cJSON* pItem = NULL;




×








UNCOV

2359



    processSimpleMeta(&metaRsp, &pItem);




×








UNCOV

2360



    tDeleteMqMetaRsp(&metaRsp);




×








UNCOV

2361



    RAW_FALSE_CHECK(tmqAddJsonArrayItem(pMetaArr, pItem));





×







2362



  }










2363













UNCOV

2364



  tDeleteMqBatchMetaRsp(&rsp);




×








UNCOV

2365



  char* fullStr = cJSON_PrintUnformatted(pJson);




×








UNCOV

2366



  cJSON_Delete(pJson);




×








UNCOV

2367



  *string = fullStr;




×








UNCOV

2368



  return;




×







2369












2370



end:




×







2371



  cJSON_Delete(pJson);




×







2372



  tDeleteMqBatchMetaRsp(&rsp);




×







2373



}










2374













UNCOV

2375



char* tmq_get_json_meta(TAOS_RES* res) {




×








UNCOV

2376



  if (res == NULL) {





×







2377



    uError("invalid parameter in %s", __func__);





×







2378



    return NULL;




×







2379



  }











UNCOV

2380



  uDebug("tmq_get_json_meta res:%p", res);





×








UNCOV

2381



  if (!TD_RES_TMQ_META(res) && !TD_RES_TMQ_METADATA(res) && !TD_RES_TMQ_BATCH_META(res)) {





×







2382



    return NULL;




×







2383



  }










2384













UNCOV

2385



  char*      string = NULL;




×








UNCOV

2386



  SMqRspObj* rspObj = (SMqRspObj*)res;




×








UNCOV

2387



  if (TD_RES_TMQ_METADATA(res)) {





×








UNCOV

2388



    processAutoCreateTable(&rspObj->dataRsp, &string);




×








UNCOV

2389



  } else if (TD_RES_TMQ_BATCH_META(res)) {





×








UNCOV

2390



    processBatchMetaToJson(&rspObj->batchMetaRsp, &string);




×








UNCOV

2391



  } else if (TD_RES_TMQ_META(res)) {





×








UNCOV

2392



    cJSON* pJson = NULL;




×








UNCOV

2393



    processSimpleMeta(&rspObj->metaRsp, &pJson);




×








UNCOV

2394



    string = cJSON_PrintUnformatted(pJson);




×








UNCOV

2395



    cJSON_Delete(pJson);




×







2396



  } else {










2397



    uError("tmq_get_json_meta res:%d, invalid type", *(int8_t*)res);





×







2398



  }










2399













UNCOV

2400



  uDebug("tmq_get_json_meta string:%s", string);





×








UNCOV

2401



  return string;




×







2402



}










2403













UNCOV

2404



void tmq_free_json_meta(char* jsonMeta) { taosMemoryFreeClear(jsonMeta); }





×







2405













UNCOV

2406



static int32_t getOffSetLen(const SMqDataRsp* pRsp) {




×








UNCOV

2407



  if (pRsp == NULL) {





×







2408



    uError("invalid parameter in %s", __func__);





×







2409



    return TSDB_CODE_INVALID_PARA;




×







2410



  }











UNCOV

2411



  SEncoder coder = {0};




×








UNCOV

2412



  tEncoderInit(&coder, NULL, 0);




×








UNCOV

2413



  if (tEncodeSTqOffsetVal(&coder, &pRsp->reqOffset) < 0) return -1;





×








UNCOV

2414



  if (tEncodeSTqOffsetVal(&coder, &pRsp->rspOffset) < 0) return -1;





×








UNCOV

2415



  int32_t pos = coder.pos;




×








UNCOV

2416



  tEncoderClear(&coder);




×








UNCOV

2417



  return pos;




×







2418



}










2419












2420



typedef int32_t __encode_func__(SEncoder* pEncoder, const SMqDataRsp* pRsp);











UNCOV

2421



static int32_t  encodeMqDataRsp(__encode_func__* encodeFunc, SMqDataRsp* rspObj, tmq_raw_data* raw) {




×








UNCOV

2422



  if (raw == NULL || encodeFunc == NULL || rspObj == NULL) {





×







2423



    uError("invalid parameter in %s", __func__);





×







2424



    return TSDB_CODE_INVALID_PARA;




×







2425



  }











UNCOV

2426



  uint32_t len = 0;




×








UNCOV

2427



  int32_t  code = 0;




×








UNCOV

2428



  SEncoder encoder = {0};




×








UNCOV

2429



  void*    buf = NULL;




×








UNCOV

2430



  tEncodeSize(encodeFunc, rspObj, len, code);





×








UNCOV

2431



  if (code < 0) {





×







2432



    code = TSDB_CODE_INVALID_MSG;




×







2433



    goto FAILED;




×







2434



  }











UNCOV

2435



  len += sizeof(int8_t) + sizeof(int32_t);




×








UNCOV

2436



  buf = taosMemoryCalloc(1, len);





×








UNCOV

2437



  if (buf == NULL) {





×







2438



    code = terrno;




×







2439



    goto FAILED;




×







2440



  }











UNCOV

2441



  tEncoderInit(&encoder, buf, len);




×








UNCOV

2442



  if (tEncodeI8(&encoder, MQ_DATA_RSP_VERSION) < 0) {





×







2443



    code = TSDB_CODE_INVALID_MSG;




×







2444



    goto FAILED;




×







2445



  }











UNCOV

2446



  int32_t offsetLen = getOffSetLen(rspObj);




×








UNCOV

2447



  if (offsetLen <= 0) {





×







2448



    code = TSDB_CODE_INVALID_MSG;




×







2449



    goto FAILED;




×







2450



  }











UNCOV

2451



  if (tEncodeI32(&encoder, offsetLen) < 0) {





×







2452



    code = TSDB_CODE_INVALID_MSG;




×







2453



    goto FAILED;




×







2454



  }











UNCOV

2455



  if (encodeFunc(&encoder, rspObj) < 0) {





×







2456



    code = TSDB_CODE_INVALID_MSG;




×







2457



    goto FAILED;




×







2458



  }











UNCOV

2459



  tEncoderClear(&encoder);




×







2460













UNCOV

2461



  raw->raw = buf;




×








UNCOV

2462



  raw->raw_len = len;




×








UNCOV

2463



  return code;




×







2464



FAILED:




×







2465



  tEncoderClear(&encoder);




×







2466



  taosMemoryFree(buf);





×







2467



  return code;




×







2468



}










2469













UNCOV

2470



int32_t tmq_get_raw(TAOS_RES* res, tmq_raw_data* raw) {




×








UNCOV

2471



  if (raw == NULL || res == NULL) {





×







2472



    uError("invalid parameter in %s", __func__);





×







2473



    return TSDB_CODE_INVALID_PARA;




×







2474



  }











UNCOV

2475



  SMqRspObj* rspObj = ((SMqRspObj*)res);




×








UNCOV

2476



  if (TD_RES_TMQ_META(res)) {





×








UNCOV

2477



    raw->raw = rspObj->metaRsp.metaRsp;




×








UNCOV

2478



    raw->raw_len = rspObj->metaRsp.metaRspLen >= 0 ? rspObj->metaRsp.metaRspLen : 0;




×








UNCOV

2479



    raw->raw_type = rspObj->metaRsp.resMsgType;




×








UNCOV

2480



    uDebug("tmq get raw type meta:%p", raw);





×








UNCOV

2481



  } else if (TD_RES_TMQ(res)) {





×








UNCOV

2482



    int32_t code = encodeMqDataRsp(tEncodeMqDataRsp, &rspObj->dataRsp, raw);




×








UNCOV

2483



    if (code != 0) {





×







2484



      uError("tmq get raw type error:%d", terrno);





×







2485



      return code;




×







2486



    }











UNCOV

2487



    raw->raw_type = RES_TYPE__TMQ;




×








UNCOV

2488



    uDebug("tmq get raw type data:%p", raw);





×








UNCOV

2489



  } else if (TD_RES_TMQ_METADATA(res)) {





×








UNCOV

2490



    int32_t code = encodeMqDataRsp(tEncodeSTaosxRsp, &rspObj->dataRsp, raw);




×








UNCOV

2491



    if (code != 0) {





×







2492



      uError("tmq get raw type error:%d", terrno);





×







2493



      return code;




×







2494



    }











UNCOV

2495



    raw->raw_type = RES_TYPE__TMQ_METADATA;




×








UNCOV

2496



    uDebug("tmq get raw type metadata:%p", raw);





×








UNCOV

2497



  } else if (TD_RES_TMQ_BATCH_META(res)) {





×








UNCOV

2498



    raw->raw = rspObj->batchMetaRsp.pMetaBuff;




×








UNCOV

2499



    raw->raw_len = rspObj->batchMetaRsp.metaBuffLen;




×








UNCOV

2500



    raw->raw_type = rspObj->resType;




×








UNCOV

2501



    uDebug("tmq get raw batch meta:%p", raw);





×







2502



  } else {










2503



    uError("tmq get raw error type:%d", *(int8_t*)res);





×







2504



    return TSDB_CODE_TMQ_INVALID_MSG;




×







2505



  }











UNCOV

2506



  return TSDB_CODE_SUCCESS;




×







2507



}










2508













UNCOV

2509



void tmq_free_raw(tmq_raw_data raw) {




×








UNCOV

2510



  uDebug("tmq free raw data type:%d", raw.raw_type);





×








UNCOV

2511



  if (raw.raw_type == RES_TYPE__TMQ || raw.raw_type == RES_TYPE__TMQ_METADATA) {





×








UNCOV

2512



    taosMemoryFree(raw.raw);





×







2513



  }











UNCOV

2514



  (void)memset(terrMsg, 0, ERR_MSG_LEN);




×








UNCOV

2515



}




×







2516













UNCOV

2517



static int32_t writeRawInit() {




×








UNCOV

2518



  while (atomic_load_8(&initedFlag) == WRITE_RAW_INIT_START) {





×








UNCOV

2519



    int8_t old = atomic_val_compare_exchange_8(&initFlag, 0, 1);




×








UNCOV

2520



    if (old == 0) {





×








UNCOV

2521



      int32_t code = initRawCacheHash();




×








UNCOV

2522



      if (code != 0) {





×







2523



        uError("tmq writeRawImpl init error:%d", code);





×







2524



        atomic_store_8(&initedFlag, WRITE_RAW_INIT_FAIL);




×







2525



        return code;




×







2526



      }











UNCOV

2527



      atomic_store_8(&initedFlag, WRITE_RAW_INIT_OK);




×







2528



    }










2529



  }










2530













UNCOV

2531



  if (atomic_load_8(&initedFlag) == WRITE_RAW_INIT_FAIL) {





×







2532



    return TSDB_CODE_INTERNAL_ERROR;




×







2533



  }











UNCOV

2534



  return 0;




×







2535



}










2536













UNCOV

2537



static int32_t writeRawImpl(TAOS* taos, void* buf, uint32_t len, uint16_t type) {




×








UNCOV

2538



  if (taos == NULL || buf == NULL) {





×







2539



    uError("invalid parameter in %s", __func__);





×







2540



    return TSDB_CODE_INVALID_PARA;




×







2541



  }











UNCOV

2542



  if (writeRawInit() != 0) {





×







2543



    return TSDB_CODE_INTERNAL_ERROR;




×







2544



  }










2545













UNCOV

2546



  if (type == TDMT_VND_CREATE_STB) {





×








UNCOV

2547



    return taosCreateStb(taos, buf, len);




×








UNCOV

2548



  } else if (type == TDMT_VND_ALTER_STB) {





×








UNCOV

2549



    return taosCreateStb(taos, buf, len);




×








UNCOV

2550



  } else if (type == TDMT_VND_DROP_STB) {





×








UNCOV

2551



    return taosDropStb(taos, buf, len);




×








UNCOV

2552



  } else if (type == TDMT_VND_CREATE_TABLE) {





×








UNCOV

2553



    return taosCreateTable(taos, buf, len);




×








UNCOV

2554



  } else if (type == TDMT_VND_ALTER_TABLE) {





×








UNCOV

2555



    return taosAlterTable(taos, buf, len);




×








UNCOV

2556



  } else if (type == TDMT_VND_DROP_TABLE) {





×








UNCOV

2557



    return taosDropTable(taos, buf, len);




×








UNCOV

2558



  } else if (type == TDMT_VND_DELETE) {





×








UNCOV

2559



    return taosDeleteData(taos, buf, len);




×








UNCOV

2560



  } else if (type == RES_TYPE__TMQ_METADATA) {





×








UNCOV

2561



    return tmqWriteRawMetaDataImpl(taos, buf, len);




×








UNCOV

2562



  } else if (type == RES_TYPE__TMQ) {





×








UNCOV

2563



    return tmqWriteRawDataImpl(taos, buf, len);




×








UNCOV

2564



  } else if (type == RES_TYPE__TMQ_BATCH_META) {





×








UNCOV

2565



    return tmqWriteBatchMetaDataImpl(taos, buf, len);




×







2566



  }










2567



  return TSDB_CODE_INVALID_PARA;




×







2568



}










2569













UNCOV

2570



int32_t tmq_write_raw(TAOS* taos, tmq_raw_data raw) {




×








UNCOV

2571



  if (taos == NULL || raw.raw == NULL || raw.raw_len <= 0) {





×








UNCOV

2572



    SET_ERROR_MSG("taos:%p or data:%p is NULL or raw_len <= 0", taos, raw.raw);




×








UNCOV

2573



    return TSDB_CODE_INVALID_PARA;




×







2574



  }










2575













UNCOV

2576



  return writeRawImpl(taos, raw.raw, raw.raw_len, raw.raw_type);




×







2577



}










2578













UNCOV

2579



static int32_t tmqWriteBatchMetaDataImpl(TAOS* taos, void* meta, uint32_t metaLen) {




×








UNCOV

2580



  if (taos == NULL || meta == NULL) {





×







2581



    uError("invalid parameter in %s", __func__);





×







2582



    return TSDB_CODE_INVALID_PARA;




×







2583



  }











UNCOV

2584



  SMqBatchMetaRsp rsp = {0};




×








UNCOV

2585



  SDecoder        coder = {0};




×








UNCOV

2586



  int32_t         code = TSDB_CODE_SUCCESS;




×







2587












2588



  // decode and process req











UNCOV

2589



  tDecoderInit(&coder, meta, metaLen);




×








UNCOV

2590



  if (tDecodeMqBatchMetaRsp(&coder, &rsp) < 0) {





×







2591



    code = TSDB_CODE_INVALID_PARA;




×







2592



    goto end;




×







2593



  }











UNCOV

2594



  int32_t num = taosArrayGetSize(rsp.batchMetaReq);




×








UNCOV

2595



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





×








UNCOV

2596



    int32_t* len = taosArrayGet(rsp.batchMetaLen, i);




×








UNCOV

2597



    RAW_NULL_CHECK(len);





×








UNCOV

2598



    void* tmpBuf = taosArrayGetP(rsp.batchMetaReq, i);




×








UNCOV

2599



    RAW_NULL_CHECK(tmpBuf);





×








UNCOV

2600



    SDecoder   metaCoder = {0};




×








UNCOV

2601



    SMqMetaRsp metaRsp = {0};




×








UNCOV

2602



    tDecoderInit(&metaCoder, POINTER_SHIFT(tmpBuf, sizeof(SMqRspHead)), *len - sizeof(SMqRspHead));




×








UNCOV

2603



    if (tDecodeMqMetaRsp(&metaCoder, &metaRsp) < 0) {





×







2604



      code = TSDB_CODE_INVALID_PARA;




×







2605



      goto end;




×







2606



    }











UNCOV

2607



    code = writeRawImpl(taos, metaRsp.metaRsp, metaRsp.metaRspLen, metaRsp.resMsgType);




×








UNCOV

2608



    tDeleteMqMetaRsp(&metaRsp);




×








UNCOV

2609



    if (code != TSDB_CODE_SUCCESS) {





×







2610



      goto end;




×







2611



    }










2612



  }










2613













UNCOV

2614



end:




×








UNCOV

2615



  tDeleteMqBatchMetaRsp(&rsp);




×








UNCOV

2616



  return code;




×







2617



}























    

  • 

    •