taosdata/TDengine | Build 3562 | source/dnode/mnode/impl/src/mndTopic.c | Coveralls

59

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

74

void mndTopicGetShowName(const char* fullTopic, char* topic) {

×

UNCOV

75

  if (fullTopic == NULL) {

×

76

    return;

×

UNCOV

78

  char* tmp = strchr(fullTopic, '.');

×

UNCOV

79

  if (tmp == NULL) {

×

80

    tstrncpy(topic, fullTopic, TSDB_TOPIC_FNAME_LEN);

×

UNCOV

82

    tstrncpy(topic, tmp+1, TSDB_TOPIC_FNAME_LEN);

×

UNCOV

86

SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {

×

UNCOV

87

  if (pTopic == NULL) {

×

88

    return NULL;

×

UNCOV

90

  int32_t code = 0;

×

UNCOV

91

  int32_t lino = 0;

×

UNCOV

92

  terrno = TSDB_CODE_OUT_OF_MEMORY;

×

UNCOV

94

  void   *swBuf = NULL;

×

UNCOV

95

  int32_t physicalPlanLen = 0;

×

UNCOV

96

  if (pTopic->physicalPlan) {

×

UNCOV

97

    physicalPlanLen = strlen(pTopic->physicalPlan) + 1;

×

UNCOV

100

  int32_t schemaLen = 0;

×

UNCOV

101

  if (pTopic->schema.nCols) {

×

UNCOV

102

    schemaLen = taosEncodeSSchemaWrapper(NULL, &pTopic->schema);

×

UNCOV

104

  int32_t ntbColLen = taosArrayGetSize(pTopic->ntbColIds) * sizeof(int16_t);

×

UNCOV

106

  int32_t size = sizeof(SMqTopicObj) + physicalPlanLen + pTopic->sqlLen + pTopic->astLen + schemaLen + ntbColLen +

×

UNCOV

108

  SSdbRaw *pRaw = sdbAllocRaw(SDB_TOPIC, MND_TOPIC_VER_NUMBER, size);

×

UNCOV

109

  if (pRaw == NULL) {

×

110

    goto TOPIC_ENCODE_OVER;

×

UNCOV

113

  int32_t dataPos = 0;

×

UNCOV

114

  SDB_SET_BINARY(pRaw, dataPos, pTopic->name, TSDB_TOPIC_FNAME_LEN, TOPIC_ENCODE_OVER);

×

UNCOV

115

  SDB_SET_BINARY(pRaw, dataPos, pTopic->db, TSDB_DB_FNAME_LEN, TOPIC_ENCODE_OVER);

×

UNCOV

116

  SDB_SET_BINARY(pRaw, dataPos, pTopic->createUser, TSDB_USER_LEN, TOPIC_ENCODE_OVER);

×

UNCOV

117

  SDB_SET_INT64(pRaw, dataPos, pTopic->createTime, TOPIC_ENCODE_OVER);

×

UNCOV

118

  SDB_SET_INT64(pRaw, dataPos, pTopic->updateTime, TOPIC_ENCODE_OVER);

×

UNCOV

119

  SDB_SET_INT64(pRaw, dataPos, pTopic->uid, TOPIC_ENCODE_OVER);

×

UNCOV

120

  SDB_SET_INT64(pRaw, dataPos, pTopic->dbUid, TOPIC_ENCODE_OVER);

×

UNCOV

121

  SDB_SET_INT32(pRaw, dataPos, pTopic->version, TOPIC_ENCODE_OVER);

×

UNCOV

122

  SDB_SET_INT8(pRaw, dataPos, pTopic->subType, TOPIC_ENCODE_OVER);

×

UNCOV

123

  SDB_SET_INT8(pRaw, dataPos, pTopic->withMeta, TOPIC_ENCODE_OVER);

×

UNCOV

125

  SDB_SET_INT64(pRaw, dataPos, pTopic->stbUid, TOPIC_ENCODE_OVER);

×

UNCOV

126

  SDB_SET_BINARY(pRaw, dataPos, pTopic->stbName, TSDB_TABLE_FNAME_LEN, TOPIC_ENCODE_OVER);

×

UNCOV

127

  SDB_SET_INT32(pRaw, dataPos, pTopic->sqlLen, TOPIC_ENCODE_OVER);

×

UNCOV

128

  SDB_SET_BINARY(pRaw, dataPos, pTopic->sql, pTopic->sqlLen, TOPIC_ENCODE_OVER);

×

UNCOV

129

  SDB_SET_INT32(pRaw, dataPos, pTopic->astLen, TOPIC_ENCODE_OVER);

×

UNCOV

131

  if (pTopic->astLen) {

×

UNCOV

132

    SDB_SET_BINARY(pRaw, dataPos, pTopic->ast, pTopic->astLen, TOPIC_ENCODE_OVER);

×

UNCOV

134

  SDB_SET_INT32(pRaw, dataPos, physicalPlanLen, TOPIC_ENCODE_OVER);

×

UNCOV

135

  if (physicalPlanLen) {

×

UNCOV

136

    SDB_SET_BINARY(pRaw, dataPos, pTopic->physicalPlan, physicalPlanLen, TOPIC_ENCODE_OVER);

×

UNCOV

138

  SDB_SET_INT32(pRaw, dataPos, schemaLen, TOPIC_ENCODE_OVER);

×

UNCOV

139

  if (schemaLen) {

×

UNCOV

140

    swBuf = taosMemoryMalloc(schemaLen);

×

UNCOV

141

    if (swBuf == NULL) {

×

142

      goto TOPIC_ENCODE_OVER;

×

UNCOV

144

    void *aswBuf = swBuf;

×

UNCOV

145

    if(taosEncodeSSchemaWrapper(&aswBuf, &pTopic->schema) < 0){

×

146

      goto TOPIC_ENCODE_OVER;

×

UNCOV

148

    SDB_SET_BINARY(pRaw, dataPos, swBuf, schemaLen, TOPIC_ENCODE_OVER);

×

UNCOV

151

  SDB_SET_INT64(pRaw, dataPos, pTopic->ntbUid, TOPIC_ENCODE_OVER);

×

UNCOV

152

  if (pTopic->ntbUid != 0) {

×

UNCOV

153

    int32_t sz = taosArrayGetSize(pTopic->ntbColIds);

×

UNCOV

154

    SDB_SET_INT32(pRaw, dataPos, sz, TOPIC_ENCODE_OVER);

×

UNCOV

155

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

×

UNCOV

156

      int16_t colId = *(int16_t *)taosArrayGet(pTopic->ntbColIds, i);

×

UNCOV

157

      SDB_SET_INT16(pRaw, dataPos, colId, TOPIC_ENCODE_OVER);

×

UNCOV

161

  SDB_SET_INT64(pRaw, dataPos, pTopic->ctbStbUid, TOPIC_ENCODE_OVER);

×

UNCOV

163

  SDB_SET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_ENCODE_OVER);

×

UNCOV

164

  SDB_SET_DATALEN(pRaw, dataPos, TOPIC_ENCODE_OVER);

×

UNCOV

166

  terrno = TSDB_CODE_SUCCESS;

×

UNCOV

168

TOPIC_ENCODE_OVER:

×

UNCOV

169

  if (swBuf) taosMemoryFree(swBuf);

×

UNCOV

170

  if (terrno != TSDB_CODE_SUCCESS) {

×

171

    mError("topic:%s, failed to encode to raw:%p since %s", pTopic->name, pRaw, terrstr());

×

172

    sdbFreeRaw(pRaw);

×

173

    return NULL;

×

UNCOV

176

  mTrace("topic:%s, encode to raw:%p, row:%p", pTopic->name, pRaw, pTopic);

×

UNCOV

177

  return pRaw;

×

UNCOV

180

SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {

×

UNCOV

181

  if (pRaw == NULL) return NULL;

×

UNCOV

182

  int32_t code = 0;

×

UNCOV

183

  int32_t lino = 0;

×

UNCOV

184

  terrno = TSDB_CODE_OUT_OF_MEMORY;

×

UNCOV

185

  SSdbRow     *pRow = NULL;

×

UNCOV

186

  SMqTopicObj *pTopic = NULL;

×

UNCOV

187

  void        *buf = NULL;

×

UNCOV

189

  int8_t sver = 0;

×

UNCOV

190

  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto TOPIC_DECODE_OVER;

×

UNCOV

192

  if (sver < 1 || sver > MND_TOPIC_VER_NUMBER) {

×

193

    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;

×

194

    goto TOPIC_DECODE_OVER;

×

UNCOV

197

  pRow = sdbAllocRow(sizeof(SMqTopicObj));

×

UNCOV

198

  if (pRow == NULL) goto TOPIC_DECODE_OVER;

×

UNCOV

200

  pTopic = sdbGetRowObj(pRow);

×

UNCOV

201

  if (pTopic == NULL) goto TOPIC_DECODE_OVER;

×

UNCOV

203

  int32_t len = 0;

×

UNCOV

204

  int32_t dataPos = 0;

×

UNCOV

205

  SDB_GET_BINARY(pRaw, dataPos, pTopic->name, TSDB_TOPIC_FNAME_LEN, TOPIC_DECODE_OVER);

×

UNCOV

206

  SDB_GET_BINARY(pRaw, dataPos, pTopic->db, TSDB_DB_FNAME_LEN, TOPIC_DECODE_OVER);

×

UNCOV

207

  if (sver >= 2) {

×

UNCOV

208

    SDB_GET_BINARY(pRaw, dataPos, pTopic->createUser, TSDB_USER_LEN, TOPIC_DECODE_OVER);

×

UNCOV

210

  SDB_GET_INT64(pRaw, dataPos, &pTopic->createTime, TOPIC_DECODE_OVER);

×

UNCOV

211

  SDB_GET_INT64(pRaw, dataPos, &pTopic->updateTime, TOPIC_DECODE_OVER);

×

UNCOV

212

  SDB_GET_INT64(pRaw, dataPos, &pTopic->uid, TOPIC_DECODE_OVER);

×

UNCOV

213

  SDB_GET_INT64(pRaw, dataPos, &pTopic->dbUid, TOPIC_DECODE_OVER);

×

UNCOV

214

  SDB_GET_INT32(pRaw, dataPos, &pTopic->version, TOPIC_DECODE_OVER);

×

UNCOV

215

  SDB_GET_INT8(pRaw, dataPos, &pTopic->subType, TOPIC_DECODE_OVER);

×

UNCOV

216

  SDB_GET_INT8(pRaw, dataPos, &pTopic->withMeta, TOPIC_DECODE_OVER);

×

UNCOV

218

  SDB_GET_INT64(pRaw, dataPos, &pTopic->stbUid, TOPIC_DECODE_OVER);

×

UNCOV

219

  if (sver >= 3) {

×

UNCOV

220

    SDB_GET_BINARY(pRaw, dataPos, pTopic->stbName, TSDB_TABLE_FNAME_LEN, TOPIC_DECODE_OVER);

×

UNCOV

222

  SDB_GET_INT32(pRaw, dataPos, &pTopic->sqlLen, TOPIC_DECODE_OVER);

×

UNCOV

223

  pTopic->sql = taosMemoryCalloc(pTopic->sqlLen, sizeof(char));

×

UNCOV

224

  if (pTopic->sql == NULL) {

×

225

    terrno = TSDB_CODE_OUT_OF_MEMORY;

×

226

    goto TOPIC_DECODE_OVER;

×

UNCOV

228

  SDB_GET_BINARY(pRaw, dataPos, pTopic->sql, pTopic->sqlLen, TOPIC_DECODE_OVER);

×

UNCOV

230

  SDB_GET_INT32(pRaw, dataPos, &pTopic->astLen, TOPIC_DECODE_OVER);

×

UNCOV

231

  if (pTopic->astLen) {

×

UNCOV

232

    pTopic->ast = taosMemoryCalloc(pTopic->astLen, sizeof(char));

×

UNCOV

233

    if (pTopic->ast == NULL) {

×

234

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

235

      goto TOPIC_DECODE_OVER;

×

UNCOV

238

    pTopic->ast = NULL;

×

UNCOV

240

  SDB_GET_BINARY(pRaw, dataPos, pTopic->ast, pTopic->astLen, TOPIC_DECODE_OVER);

×

UNCOV

241

  SDB_GET_INT32(pRaw, dataPos, &len, TOPIC_DECODE_OVER);

×

UNCOV

242

  if (len) {

×

UNCOV

243

    pTopic->physicalPlan = taosMemoryCalloc(len, sizeof(char));

×

UNCOV

244

    if (pTopic->physicalPlan == NULL) {

×

245

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

246

      goto TOPIC_DECODE_OVER;

×

UNCOV

248

    SDB_GET_BINARY(pRaw, dataPos, pTopic->physicalPlan, len, TOPIC_DECODE_OVER);

×

UNCOV

250

    pTopic->physicalPlan = NULL;

×

UNCOV

253

  SDB_GET_INT32(pRaw, dataPos, &len, TOPIC_DECODE_OVER);

×

UNCOV

254

  if (len) {

×

UNCOV

255

    buf = taosMemoryMalloc(len);

×

UNCOV

256

    if (buf == NULL) {

×

257

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

258

      goto TOPIC_DECODE_OVER;

×

UNCOV

260

    SDB_GET_BINARY(pRaw, dataPos, buf, len, TOPIC_DECODE_OVER);

×

UNCOV

261

    if (taosDecodeSSchemaWrapper(buf, &pTopic->schema) == NULL) {

×

262

      goto TOPIC_DECODE_OVER;

×

UNCOV

265

    pTopic->schema.nCols = 0;

×

UNCOV

266

    pTopic->schema.version = 0;

×

UNCOV

267

    pTopic->schema.pSchema = NULL;

×

UNCOV

269

  SDB_GET_INT64(pRaw, dataPos, &pTopic->ntbUid, TOPIC_DECODE_OVER);

×

UNCOV

270

  if (pTopic->ntbUid != 0) {

×

UNCOV

272

    SDB_GET_INT32(pRaw, dataPos, &ntbColNum, TOPIC_DECODE_OVER);

×

UNCOV

273

    pTopic->ntbColIds = taosArrayInit(ntbColNum, sizeof(int16_t));

×

UNCOV

274

    if (pTopic->ntbColIds == NULL) {

×

275

      terrno = TSDB_CODE_OUT_OF_MEMORY;

×

276

      goto TOPIC_DECODE_OVER;

×

UNCOV

279

    SDB_GET_INT16(pRaw, dataPos, &colId, TOPIC_DECODE_OVER);

×

UNCOV

280

    if (taosArrayPush(pTopic->ntbColIds, &colId) == NULL) {

×

281

      goto TOPIC_DECODE_OVER;

×

UNCOV

285

  SDB_GET_INT64(pRaw, dataPos, &pTopic->ctbStbUid, TOPIC_DECODE_OVER);

×

UNCOV

286

  SDB_GET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_DECODE_OVER);

×

UNCOV

287

  terrno = TSDB_CODE_SUCCESS;

×

UNCOV

289

TOPIC_DECODE_OVER:

×

UNCOV

290

  taosMemoryFreeClear(buf);

×

UNCOV

291

  if (terrno != TSDB_CODE_SUCCESS) {

×

292

    mError("topic:%s, failed to decode from raw:%p since %s", pTopic == NULL ? "null" : pTopic->name, pRaw, terrstr());

×

293

    taosMemoryFreeClear(pRow);

×

294

    return NULL;

×

UNCOV

297

  mTrace("topic:%s, decode from raw:%p, row:%p", pTopic->name, pRaw, pTopic);

×

UNCOV

298

  return pRow;

×

UNCOV

301

static int32_t mndTopicActionInsert(SSdb *pSdb, SMqTopicObj *pTopic) {

×

UNCOV

302

  mTrace("topic:%s perform insert action", pTopic != NULL ? pTopic->name : "null");

×

UNCOV

303

  return 0;

×

UNCOV

306

static int32_t mndTopicActionDelete(SSdb *pSdb, SMqTopicObj *pTopic) {

×

UNCOV

307

  if (pTopic == NULL) return 0;

×

UNCOV

308

  mTrace("topic:%s perform delete action", pTopic->name);

×

UNCOV

309

  taosMemoryFreeClear(pTopic->sql);

×

UNCOV

310

  taosMemoryFreeClear(pTopic->ast);

×

UNCOV

311

  taosMemoryFreeClear(pTopic->physicalPlan);

×

UNCOV

312

  if (pTopic->schema.nCols) taosMemoryFreeClear(pTopic->schema.pSchema);

×

UNCOV

313

  taosArrayDestroy(pTopic->ntbColIds);

×

UNCOV

314

  return 0;

×

317

static int32_t mndTopicActionUpdate(SSdb *pSdb, SMqTopicObj *pOldTopic, SMqTopicObj *pNewTopic) {

×

318

  if (pOldTopic == NULL || pNewTopic == NULL) return 0;

×

319

  mTrace("topic:%s perform update action", pOldTopic->name);

×

320

  (void)atomic_exchange_64(&pOldTopic->updateTime, pNewTopic->updateTime);

×

321

  (void)atomic_exchange_32(&pOldTopic->version, pNewTopic->version);

×

323

  return 0;

×

UNCOV

326

int32_t mndAcquireTopic(SMnode *pMnode, const char *topicName, SMqTopicObj **pTopic) {

×

UNCOV

327

  if (pMnode == NULL || topicName == NULL || pTopic == NULL){

×

328

    return TSDB_CODE_INVALID_PARA;

×

UNCOV

330

  SSdb        *pSdb = pMnode->pSdb;

×

UNCOV

331

  *pTopic = sdbAcquire(pSdb, SDB_TOPIC, topicName);

×

UNCOV

332

  if (*pTopic == NULL) {

×

UNCOV

333

    return TSDB_CODE_MND_TOPIC_NOT_EXIST;

×

UNCOV

335

  return TDB_CODE_SUCCESS;

×

UNCOV

338

void mndReleaseTopic(SMnode *pMnode, SMqTopicObj *pTopic) {

×

UNCOV

339

  if (pMnode == NULL) return;

×

UNCOV

340

  SSdb *pSdb = pMnode->pSdb;

×

UNCOV

341

  sdbRelease(pSdb, pTopic);

×

UNCOV

344

static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {

×

UNCOV

345

  if (pCreate == NULL) return TSDB_CODE_INVALID_PARA;

×

UNCOV

346

  if (pCreate->sql == NULL) return TSDB_CODE_MND_INVALID_TOPIC;

×

UNCOV

348

  if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) {

×

UNCOV

349

    if (pCreate->ast == NULL || pCreate->ast[0] == 0) return TSDB_CODE_MND_INVALID_TOPIC;

×

UNCOV

350

  } else if (pCreate->subType == TOPIC_SUB_TYPE__TABLE) {

×

UNCOV

351

    if (pCreate->subStbName[0] == 0) return TSDB_CODE_MND_INVALID_TOPIC;

×

UNCOV

352

  } else if (pCreate->subType == TOPIC_SUB_TYPE__DB) {

×

UNCOV

353

    if (pCreate->subDbName[0] == 0) return TSDB_CODE_MND_INVALID_TOPIC;

×

UNCOV

356

  return 0;

×

UNCOV

359

static int32_t extractTopicTbInfo(SNode *pAst, SMqTopicObj *pTopic) {

×

UNCOV

360

  if (pAst == NULL || pTopic == NULL) return TSDB_CODE_INVALID_PARA;

×

UNCOV

361

  SNodeList *pNodeList = NULL;

×

UNCOV

362

  int32_t   code = 0;

×

UNCOV

363

  MND_TMQ_RETURN_CHECK(nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList));

×

UNCOV

364

  int64_t suid = ((SRealTableNode *)((SSelectStmt *)pAst)->pFromTable)->pMeta->suid;

×

UNCOV

365

  int8_t  tableType = ((SRealTableNode *)((SSelectStmt *)pAst)->pFromTable)->pMeta->tableType;

×

UNCOV

366

  if (tableType == TSDB_CHILD_TABLE) {

×

UNCOV

367

    pTopic->ctbStbUid = suid;

×

UNCOV

368

  } else if (tableType == TSDB_NORMAL_TABLE) {

×

UNCOV

369

    SNode *pNode = NULL;

×

UNCOV

370

    FOREACH(pNode, pNodeList) {

×

UNCOV

371

      SColumnNode *pCol = (SColumnNode *)pNode;

×

UNCOV

372

      if (pCol->tableType == TSDB_NORMAL_TABLE) {

×

UNCOV

373

        pTopic->ntbUid = pCol->tableId;

×

UNCOV

374

        MND_TMQ_NULL_CHECK(taosArrayPush(pTopic->ntbColIds, &pCol->colId));

×

UNCOV

378

  nodesDestroyList(pNodeList);

×

UNCOV

380

END:

×

UNCOV

381

  return code;

×

UNCOV

384

static int32_t sendCheckInfoToVnode(STrans *pTrans, SMnode *pMnode, SMqTopicObj *topicObj){

×

UNCOV

385

  if (pTrans == NULL || pMnode == NULL || topicObj == NULL) return TSDB_CODE_INVALID_PARA;

×

UNCOV

386

  STqCheckInfo info = {0};

×

UNCOV

387

  (void)memcpy(info.topic, topicObj->name, TSDB_TOPIC_FNAME_LEN);

×

UNCOV

388

  info.ntbUid = topicObj->ntbUid;

×

UNCOV

389

  info.colIdList = topicObj->ntbColIds;

×

UNCOV

391

  void   *pIter = NULL;

×

UNCOV

392

  SSdb   *pSdb = pMnode->pSdb;

×

UNCOV

393

  SVgObj *pVgroup = NULL;

×

UNCOV

394

  int32_t code = 0;

×

UNCOV

395

  void   *buf = NULL;

×

UNCOV

397

  while (1) {

×

UNCOV

399

    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);

×

UNCOV

400

    if (pIter == NULL) break;

×

UNCOV

401

    if (!mndVgroupInDb(pVgroup, topicObj->dbUid)) {

×

UNCOV

402

      sdbRelease(pSdb, pVgroup);

×

UNCOV

403

      continue;

×

UNCOV

407

    int32_t len = 0;

×

UNCOV

408

    tEncodeSize(tEncodeSTqCheckInfo, &info, len, code);

×

UNCOV

409

    if (code != 0) {

×

410

      code = TSDB_CODE_OUT_OF_MEMORY;

×

411

      goto END;

×

UNCOV

413

    buf = taosMemoryCalloc(1, sizeof(SMsgHead) + len);

×

UNCOV

414

    MND_TMQ_NULL_CHECK(buf);

×

UNCOV

415

    void    *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));

×

UNCOV

416

    SEncoder encoder = {0};

×

UNCOV

417

    tEncoderInit(&encoder, abuf, len);

×

UNCOV

418

    code = tEncodeSTqCheckInfo(&encoder, &info);

×

UNCOV

419

    if (code < 0) {

×

420

      tEncoderClear(&encoder);

×

421

      goto END;

×

UNCOV

423

    tEncoderClear(&encoder);

×

UNCOV

424

    ((SMsgHead *)buf)->vgId = htonl(pVgroup->vgId);

×

UNCOV

426

    STransAction action = {0};

×

UNCOV

427

    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

UNCOV

428

    action.pCont = buf;

×

UNCOV

429

    action.contLen = sizeof(SMsgHead) + len;

×

UNCOV

430

    action.msgType = TDMT_VND_TMQ_ADD_CHECKINFO;

×

UNCOV

431

    MND_TMQ_RETURN_CHECK(mndTransAppendRedoAction(pTrans, &action));

×

UNCOV

432

    sdbRelease(pSdb, pVgroup);

×

UNCOV

433

    buf = NULL;

×

UNCOV

436

END:

×

UNCOV

437

  taosMemoryFree(buf);

×

UNCOV

438

  sdbRelease(pSdb, pVgroup);

×

UNCOV

439

  sdbCancelFetch(pSdb, pIter);

×

UNCOV

440

  return code;

×

UNCOV

443

static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *pCreate, SDbObj *pDb,

×

UNCOV

445

  if (pMnode == NULL || pReq == NULL || pCreate == NULL || pDb == NULL || userName == NULL) return TSDB_CODE_INVALID_PARA;

×

UNCOV

446

  mInfo("start to create topic:%s", pCreate->name);

×

UNCOV

447

  STrans *pTrans = NULL;

×

UNCOV

448

  int32_t code = 0;

×

UNCOV

449

  SNode  *pAst = NULL;

×

UNCOV

450

  SQueryPlan *pPlan = NULL;

×

UNCOV

451

  SMqTopicObj topicObj = {0};

×

UNCOV

453

  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq, "create-topic");

×

UNCOV

454

  MND_TMQ_NULL_CHECK(pTrans);

×

UNCOV

455

  mndTransSetDbName(pTrans, pDb->name, NULL);

×

UNCOV

456

  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

×

UNCOV

457

  mInfo("trans:%d to create topic:%s", pTrans->id, pCreate->name);

×

UNCOV

459

  tstrncpy(topicObj.name, pCreate->name, TSDB_TOPIC_FNAME_LEN);

×

UNCOV

460

  tstrncpy(topicObj.db, pDb->name, TSDB_DB_FNAME_LEN);

×

UNCOV

461

  tstrncpy(topicObj.createUser, userName, TSDB_USER_LEN);

×

UNCOV

463

  MND_TMQ_RETURN_CHECK(mndCheckTopicPrivilege(pMnode, pReq->info.conn.user, MND_OPER_CREATE_TOPIC, &topicObj));

×

UNCOV

465

  topicObj.createTime = taosGetTimestampMs();

×

UNCOV

466

  topicObj.updateTime = topicObj.createTime;

×

UNCOV

467

  topicObj.uid = mndGenerateUid(pCreate->name, strlen(pCreate->name));

×

UNCOV

468

  topicObj.dbUid = pDb->uid;

×

UNCOV

469

  topicObj.version = 1;

×

UNCOV

470

  topicObj.sql = taosStrdup(pCreate->sql);

×

UNCOV

471

  MND_TMQ_NULL_CHECK(topicObj.sql);

×

UNCOV

472

  topicObj.sqlLen = strlen(pCreate->sql) + 1;

×

UNCOV

473

  topicObj.subType = pCreate->subType;

×

UNCOV

474

  topicObj.withMeta = pCreate->withMeta;

×

UNCOV

476

  if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) {

×

UNCOV

477

    topicObj.ast = taosStrdup(pCreate->ast);

×

UNCOV

478

    MND_TMQ_NULL_CHECK(topicObj.ast);

×

UNCOV

479

    topicObj.astLen = strlen(pCreate->ast) + 1;

×

UNCOV

480

    qDebugL("topic:%s ast %s", topicObj.name, topicObj.ast);

×

UNCOV

481

    MND_TMQ_RETURN_CHECK(nodesStringToNode(pCreate->ast, &pAst));

×

UNCOV

482

    SPlanContext cxt = {.pAstRoot = pAst, .topicQuery = true};

×

UNCOV

483

    MND_TMQ_RETURN_CHECK(qCreateQueryPlan(&cxt, &pPlan, NULL));

×

UNCOV

485

    topicObj.ntbColIds = taosArrayInit(0, sizeof(int16_t));

×

UNCOV

486

    MND_TMQ_NULL_CHECK(topicObj.ntbColIds);

×

UNCOV

487

    MND_TMQ_RETURN_CHECK(extractTopicTbInfo(pAst, &topicObj));

×

UNCOV

488

    if (topicObj.ntbUid == 0) {

×

UNCOV

489

      taosArrayDestroy(topicObj.ntbColIds);

×

UNCOV

490

      topicObj.ntbColIds = NULL;

×

UNCOV

493

    MND_TMQ_RETURN_CHECK(qExtractResultSchema(pAst, &topicObj.schema.nCols, &topicObj.schema.pSchema));

×

UNCOV

494

    MND_TMQ_RETURN_CHECK(nodesNodeToString((SNode *)pPlan, false, &topicObj.physicalPlan, NULL));

×

UNCOV

495

  } else if (pCreate->subType == TOPIC_SUB_TYPE__TABLE) {

×

UNCOV

496

    SStbObj *pStb = mndAcquireStb(pMnode, pCreate->subStbName);

×

UNCOV

497

    MND_TMQ_NULL_CHECK(pStb);

×

UNCOV

498

    tstrncpy(topicObj.stbName, pCreate->subStbName, TSDB_TABLE_FNAME_LEN);

×

UNCOV

499

    topicObj.stbUid = pStb->uid;

×

UNCOV

500

    mndReleaseStb(pMnode, pStb);

×

UNCOV

501

    if(pCreate->ast != NULL){

×

UNCOV

502

      qDebugL("topic:%s ast %s", topicObj.name, pCreate->ast);

×

UNCOV

503

      topicObj.ast = taosStrdup(pCreate->ast);

×

UNCOV

504

      MND_TMQ_NULL_CHECK(topicObj.ast);

×

UNCOV

505

      topicObj.astLen = strlen(pCreate->ast) + 1;

×

UNCOV

509

  SSdbRaw *pCommitRaw = mndTopicActionEncode(&topicObj);

×

UNCOV

510

  MND_TMQ_NULL_CHECK(pCommitRaw);

×

UNCOV

511

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

×

UNCOV

512

  if(code != 0) {

×

513

    sdbFreeRaw(pCommitRaw);

×

514

    goto END;

×

UNCOV

517

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));

×

UNCOV

518

  if (topicObj.ntbUid != 0) {

×

UNCOV

519

    MND_TMQ_RETURN_CHECK(sendCheckInfoToVnode(pTrans, pMnode, &topicObj));

×

UNCOV

521

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

×

UNCOV

522

  code = TSDB_CODE_ACTION_IN_PROGRESS;

×

UNCOV

524

END:

×

UNCOV

525

  taosMemoryFreeClear(topicObj.physicalPlan);

×

UNCOV

526

  taosMemoryFreeClear(topicObj.sql);

×

UNCOV

527

  taosMemoryFreeClear(topicObj.ast);

×

UNCOV

528

  taosArrayDestroy(topicObj.ntbColIds);

×

UNCOV

529

  if (topicObj.schema.nCols) {

×

UNCOV

530

    taosMemoryFreeClear(topicObj.schema.pSchema);

×

UNCOV

532

  nodesDestroyNode(pAst);

×

UNCOV

533

  nodesDestroyNode((SNode *)pPlan);

×

UNCOV

534

  mndTransDrop(pTrans);

×

UNCOV

535

  return code;

×

UNCOV

538

static int32_t mndProcessCreateTopicReq(SRpcMsg *pReq) {

×

UNCOV

539

  if (pReq == NULL || pReq->contLen <= 0) {

×

540

    return TSDB_CODE_INVALID_MSG;

×

UNCOV

542

  SMnode           *pMnode = pReq->info.node;

×

UNCOV

543

  int32_t           code = TDB_CODE_SUCCESS;

×

UNCOV

544

  SMqTopicObj      *pTopic = NULL;

×

UNCOV

545

  SDbObj           *pDb = NULL;

×

UNCOV

546

  SCMCreateTopicReq createTopicReq = {0};

×

UNCOV

548

  if (tDeserializeSCMCreateTopicReq(pReq->pCont, pReq->contLen, &createTopicReq) != 0) {

×

549

    code = TSDB_CODE_INVALID_MSG;

×

550

    goto END;

×

UNCOV

553

  mInfo("topic:%s start to create, sql:%s", createTopicReq.name, createTopicReq.sql);

×

UNCOV

555

  MND_TMQ_RETURN_CHECK(mndCheckCreateTopicReq(&createTopicReq));

×

UNCOV

557

  code = mndAcquireTopic(pMnode, createTopicReq.name, &pTopic);

×

UNCOV

558

  if (code == TDB_CODE_SUCCESS) {

×

UNCOV

559

    if (createTopicReq.igExists) {

×

UNCOV

560

      mInfo("topic:%s already exist, ignore exist is set", createTopicReq.name);

×

UNCOV

561

      goto END;

×

563

      code = TSDB_CODE_MND_TOPIC_ALREADY_EXIST;

×

564

      goto END;

×

UNCOV

566

  } else if (code != TSDB_CODE_MND_TOPIC_NOT_EXIST) {

×

567

    goto END;

×

UNCOV

570

  pDb = mndAcquireDb(pMnode, createTopicReq.subDbName);

×

UNCOV

571

  MND_TMQ_NULL_CHECK(pDb);

×

UNCOV

573

  if (pDb->cfg.walRetentionPeriod == 0) {

×

574

    code = TSDB_CODE_MND_DB_RETENTION_PERIOD_ZERO;

×

575

    mError("db:%s, not allowed to create topic when WAL_RETENTION_PERIOD is zero", pDb->name);

×

576

    goto END;

×

UNCOV

579

  if (sdbGetSize(pMnode->pSdb, SDB_TOPIC) >= tmqMaxTopicNum){

×

UNCOV

580

    code = TSDB_CODE_TMQ_TOPIC_OUT_OF_RANGE;

×

UNCOV

581

    mError("topic num out of range");

×

UNCOV

582

    goto END;

×

UNCOV

585

  MND_TMQ_RETURN_CHECK(grantCheck(TSDB_GRANT_SUBSCRIPTION));

×

UNCOV

586

  code = mndCreateTopic(pMnode, pReq, &createTopicReq, pDb, pReq->info.conn.user);

×

UNCOV

587

  if (code == 0) {

×

588

    code = TSDB_CODE_ACTION_IN_PROGRESS;

×

UNCOV

592

    SName dbname = {0};

×

UNCOV

593

    int32_t ret = tNameFromString(&dbname, createTopicReq.subDbName, T_NAME_ACCT | T_NAME_DB);

×

UNCOV

594

    if (ret != 0){

×

595

      mError("failed to parse db name:%s, ret:%d", createTopicReq.subDbName, ret);

×

UNCOV

597

    SName topicName = {0};

×

UNCOV

598

    ret = tNameFromString(&topicName, createTopicReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);

×

UNCOV

599

    if (ret != 0){

×

UNCOV

600

      mError("failed to parse topic name:%s, ret:%d", createTopicReq.name, ret);

×

UNCOV

602

    auditRecord(pReq, pMnode->clusterId, "createTopic", dbname.dbname, topicName.dbname,

×

UNCOV

603

                createTopicReq.sql, strlen(createTopicReq.sql));

×

UNCOV

606

END:

×

UNCOV

607

  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {

×

UNCOV

608

    mError("failed to create topic:%s since %s", createTopicReq.name, tstrerror(code));

×

UNCOV

611

  mndReleaseTopic(pMnode, pTopic);

×

UNCOV

612

  mndReleaseDb(pMnode, pDb);

×

UNCOV

614

  tFreeSCMCreateTopicReq(&createTopicReq);

×

UNCOV

615

  return code;

×

UNCOV

618

static int32_t mndDropTopic(SMnode *pMnode, STrans *pTrans, SRpcMsg *pReq, SMqTopicObj *pTopic) {

×

UNCOV

619

  if (pMnode == NULL || pTrans == NULL || pReq == NULL || pTopic == NULL) {

×

620

    return TSDB_CODE_INVALID_MSG;

×

UNCOV

622

  int32_t code = 0;

×

UNCOV

623

  SSdbRaw *pCommitRaw = NULL;

×

UNCOV

624

  MND_TMQ_RETURN_CHECK(mndUserRemoveTopic(pMnode, pTrans, pTopic->name));

×

UNCOV

625

  pCommitRaw = mndTopicActionEncode(pTopic);

×

UNCOV

626

  MND_TMQ_NULL_CHECK(pCommitRaw);

×

UNCOV

627

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

×

UNCOV

628

  if(code != 0) {

×

629

    sdbFreeRaw(pCommitRaw);

×

630

    goto END;

×

UNCOV

632

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));

×

UNCOV

633

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

×

UNCOV

635

END:

×

UNCOV

636

  return code;

×

UNCOV

639

bool checkTopic(SArray *topics, char *topicName){

×

UNCOV

640

  if (topics == NULL || topicName == NULL) {

×

641

    return false;

×

UNCOV

643

  int32_t sz = taosArrayGetSize(topics);

×

UNCOV

644

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

×

UNCOV

645

    char *name = taosArrayGetP(topics, i);

×

UNCOV

646

    if (name && strcmp(name, topicName) == 0) {

×

647

      return true;

×

UNCOV

650

  return false;

×

UNCOV

653

static int32_t mndCheckConsumerByTopic(SMnode *pMnode, STrans *pTrans, char *topicName){

×

UNCOV

654

  if (pMnode == NULL || pTrans == NULL || topicName == NULL) {

×

655

    return TSDB_CODE_INVALID_MSG;

×

UNCOV

657

  int32_t         code = 0;

×

UNCOV

658

  SSdb           *pSdb    = pMnode->pSdb;

×

UNCOV

659

  void           *pIter = NULL;

×

UNCOV

660

  SMqConsumerObj *pConsumer = NULL;

×

UNCOV

661

  while (1) {

×

UNCOV

662

    pIter = sdbFetch(pSdb, SDB_CONSUMER, pIter, (void **)&pConsumer);

×

UNCOV

663

    if (pIter == NULL) {

×

UNCOV

664

      break;

×

UNCOV

667

    bool found = checkTopic(pConsumer->assignedTopics, topicName);

×

UNCOV

668

    if (found){

×

669

      mError("topic:%s, failed to drop since subscribed by consumer:0x%" PRIx64 ", in consumer group %s",

×

671

      code = TSDB_CODE_MND_TOPIC_SUBSCRIBED;

×

672

      goto END;

×

UNCOV

675

    sdbRelease(pSdb, pConsumer);

×

UNCOV

678

END:

×

UNCOV

679

  sdbRelease(pSdb, pConsumer);

×

UNCOV

680

  sdbCancelFetch(pSdb, pIter);

×

UNCOV

681

  return code;

×

UNCOV

684

static int32_t mndDropCheckInfoByTopic(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic){

×

UNCOV

685

  if (pMnode == NULL || pTrans == NULL || pTopic == NULL) {

×

686

    return TSDB_CODE_INVALID_MSG;

×

UNCOV

689

  void   *pIter = NULL;

×

UNCOV

690

  SVgObj *pVgroup = NULL;

×

UNCOV

691

  int32_t code    = 0;

×

UNCOV

692

  SSdb   *pSdb    = pMnode->pSdb;

×

UNCOV

693

  void   *buf     = NULL;

×

UNCOV

694

  while (1) {

×

UNCOV

695

    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);

×

UNCOV

696

    if (pIter == NULL) break;

×

UNCOV

697

    if (!mndVgroupInDb(pVgroup, pTopic->dbUid)) {

×

UNCOV

698

      sdbRelease(pSdb, pVgroup);

×

UNCOV

699

      continue;

×

UNCOV

702

    buf = taosMemoryCalloc(1, sizeof(SMsgHead) + TSDB_TOPIC_FNAME_LEN);

×

UNCOV

703

    MND_TMQ_NULL_CHECK(buf);

×

UNCOV

704

    void *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));

×

UNCOV

705

    ((SMsgHead *)buf)->vgId = htonl(pVgroup->vgId);

×

UNCOV

706

    (void)memcpy(abuf, pTopic->name, TSDB_TOPIC_FNAME_LEN);

×

UNCOV

708

    STransAction action = {0};

×

UNCOV

709

    action.epSet = mndGetVgroupEpset(pMnode, pVgroup);

×

UNCOV

710

    action.pCont = buf;

×

UNCOV

711

    action.contLen = sizeof(SMsgHead) + TSDB_TOPIC_FNAME_LEN;

×

UNCOV

712

    action.msgType = TDMT_VND_TMQ_DEL_CHECKINFO;

×

UNCOV

713

    code = mndTransAppendRedoAction(pTrans, &action);

×

UNCOV

714

    if (code != 0) {

×

715

      taosMemoryFree(buf);

×

716

      goto END;

×

UNCOV

718

    sdbRelease(pSdb, pVgroup);

×

UNCOV

721

END:

×

UNCOV

722

  sdbRelease(pSdb, pVgroup);

×

UNCOV

723

  sdbCancelFetch(pSdb, pIter);

×

UNCOV

724

  return code;

×

UNCOV

727

static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {

×

UNCOV

728

  if (pReq == NULL) {

×

729

    return TSDB_CODE_INVALID_MSG;

×

UNCOV

731

  SMnode        *pMnode  = pReq->info.node;

×

UNCOV

732

  SMDropTopicReq dropReq = {0};

×

UNCOV

733

  int32_t        code = 0;

×

UNCOV

734

  SMqTopicObj *pTopic = NULL;

×

UNCOV

735

  STrans      *pTrans = NULL;

×

UNCOV

737

  if (tDeserializeSMDropTopicReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {

×

738

    return TSDB_CODE_INVALID_MSG;

×

UNCOV

741

  code = mndAcquireTopic(pMnode, dropReq.name, &pTopic);

×

UNCOV

742

  if (code != 0) {

×

UNCOV

743

    if (dropReq.igNotExists) {

×

UNCOV

744

      mInfo("topic:%s, not exist, ignore not exist is set", dropReq.name);

×

UNCOV

745

      tFreeSMDropTopicReq(&dropReq);

×

UNCOV

746

      return 0;

×

748

      mError("topic:%s, failed to drop since %s", dropReq.name, tstrerror(code));

×

749

      tFreeSMDropTopicReq(&dropReq);

×

750

      return code;

×

UNCOV

754

  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_DB, pReq, "drop-topic");

×

UNCOV

755

  MND_TMQ_NULL_CHECK(pTrans);

×

UNCOV

757

  mndTransSetDbName(pTrans, pTopic->db, NULL);

×

UNCOV

758

  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

×

UNCOV

759

  mInfo("trans:%d, used to drop topic:%s", pTrans->id, pTopic->name);

×

UNCOV

761

  MND_TMQ_RETURN_CHECK(mndCheckTopicPrivilege(pMnode, pReq->info.conn.user, MND_OPER_DROP_TOPIC, pTopic));

×

UNCOV

762

  MND_TMQ_RETURN_CHECK(mndCheckDbPrivilegeByName(pMnode, pReq->info.conn.user, MND_OPER_READ_DB, pTopic->db));

×

UNCOV

763

  MND_TMQ_RETURN_CHECK(mndCheckConsumerByTopic(pMnode, pTrans, dropReq.name));

×

UNCOV

764

  MND_TMQ_RETURN_CHECK(mndDropSubByTopic(pMnode, pTrans, dropReq.name));

×

UNCOV

766

  if (pTopic->ntbUid != 0) {

×

UNCOV

767

    MND_TMQ_RETURN_CHECK(mndDropCheckInfoByTopic(pMnode, pTrans, pTopic));

×

UNCOV

770

  code = mndDropTopic(pMnode, pTrans, pReq, pTopic);

×

UNCOV

772

END:

×

UNCOV

773

  mndReleaseTopic(pMnode, pTopic);

×

UNCOV

774

  mndTransDrop(pTrans);

×

UNCOV

775

  if (code != 0) {

×

UNCOV

776

    mError("topic:%s, failed to drop since %s", dropReq.name, tstrerror(code));

×

UNCOV

777

    tFreeSMDropTopicReq(&dropReq);

×

UNCOV

778

    return code;

×

UNCOV

781

  SName name = {0};

×

UNCOV

782

  int32_t ret = tNameFromString(&name, dropReq.name, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);

×

UNCOV

783

  if (ret != 0) {

×

784

    mError("topic:%s, failed to drop since %s", dropReq.name, tstrerror(ret));

×

UNCOV

786

  auditRecord(pReq, pMnode->clusterId, "dropTopic", name.dbname, name.tname, dropReq.sql, dropReq.sqlLen);

×

UNCOV

788

  tFreeSMDropTopicReq(&dropReq);

×

UNCOV

790

  return TSDB_CODE_ACTION_IN_PROGRESS;

×

795

    return TSDB_CODE_INVALID_MSG;

×

802

    return TSDB_CODE_MND_DB_NOT_SELECTED;

×

807

  while (1) {

×

814

    if (pTopic->dbUid == pDb->uid) {

×

815

      numOfTopics++;

×

818

    sdbRelease(pSdb, pTopic);

×

UNCOV

826

static void schemaToJson(SSchema *schema, int32_t nCols, char *schemaJson){

×

UNCOV

827

  if (schema == NULL || schemaJson == NULL) {

×

828

    return;

×

UNCOV

830

  char*   string = NULL;

×

UNCOV

831

  int32_t code = 0;

×

UNCOV

832

  cJSON* columns = cJSON_CreateArray();

×

UNCOV

833

  MND_TMQ_NULL_CHECK(columns);

×

UNCOV

834

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

×

UNCOV

835

    cJSON*   column = cJSON_CreateObject();

×

UNCOV

836

    MND_TMQ_NULL_CHECK(column);

×

UNCOV

837

    SSchema* s = schema + i;

×

UNCOV

838

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

×

UNCOV

839

    MND_TMQ_NULL_CHECK(cname);

×

UNCOV

840

    if (!cJSON_AddItemToObject(column, "name", cname)) {

×

841

      return;

×

UNCOV

843

    cJSON* ctype = cJSON_CreateString(tDataTypes[s->type].name);

×

UNCOV

844

    MND_TMQ_NULL_CHECK(ctype);

×

UNCOV

845

    if (!cJSON_AddItemToObject(column, "type", ctype)) {

×

846

      return;

×

UNCOV

848

    int32_t length = 0;

×

UNCOV

849

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

×

UNCOV

850

        s->type == TSDB_DATA_TYPE_GEOMETRY) {

×

UNCOV

851

      length = s->bytes - VARSTR_HEADER_SIZE;

×

UNCOV

852

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

×

UNCOV

853

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

×

UNCOV

855

      length = s->bytes;

×

UNCOV

857

    cJSON*  cbytes = cJSON_CreateNumber(length);

×

UNCOV

858

    MND_TMQ_NULL_CHECK(cbytes);

×

UNCOV

859

    if (!cJSON_AddItemToObject(column, "length", cbytes)){

×

860

      return;

×

UNCOV

862

    if (!cJSON_AddItemToArray(columns, column)){

×

863

      return;

×

UNCOV

866

  string = cJSON_PrintUnformatted(columns);

×

UNCOV

867

  cJSON_Delete(columns);

×

UNCOV

869

  size_t len = strlen(string);

×

UNCOV

870

  if(string && len <= TSDB_SHOW_SCHEMA_JSON_LEN){

×

UNCOV

871

    STR_TO_VARSTR(schemaJson, string);

×

873

    mError("mndRetrieveTopic build schema error json:%p, json len:%zu", string, len);

×

UNCOV

875

  taosMemoryFree(string);

×

UNCOV

877

END:

×

UNCOV

878

  return;

×

UNCOV

881

static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) {

×

UNCOV

882

  if (pReq == NULL || pShow == NULL || pBlock == NULL) {

×

883

    return TSDB_CODE_INVALID_MSG;

×

UNCOV

885

  SMnode      *pMnode = pReq->info.node;

×

UNCOV

886

  SSdb        *pSdb = pMnode->pSdb;

×

UNCOV

887

  int32_t      numOfRows = 0;

×

UNCOV

888

  SMqTopicObj *pTopic = NULL;

×

UNCOV

889

  int32_t      code = 0;

×

UNCOV

890

  char        *sql  = NULL;

×

UNCOV

891

  char        *schemaJson  = NULL;

×

UNCOV

893

  while (numOfRows < rowsCapacity) {

×

UNCOV

894

    pShow->pIter = sdbFetch(pSdb, SDB_TOPIC, pShow->pIter, (void **)&pTopic);

×

UNCOV

895

    if (pShow->pIter == NULL) break;

×

UNCOV

897

    SColumnInfoData *pColInfo= NULL;

×

UNCOV

898

    SName            n = {0};

×

UNCOV

899

    int32_t          cols = 0;

×

UNCOV

901

    char        topicName[TSDB_TOPIC_NAME_LEN + VARSTR_HEADER_SIZE + 5] = {0};

×

UNCOV

902

    const char *pName = mndGetDbStr(pTopic->name);

×

UNCOV

903

    STR_TO_VARSTR(topicName, pName);

×

UNCOV

905

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

906

    MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

907

    MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)topicName, false));

×

UNCOV

909

    char dbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};

×

UNCOV

910

    MND_TMQ_RETURN_CHECK(tNameFromString(&n, pTopic->db, T_NAME_ACCT | T_NAME_DB));

×

UNCOV

911

    MND_TMQ_RETURN_CHECK(tNameGetDbName(&n, varDataVal(dbName)));

×

UNCOV

912

    varDataSetLen(dbName, strlen(varDataVal(dbName)));

×

UNCOV

914

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

915

    MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

916

    MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)dbName, false));

×

UNCOV

918

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

919

    MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

920

    MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)&pTopic->createTime, false));

×

UNCOV

922

    sql = taosMemoryMalloc(strlen(pTopic->sql) + VARSTR_HEADER_SIZE);

×

UNCOV

923

    MND_TMQ_NULL_CHECK(sql);

×

UNCOV

924

    STR_TO_VARSTR(sql, pTopic->sql);

×

UNCOV

926

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

927

    MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

928

    MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)sql, false));

×

UNCOV

930

    taosMemoryFreeClear(sql);

×

UNCOV

932

    schemaJson = taosMemoryMalloc(TSDB_SHOW_SCHEMA_JSON_LEN + VARSTR_HEADER_SIZE);

×

UNCOV

933

    MND_TMQ_NULL_CHECK(schemaJson);

×

UNCOV

934

    if(pTopic->subType == TOPIC_SUB_TYPE__COLUMN){

×

UNCOV

935

      schemaToJson(pTopic->schema.pSchema, pTopic->schema.nCols, schemaJson);

×

UNCOV

936

    }else if(pTopic->subType == TOPIC_SUB_TYPE__TABLE){

×

UNCOV

937

      SStbObj *pStb = mndAcquireStb(pMnode, pTopic->stbName);

×

UNCOV

938

      if (pStb == NULL) {

×

939

        STR_TO_VARSTR(schemaJson, "NULL");

×

940

        mError("mndRetrieveTopic mndAcquireStb null stbName:%s", pTopic->stbName);

×

UNCOV

942

        schemaToJson(pStb->pColumns, pStb->numOfColumns, schemaJson);

×

UNCOV

943

        mndReleaseStb(pMnode, pStb);

×

UNCOV

946

      STR_TO_VARSTR(schemaJson, "NULL");

×

UNCOV

949

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

950

    MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

951

    MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)schemaJson, false));

×

UNCOV

952

    taosMemoryFreeClear(schemaJson);

×

UNCOV

954

    char mete[4 + VARSTR_HEADER_SIZE] = {0};

×

UNCOV

955

    if(pTopic->withMeta){

×

UNCOV

956

      STR_TO_VARSTR(mete, "yes");

×

UNCOV

958

      STR_TO_VARSTR(mete, "no");

×

UNCOV

961

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

962

    MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

963

    MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)mete, false));

×

UNCOV

965

    char type[8 + VARSTR_HEADER_SIZE] = {0};

×

UNCOV

966

    if(pTopic->subType == TOPIC_SUB_TYPE__COLUMN){

×

UNCOV

967

      STR_TO_VARSTR(type, "column");

×

UNCOV

968

    }else if(pTopic->subType == TOPIC_SUB_TYPE__TABLE){

×

UNCOV

969

      STR_TO_VARSTR(type, "stable");

×

UNCOV

971

      STR_TO_VARSTR(type, "db");

×

UNCOV

974

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

×

UNCOV

975

    MND_TMQ_NULL_CHECK(pColInfo);

×

UNCOV

976

    MND_TMQ_RETURN_CHECK(colDataSetVal(pColInfo, numOfRows, (const char *)type, false));

×

UNCOV

978

    numOfRows++;

×

UNCOV

979

    sdbRelease(pSdb, pTopic);

×

UNCOV

982

  pShow->numOfRows += numOfRows;

×

UNCOV

983

  return numOfRows;

×

985

END:

×

986

  taosMemoryFreeClear(schemaJson);

×

987

  taosMemoryFreeClear(sql);

×

988

  return code;

×

991

static void mndCancelGetNextTopic(SMnode *pMnode, void *pIter) {

×

992

  if (pMnode == NULL) return;

×

993

  SSdb *pSdb = pMnode->pSdb;

×

994

  sdbCancelFetchByType(pSdb, pIter, SDB_TOPIC);

×

999

    return false;

×

UNCOV

1011

    if (pTopic->dbUid == pDb->uid) {

×

UNCOV

1012

      sdbRelease(pSdb, pTopic);

×

UNCOV

1013

      sdbCancelFetch(pSdb, pIter);

×

UNCOV

1014

      return true;

×

UNCOV

1017

    sdbRelease(pSdb, pTopic);

×

taosdata / TDengine / #3562

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

taosdata / TDengine / #3562

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

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