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

53

int32_t mndInitTopic(SMnode *pMnode) {

544,420✔

54

  SSdbTable table = {

544,420✔

64

  if (pMnode == NULL) {

544,420✔

67

  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CREATE_TOPIC, mndProcessCreateTopicReq);

544,420✔

68

  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_TOPIC, mndProcessDropTopicReq);

544,420✔

70

  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndRetrieveTopic);

544,420✔

71

  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndCancelGetNextTopic);

544,420✔

73

  return sdbSetTable(pMnode->pSdb, table);

544,420✔

76

void mndCleanupTopic(SMnode *pMnode) {}

544,351✔

78

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

181,263✔

79

  if (fullTopic == NULL) {

181,263✔

82

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

181,263✔

83

  if (tmp == NULL) {

181,263✔

86

    tstrncpy(topic, tmp + 1, TSDB_TOPIC_FNAME_LEN);

181,263✔

89

SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {

378,745✔

90

  if (pTopic == NULL) {

378,745✔

93

  int32_t code = 0;

378,745✔

94

  int32_t lino = 0;

378,745✔

95

  terrno = TSDB_CODE_OUT_OF_MEMORY;

378,745✔

97

  void *  swBuf = NULL;

378,745✔

98

  int32_t physicalPlanLen = 0;

378,745✔

99

  if (pTopic->physicalPlan) {

378,745✔

100

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

378,745✔

103

  int32_t schemaLen = 0;

378,745✔

104

  if (pTopic->schema.nCols) {

378,745✔

105

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

585,578✔

108

  int32_t  size = sizeof(SMqTopicObj) + physicalPlanLen + pTopic->sqlLen + schemaLen + MND_TOPIC_RESERVE_SIZE;

378,745✔

109

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

378,745✔

110

  if (pRaw == NULL) {

378,745✔

114

  int32_t dataPos = 0;

378,745✔

115

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

378,745✔

116

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

378,745✔

117

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

378,745✔

118

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

378,745✔

119

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

378,745✔

120

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

378,745✔

121

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

378,745✔

122

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

378,745✔

123

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

378,745✔

124

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

378,745✔

126

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

378,745✔

127

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

378,745✔

128

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

378,745✔

129

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

378,745✔

130

  SDB_SET_INT32(pRaw, dataPos, physicalPlanLen, TOPIC_ENCODE_OVER);

378,745✔

131

  if (physicalPlanLen) {

378,745✔

132

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

378,745✔

134

  SDB_SET_INT32(pRaw, dataPos, schemaLen, TOPIC_ENCODE_OVER);

378,745✔

135

  if (schemaLen) {

378,745✔

136

    swBuf = taosMemoryMalloc(schemaLen);

292,789✔

137

    if (swBuf == NULL) {

292,789✔

140

    void *aswBuf = swBuf;

292,789✔

141

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

585,578✔

144

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

292,789✔

146

  SDB_SET_INT64(pRaw, dataPos, pTopic->ownerId, TOPIC_ENCODE_OVER); // since ver 4

378,745✔

147

  SDB_SET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_ENCODE_OVER);

378,745✔

148

  SDB_SET_DATALEN(pRaw, dataPos, TOPIC_ENCODE_OVER);

378,745✔

150

  terrno = TSDB_CODE_SUCCESS;

378,745✔

152

TOPIC_ENCODE_OVER:

378,745✔

153

  if (swBuf) taosMemoryFree(swBuf);

378,745✔

154

  if (terrno != TSDB_CODE_SUCCESS) {

378,745✔

160

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

378,745✔

161

  return pRaw;

378,745✔

164

SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {

321,191✔

165

  if (pRaw == NULL) return NULL;

321,191✔

166

  int32_t code = 0;

321,191✔

167

  int32_t lino = 0;

321,191✔

168

  terrno = TSDB_CODE_OUT_OF_MEMORY;

321,191✔

169

  SSdbRow *    pRow = NULL;

321,191✔

170

  SMqTopicObj *pTopic = NULL;

321,191✔

171

  void *       buf = NULL;

321,191✔

172

  char*        ast = NULL;

321,191✔

174

  int8_t sver = 0;

321,191✔

175

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

321,191✔

177

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

321,191✔

182

  pRow = sdbAllocRow(sizeof(SMqTopicObj));

321,191✔

183

  if (pRow == NULL) goto TOPIC_DECODE_OVER;

321,191✔

185

  pTopic = sdbGetRowObj(pRow);

321,191✔

186

  if (pTopic == NULL) goto TOPIC_DECODE_OVER;

321,191✔

188

  int32_t len = 0;

321,191✔

189

  int32_t dataPos = 0;

321,191✔

190

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

321,191✔

191

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

321,191✔

192

  if (sver >= 2) {

321,191✔

193

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

321,191✔

195

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

321,191✔

196

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

321,191✔

197

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

321,191✔

198

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

321,191✔

199

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

321,191✔

200

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

321,191✔

201

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

321,191✔

203

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

321,191✔

204

  if (sver >= 3) {

321,191✔

205

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

321,191✔

207

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

321,191✔

208

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

321,191✔

209

  if (pTopic->sql == NULL) {

321,191✔

213

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

321,191✔

215

  if (sver < MND_TOPIC_VER_SUPPORT_OWNER) {

321,191✔

231

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

321,191✔

232

    if (len) {

321,191✔

233

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

321,191✔

234

      if (pTopic->physicalPlan == NULL) {

321,191✔

238

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

321,191✔

243

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

321,191✔

244

    if (len) {

321,191✔

245

      buf = taosMemoryMalloc(len);

250,534✔

246

      if (buf == NULL) {

250,534✔

250

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

250,534✔

251

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

501,068✔

255

      pTopic->schema.nCols = 0;

70,657✔

256

      pTopic->schema.version = 0;

70,657✔

257

      pTopic->schema.pSchema = NULL;

70,657✔

259

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

321,191✔

262

  SDB_GET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_DECODE_OVER);

321,191✔

263

  terrno = TSDB_CODE_SUCCESS;

321,191✔

265

TOPIC_DECODE_OVER:

321,191✔

266

  taosMemoryFreeClear(buf);

321,191✔

267

  taosMemoryFreeClear(ast);

321,191✔

269

  if (terrno != TSDB_CODE_SUCCESS) {

321,191✔

275

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

321,191✔

276

  return pRow;

321,191✔

279

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

191,295✔

280

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

191,295✔

281

  return 0;

191,295✔

284

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

321,191✔

285

  if (pTopic == NULL) return 0;

321,191✔

286

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

321,191✔

287

  taosMemoryFreeClear(pTopic->sql);

321,191✔

288

  taosMemoryFreeClear(pTopic->physicalPlan);

321,191✔

289

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

321,191✔

290

  return 0;

321,191✔

293

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

2,583✔

294

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

2,583✔

295

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

2,583✔

296

  taosWLockLatch(&pOldTopic->lock);

2,583✔

297

  SMqTopicObj tmpTopic = *pOldTopic;

2,583✔

298

  (void)memcpy(pOldTopic, pNewTopic, offsetof(SMqTopicObj, lock));

2,583✔

299

  *pNewTopic = tmpTopic;

2,583✔

300

  taosWUnLockLatch(&pOldTopic->lock);

2,583✔

301

  return 0;

2,583✔

304

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

3,058,162✔

305

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

3,058,162✔

308

  SSdb *pSdb = pMnode->pSdb;

3,058,162✔

309

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

3,058,162✔

310

  if (*pTopic == NULL) {

3,058,162✔

311

    return TSDB_CODE_MND_TOPIC_NOT_EXIST;

218,352✔

313

  return TSDB_CODE_SUCCESS;

2,839,810✔

316

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

3,442,991✔

317

  if (pMnode == NULL) return;

3,442,991✔

318

  SSdb *pSdb = pMnode->pSdb;

3,442,991✔

319

  sdbRelease(pSdb, pTopic);

3,442,991✔

322

static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {

171,062✔

323

  if (pCreate == NULL) return TSDB_CODE_INVALID_PARA;

171,062✔

324

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

171,062✔

326

  if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) {

171,062✔

327

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

124,836✔

328

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

46,226✔

329

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

15,612✔

330

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

30,614✔

331

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

30,614✔

334

  return 0;

171,062✔

337

static int32_t processAst(SMqTopicObj *topicObj, const char *ast) {

167,009✔

338

  SNode *     pAst = NULL;

167,009✔

339

  SQueryPlan *pPlan = NULL;

167,009✔

340

  int32_t     code = TSDB_CODE_SUCCESS;

167,009✔

341

  int32_t     lino = 0;

167,009✔

343

  PRINT_LOG_START

167,009✔

344

  if (ast == NULL) {

167,009✔

345

    topicObj->physicalPlan = taosStrdup("");

37,246✔

346

    goto END;

37,246✔

348

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

129,763✔

349

  MND_TMQ_RETURN_CHECK(nodesStringToNode(ast, &pAst));

129,763✔

350

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

129,763✔

352

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

129,763✔

353

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

129,763✔

354

  if (pPlan == NULL) {

129,763✔

358

  int32_t levelNum = LIST_LENGTH(pPlan->pSubplans);

129,763✔

359

  if (levelNum != 1) {

129,763✔

364

  SNodeListNode *pNodeListNode = (SNodeListNode *)nodesListGetNode(pPlan->pSubplans, 0);

129,763✔

365

  MND_TMQ_NULL_CHECK(pNodeListNode);

129,763✔

366

  int32_t opNum = LIST_LENGTH(pNodeListNode->pNodeList);

129,763✔

367

  if (opNum != 1) {

129,763✔

372

  code = nodesNodeToString(nodesListGetNode(pNodeListNode->pNodeList, 0), false, &topicObj->physicalPlan, NULL);

129,763✔

374

END:

167,009✔

375

  nodesDestroyNode(pAst);

167,009✔

376

  qDestroyQueryPlan(pPlan);

167,009✔

377

  PRINT_LOG_END

167,009✔

378

  return code;

167,009✔

381

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

164,426✔

383

  if (pMnode == NULL || pReq == NULL || pCreate == NULL || pDb == NULL || pOperUser == NULL)

164,426✔

385

  STrans *    pTrans = NULL;

164,426✔

386

  int32_t     code = 0;

164,426✔

387

  int32_t     lino = 0;

164,426✔

388

  SMqTopicObj topicObj = {0};

164,426✔

390

  PRINT_LOG_START

164,426✔

391

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

164,426✔

392

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

164,426✔

393

  MND_TMQ_NULL_CHECK(pTrans);

164,426✔

394

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

164,426✔

395

  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

164,426✔

397

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

164,426✔

398

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

164,426✔

399

  tstrncpy(topicObj.createUser, pOperUser->name, TSDB_USER_LEN);

164,426✔

400

  topicObj.ownerId = pOperUser->uid;

164,426✔

403

  if (pDb) {

164,426✔

405

    MND_TMQ_RETURN_CHECK(

164,426✔

409

  topicObj.createTime = taosGetTimestampMs();

164,426✔

410

  topicObj.updateTime = topicObj.createTime;

164,426✔

411

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

164,426✔

412

  topicObj.dbUid = pDb->uid;

164,426✔

413

  topicObj.version = 1;

164,426✔

414

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

164,426✔

415

  MND_TMQ_NULL_CHECK(topicObj.sql);

164,426✔

416

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

164,426✔

417

  topicObj.subType = pCreate->subType;

164,426✔

418

  topicObj.withMeta = pCreate->withMeta;

164,426✔

419

  taosInitRWLatch(&topicObj.lock);

164,426✔

421

  MND_TMQ_RETURN_CHECK(processAst(&topicObj, pCreate->ast));

164,426✔

423

  if (pCreate->subStbName[0] != 0) {

164,426✔

424

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

123,151✔

425

    SStbObj *pStb = mndAcquireStb(pMnode, topicObj.stbName);

123,151✔

426

    MND_TMQ_NULL_CHECK(pStb);

123,151✔

427

    char stbName[TSDB_TABLE_NAME_LEN] = {0};

122,375✔

428

    mndExtractTbNameFromStbFullName(pStb->name, stbName, TSDB_TABLE_NAME_LEN);

122,375✔

429

    MND_TMQ_RETURN_CHECK(

122,375✔

431

    MND_TMQ_RETURN_CHECK(

122,375✔

433

    topicObj.stbUid = pStb->uid;

122,375✔

434

    mndReleaseStb(pMnode, pStb);

122,375✔

437

  if (pCreate->subType == TOPIC_SUB_TYPE__DB) {

163,650✔

438

    MND_TMQ_RETURN_CHECK(

29,414✔

440

    MND_TMQ_RETURN_CHECK(mndCheckObjPrivilegeRecF(pMnode, pOperUser, PRIV_TBL_SELECT, PRIV_OBJ_TBL, 0, pDb->name, "*"));

29,414✔

441

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

134,236✔

445

  SSdbRaw *pCommitRaw = mndTopicActionEncode(&topicObj);

163,650✔

446

  MND_TMQ_NULL_CHECK(pCommitRaw);

163,650✔

447

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

163,650✔

448

  if (code != 0) {

163,650✔

453

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));

163,650✔

454

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

163,650✔

456

END:

164,426✔

457

  PRINT_LOG_END

164,426✔

458

  taosMemoryFreeClear(topicObj.sql);

164,426✔

459

  taosMemoryFreeClear(topicObj.physicalPlan);

164,426✔

460

  if (topicObj.schema.nCols) {

164,426✔

461

    taosMemoryFreeClear(topicObj.schema.pSchema);

127,180✔

463

  mndTransDrop(pTrans);

164,426✔

464

  return code;

164,426✔

467

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

2,583✔

469

  if (pMnode == NULL || pReq == NULL || pCreate == NULL || pDb == NULL || userName == NULL)

2,583✔

471

  STrans *    pTrans = NULL;

2,583✔

472

  int32_t     code = 0;

2,583✔

473

  int32_t     lino = 0;

2,583✔

474

  SMqTopicObj topicObj = {0};

2,583✔

476

  PRINT_LOG_START

2,583✔

477

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

2,583✔

478

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

2,583✔

479

  MND_TMQ_NULL_CHECK(pTrans);

2,583✔

480

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

2,583✔

481

  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

2,583✔

483

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

2,583✔

484

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

2,583✔

485

  tstrncpy(topicObj.createUser, userName, TSDB_USER_LEN);

2,583✔

487

  MND_TMQ_RETURN_CHECK(mndCheckTopicPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_CREATE_TOPIC, &topicObj));

2,583✔

489

  taosRLockLatch(&topicObjOri->lock);

2,583✔

490

  topicObj.createTime = topicObjOri->createTime;

2,583✔

491

  topicObj.updateTime = taosGetTimestampMs();

2,583✔

492

  topicObj.uid = topicObjOri->uid;

2,583✔

493

  topicObj.dbUid = pDb->uid;

2,583✔

494

  topicObj.version = topicObjOri->version + 1;

2,583✔

495

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

2,583✔

496

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

2,583✔

497

  topicObj.subType = pCreate->subType;

2,583✔

498

  topicObj.withMeta = pCreate->withMeta;

2,583✔

499

  taosInitRWLatch(&topicObj.lock);

2,583✔

500

  taosRUnLockLatch(&topicObjOri->lock);

2,583✔

502

  MND_TMQ_RETURN_CHECK(processAst(&topicObj, pCreate->ast));

2,583✔

504

  if (pCreate->subStbName[0] != 0) {

2,583✔

505

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

1,476✔

506

    SStbObj *pStb = mndAcquireStb(pMnode, topicObj.stbName);

1,476✔

507

    MND_TMQ_NULL_CHECK(pStb);

1,476✔

508

    topicObj.stbUid = pStb->uid;

1,476✔

509

    mndReleaseStb(pMnode, pStb);

1,476✔

512

  SSdbRaw *pCommitRaw = mndTopicActionEncode(&topicObj);

2,583✔

513

  MND_TMQ_NULL_CHECK(pCommitRaw);

2,583✔

514

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

2,583✔

515

  if (code != 0) {

2,583✔

520

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));

2,583✔

521

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

2,583✔

523

END:

2,583✔

524

  PRINT_LOG_END

2,583✔

525

  taosMemoryFreeClear(topicObj.sql);

2,583✔

526

  taosMemoryFreeClear(topicObj.physicalPlan);

2,583✔

527

  if (topicObj.schema.nCols) {

2,583✔

528

    taosMemoryFreeClear(topicObj.schema.pSchema);

2,583✔

530

  mndTransDrop(pTrans);

2,583✔

531

  return code;

2,583✔

534

static int32_t creatTopic(SRpcMsg *pReq, SCMCreateTopicReq *createTopicReq, SUserObj *pOperUser) {

167,741✔

535

  SMqTopicObj *pTopic = NULL;

167,741✔

536

  SDbObj *     pDb = NULL;

167,741✔

537

  int32_t      code = TSDB_CODE_SUCCESS;

167,741✔

538

  int32_t      lino = 0;

167,741✔

539

  SMnode *     pMnode = pReq->info.node;

167,741✔

540

  int64_t      tss = taosGetTimestampMs();

167,741✔

542

  PRINT_LOG_START

167,741✔

543

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

167,741✔

544

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

167,741✔

545

  if (code == TSDB_CODE_SUCCESS) {

167,741✔

546

    mndReleaseTopic(pMnode, pTopic);

843✔

547

    if (createTopicReq->igExists) {

843✔

548

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

843✔

552

    goto END;

843✔

553

  } else if (code != TSDB_CODE_MND_TOPIC_NOT_EXIST) {

166,898✔

557

  pDb = mndAcquireDb(pMnode, createTopicReq->subDbName);

166,898✔

558

  MND_TMQ_NULL_CHECK(pDb);

166,898✔

560

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

166,122✔

565

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

166,122✔

566

    code = TSDB_CODE_TMQ_TOPIC_OUT_OF_RANGE;

1,696✔

567

    goto END;

1,696✔

570

  MND_TMQ_RETURN_CHECK(grantCheck(TSDB_GRANT_SUBSCRIPTION));

164,426✔

571

  MND_TMQ_RETURN_CHECK(mndCreateTopic(pMnode, pReq, createTopicReq, pDb, pOperUser));

164,426✔

572

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {

163,650✔

573

    int64_t tse = taosGetTimestampMs();

163,650✔

574

    double  duration = (double)(tse - tss);

163,650✔

575

    duration = duration / 1000;

163,650✔

576

    auditRecord(pReq, pMnode->clusterId, "createTopic", createTopicReq->subDbName, createTopicReq->name,

163,650✔

577

                createTopicReq->sql, strlen(createTopicReq->sql), duration, 0);

163,650✔

579

  code = TSDB_CODE_ACTION_IN_PROGRESS;

163,650✔

581

END:

167,741✔

582

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

167,741✔

583

    mError("%s failed, topic:%s since %s", __func__, createTopicReq->name, tstrerror(code));

3,248✔

585

    mInfo("topic:%s create successfully", createTopicReq->name);

164,493✔

587

  mndReleaseDb(pMnode, pDb);

167,741✔

588

  return code;

167,741✔

591

static int32_t reloadTopic(SRpcMsg *pReq, SCMCreateTopicReq *createTopicReq) {

3,321✔

592

  SMnode *     pMnode = pReq->info.node;

3,321✔

593

  int32_t      code = TSDB_CODE_SUCCESS;

3,321✔

594

  int32_t      lino = 0;

3,321✔

595

  SDbObj *     pDb = NULL;

3,321✔

596

  SMqTopicObj *pTopic = NULL;

3,321✔

597

  int64_t      tss = taosGetTimestampMs();

3,321✔

599

  PRINT_LOG_START

3,321✔

600

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

3,321✔

601

  if (code != 0) {

3,321✔

602

    if (createTopicReq->igExists) {

738✔

603

      mInfo("topic:%s, not exist, ignore not exist is set", createTopicReq->name);

369✔

604

      code = 0;

369✔

605

      goto END;

369✔

607

      mError("topic:%s, failed to reload since %s", createTopicReq->name, tstrerror(code));

369✔

608

      goto END;

369✔

612

  pDb = mndAcquireDb(pMnode, createTopicReq->subDbName);

2,583✔

613

  MND_TMQ_NULL_CHECK(pDb);

2,583✔

615

  MND_TMQ_RETURN_CHECK(grantCheck(TSDB_GRANT_SUBSCRIPTION));

2,583✔

616

  MND_TMQ_RETURN_CHECK(mndReloadTopic(pMnode, pReq, createTopicReq, pDb, RPC_MSG_USER(pReq), pTopic));

2,583✔

618

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {

2,583✔

619

    int64_t tse = taosGetTimestampMs();

2,583✔

620

    double  duration = (double)(tse - tss);

2,583✔

621

    duration = duration / 1000;

2,583✔

622

    auditRecord(pReq, pMnode->clusterId, "reloadTopic", createTopicReq->subDbName, createTopicReq->name,

2,583✔

623

                createTopicReq->sql, strlen(createTopicReq->sql), duration, 0);

2,583✔

626

  code = TSDB_CODE_ACTION_IN_PROGRESS;

2,583✔

628

  if (topicsToReload == NULL) {

2,583✔

629

    topicsToReload = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);

369✔

630

    MND_TMQ_NULL_CHECK(topicsToReload);

369✔

632

  MND_TMQ_RETURN_CHECK(

2,583✔

634

  mInfo("topic:%s, marked to reload", createTopicReq->name);

2,583✔

636

END:

3,321✔

637

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

3,321✔

638

    mError("%s failed, topic:%s since %s", __func__, createTopicReq->name, tstrerror(code));

369✔

640

    mInfo("topic:%s create successfully", createTopicReq->name);

2,952✔

642

  mndReleaseTopic(pMnode, pTopic);

3,321✔

643

  mndReleaseDb(pMnode, pDb);

3,321✔

645

  return code;

3,321✔

648

static int32_t mndProcessCreateTopicReq(SRpcMsg *pReq) {

171,062✔

649

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

171,062✔

652

  SMnode *pMnode = pReq->info.node;

171,062✔

653

  SUserObj *pOperUser = NULL;

171,062✔

654

  int32_t code = TSDB_CODE_SUCCESS;

171,062✔

655

  int32_t lino = 0;

171,062✔

657

  SCMCreateTopicReq createTopicReq = {0};

171,062✔

659

  PRINT_LOG_START

171,062✔

660

  MND_TMQ_RETURN_CHECK(tDeserializeSCMCreateTopicReq(pReq->pCont, pReq->contLen, &createTopicReq));

171,062✔

662

  if ((code = mndAcquireUser(pMnode, RPC_MSG_USER(pReq), &pOperUser)) != 0) goto END;

171,062✔

664

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

171,062✔

666

  MND_TMQ_RETURN_CHECK(mndCheckCreateTopicReq(&createTopicReq));

171,062✔

668

  if (createTopicReq.reload) {

171,062✔

669

    MND_TMQ_RETURN_CHECK(reloadTopic(pReq, &createTopicReq));

3,321✔

671

    MND_TMQ_RETURN_CHECK(creatTopic(pReq, &createTopicReq, pOperUser));

167,741✔

674

END:

170,948✔

675

  tFreeSCMCreateTopicReq(&createTopicReq);

171,062✔

676

  mndReleaseUser(pMnode, pOperUser);

171,062✔

677

  return code;

171,062✔

680

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

126,965✔

681

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

126,965✔

684

  int32_t  code = 0;

126,965✔

685

  int32_t  lino = 0;

126,965✔

686

  SSdbRaw *pCommitRaw = NULL;

126,965✔

687

  PRINT_LOG_START

126,965✔

688

  char topicFName[TSDB_TOPIC_FNAME_LEN + 1] = {0};                       // 1.topic

126,965✔

689

  mndTopicGetShowName(pTopic->name, topicFName);

126,965✔

690

  char topicDbFName[TSDB_DB_NAME_LEN + TSDB_TOPIC_FNAME_LEN + 1] = {0};  // 1.db.topic

126,965✔

691

  (void)snprintf(topicDbFName, sizeof(topicDbFName), "%s.%s", pTopic->db, topicFName);

126,965✔

692

  MND_TMQ_RETURN_CHECK(mndUserRemoveTopic(pMnode, pTrans, topicDbFName));

126,965✔

693

  pCommitRaw = mndTopicActionEncode(pTopic);

126,965✔

694

  MND_TMQ_NULL_CHECK(pCommitRaw);

126,965✔

695

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

126,965✔

696

  if (code != 0) {

126,965✔

700

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));

126,965✔

701

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

126,965✔

703

END:

126,965✔

704

  PRINT_LOG_END

126,965✔

705

  return code;

126,965✔

708

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

124,266✔

709

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

124,266✔

712

  int32_t sz = taosArrayGetSize(topics);

124,266✔

713

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

125,042✔

714

    char *name = taosArrayGetP(topics, i);

1,988✔

715

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

1,988✔

716

      return true;

1,212✔

719

  return false;

123,054✔

722

static int32_t checkConsumer(STrans *pTrans, SMqConsumerObj *pConsumer, bool deleteConsumer, char *topicName) {

40,694✔

723

  int32_t         code = 0;

40,694✔

724

  int32_t         lino = 0;

40,694✔

725

  SMqConsumerObj *pConsumerNew = NULL;

40,694✔

727

  taosRLockLatch(&pConsumer->lock);

40,694✔

728

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

40,694✔

729

  bool found2 = checkTopic(pConsumer->rebRemovedTopics, topicName);

40,694✔

730

  bool found3 = checkTopic(pConsumer->rebNewTopics, topicName);

40,694✔

731

  if (found1 || found2 || found3) {

40,694✔

732

    if (deleteConsumer) {

824✔

733

      MND_TMQ_RETURN_CHECK(tNewSMqConsumerObj(pConsumer->consumerId, pConsumer->cgroup, CONSUMER_CLEAR, NULL, NULL, &pConsumerNew));

824✔

734

      MND_TMQ_RETURN_CHECK(mndSetConsumerDropLogs(pTrans, pConsumerNew));

824✔

735

      tDeleteSMqConsumerObj(pConsumerNew);

824✔

736

      pConsumerNew = NULL;

824✔

744

END:

40,694✔

745

  taosRUnLockLatch(&pConsumer->lock);

40,694✔

746

  tDeleteSMqConsumerObj(pConsumerNew);

40,694✔

747

  return code;

40,694✔

750

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

126,965✔

751

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

126,965✔

754

  int32_t         code = 0;

126,965✔

755

  int32_t         lino = 0;

126,965✔

756

  SSdb *          pSdb = pMnode->pSdb;

126,965✔

757

  void *          pIter = NULL;

126,965✔

758

  SMqConsumerObj *pConsumer = NULL;

126,965✔

760

  PRINT_LOG_START

126,965✔

762

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

167,659✔

763

    if (pIter == NULL) {

167,659✔

764

      break;

126,965✔

767

    MND_TMQ_RETURN_CHECK(checkConsumer(pTrans, pConsumer, deleteConsumer, topicName));

40,694✔

768

    sdbRelease(pSdb, pConsumer);

40,694✔

771

END:

126,965✔

772

  PRINT_LOG_END

126,965✔

773

  sdbRelease(pSdb, pConsumer);

126,965✔

774

  sdbCancelFetch(pSdb, pIter);

126,965✔

775

  return code;

126,965✔

778

static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {

137,192✔

779

  if (pReq == NULL) {

137,192✔

782

  SMnode *       pMnode = pReq->info.node;

137,192✔

783

  SMDropTopicReq dropReq = {0};

137,192✔

784

  int32_t        code = 0;

137,192✔

785

  int32_t        lino = 0;

137,192✔

786

  SMqTopicObj *  pTopic = NULL;

137,192✔

787

  STrans *       pTrans = NULL;

137,192✔

788

  SUserObj      *pOperUser = NULL;

137,192✔

789

  int64_t        tss = taosGetTimestampMs();

137,192✔

791

  PRINT_LOG_START

137,192✔

792

  MND_TMQ_RETURN_CHECK(tDeserializeSMDropTopicReq(pReq->pCont, pReq->contLen, &dropReq));

137,192✔

794

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

137,192✔

795

  if (code != 0) {

137,192✔

796

    if (dropReq.igNotExists) {

9,639✔

797

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

9,639✔

798

      code = 0;

9,639✔

800

    goto END;

9,639✔

802

  taosRLockLatch(&pTopic->lock);

127,553✔

804

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

127,553✔

805

  MND_TMQ_NULL_CHECK(pTrans);

127,553✔

807

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

127,553✔

808

  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

127,553✔

809

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

127,553✔

811

  MND_TMQ_RETURN_CHECK(mndAcquireUser(pMnode, RPC_MSG_USER(pReq), &pOperUser));

127,553✔

815

  MND_TMQ_RETURN_CHECK(

127,553✔

817

  MND_TMQ_RETURN_CHECK(mndCheckObjPrivilegeRecF(pMnode, pOperUser, PRIV_CM_DROP, PRIV_OBJ_TOPIC, pTopic->ownerId,

127,369✔

819

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

126,965✔

820

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

126,965✔

821

  MND_TMQ_RETURN_CHECK(mndDropTopic(pMnode, pTrans, pReq, pTopic));

126,965✔

822

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {

126,965✔

823

    int64_t tse = taosGetTimestampMs();

126,965✔

824

    double  duration = (double)(tse - tss);

126,965✔

825

    duration = duration / 1000;

126,965✔

826

    auditRecord(pReq, pMnode->clusterId, "dropTopic", pTopic->db, dropReq.name, dropReq.sql, dropReq.sqlLen, duration,

126,965✔

830

  code = TSDB_CODE_ACTION_IN_PROGRESS;

126,965✔

832

END:

137,192✔

833

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

137,192✔

834

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

588✔

836

    mInfo("topic:%s dropped successfully", dropReq.name);

136,604✔

838

  if (pTopic != NULL) {

137,192✔

839

    taosRUnLockLatch(&pTopic->lock);

127,553✔

841

  mndReleaseTopic(pMnode, pTopic);

137,192✔

842

  mndReleaseUser(pMnode, pOperUser);

137,192✔

843

  mndTransDrop(pTrans);

137,192✔

844

  tFreeSMDropTopicReq(&dropReq);

137,192✔

845

  return code;

137,192✔

848

int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTopics) {

132,789✔

849

  if (pMnode == NULL || dbName == NULL || pNumOfTopics == NULL) {

132,789✔

852

  *pNumOfTopics = 0;

132,789✔

854

  SSdb *  pSdb = pMnode->pSdb;

132,789✔

855

  SDbObj *pDb = mndAcquireDb(pMnode, dbName);

132,789✔

856

  if (pDb == NULL) {

132,789✔

860

  int32_t numOfTopics = 0;

132,789✔

861

  void *  pIter = NULL;

132,789✔

862

  while (1) {

247✔

863

    SMqTopicObj *pTopic = NULL;

133,036✔

864

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

133,036✔

865

    if (pIter == NULL) {

133,036✔

866

      break;

132,789✔

868

    taosRLockLatch(&pTopic->lock);

247✔

869

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

247✔

870

      numOfTopics++;

247✔

872

    taosRUnLockLatch(&pTopic->lock);

247✔

874

    sdbRelease(pSdb, pTopic);

247✔

877

  *pNumOfTopics = numOfTopics;

132,789✔

878

  mndReleaseDb(pMnode, pDb);

132,789✔

879

  return 0;

132,789✔

882

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

63,100✔

883

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

63,100✔

886

  char *  string = NULL;

63,100✔

887

  int32_t code = 0;

63,100✔

888

  int32_t lino = 0;

63,100✔

890

  cJSON *cbytes = NULL;

63,100✔

891

  cJSON *ctype = NULL;

63,100✔

892

  cJSON *cname = NULL;

63,100✔

893

  cJSON *column = NULL;

63,100✔

894

  cJSON *columns = cJSON_CreateArray();

63,100✔

895

  MND_TMQ_NULL_CHECK(columns);

63,100✔

896

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

742,036✔

897

    column = cJSON_CreateObject();

678,936✔

898

    MND_TMQ_NULL_CHECK(column);

678,936✔

899

    SSchema *s = schema + i;

678,936✔

900

    cname = cJSON_CreateString(s->name);

678,936✔

901

    MND_TMQ_NULL_CHECK(cname);

678,936✔

902

    MND_TMQ_CONDITION_CHECK(cJSON_AddItemToObject(column, "name", cname), 0);

678,936✔

903

    cname = NULL;  // ownership transferred to column object

678,936✔

905

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

678,936✔

906

    MND_TMQ_NULL_CHECK(ctype);

678,936✔

907

    MND_TMQ_CONDITION_CHECK(cJSON_AddItemToObject(column, "type", ctype), 0);

678,936✔

908

    ctype = NULL;  // ownership transferred to column object

678,936✔

910

    int32_t length = 0;

678,936✔

911

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

678,936✔

912

      length = s->bytes - VARSTR_HEADER_SIZE;

131,836✔

913

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

547,100✔

914

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

79,616✔

916

      length = s->bytes;

467,484✔

918

    cbytes = cJSON_CreateNumber(length);

678,936✔

919

    MND_TMQ_NULL_CHECK(cbytes);

678,936✔

920

    MND_TMQ_CONDITION_CHECK(cJSON_AddItemToObject(column, "length", cbytes), 0);

678,936✔

921

    cbytes = NULL;  // ownership transferred to column object

678,936✔

923

    MND_TMQ_CONDITION_CHECK(cJSON_AddItemToArray(columns, column), 0);

678,936✔

924

    column = NULL;  // ownership transferred to columns array

678,936✔

927

END:

63,100✔

928

  string = cJSON_PrintUnformatted(columns);

63,100✔

929

  cJSON_Delete(columns);

63,100✔

930

  cJSON_Delete(column);

63,100✔

931

  cJSON_Delete(cname);

63,100✔

932

  cJSON_Delete(ctype);

63,100✔

933

  cJSON_Delete(cbytes);

63,100✔

935

  size_t len = strlen(string);

63,100✔

936

  if (string && len <= TSDB_SHOW_SCHEMA_JSON_LEN) {

63,100✔

937

    STR_TO_VARSTR(schemaJson, string);

63,100✔

942

  taosMemoryFree(string);

63,100✔

945

static int32_t buildResult(SMqTopicObj *pTopic, int32_t *numOfRows, SMnode *pMnode, SSDataBlock *pBlock) {

64,819✔

946

  SColumnInfoData *pColInfo = NULL;

64,819✔

947

  SName            n = {0};

64,819✔

948

  int32_t          cols = 0;

64,819✔

949

  char *           schemaJson = NULL;

64,819✔

950

  char *           sql = NULL;

64,819✔

951

  int32_t          code = 0;

64,819✔

952

  int32_t          lino = 0;

64,819✔

954

  taosRLockLatch(&pTopic->lock);

64,819✔

956

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

64,819✔

957

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

64,819✔

958

  STR_TO_VARSTR(topicName, pName);

64,819✔

960

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

64,819✔

961

  MND_TMQ_NULL_CHECK(pColInfo);

64,819✔

962

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

64,819✔

964

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

64,819✔

965

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

64,819✔

966

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

64,819✔

967

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

64,819✔

969

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

64,819✔

970

  MND_TMQ_NULL_CHECK(pColInfo);

64,819✔

971

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

64,819✔

973

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

64,819✔

974

  MND_TMQ_NULL_CHECK(pColInfo);

64,819✔

975

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

64,819✔

977

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

64,819✔

978

  MND_TMQ_NULL_CHECK(sql);

64,819✔

979

  STR_TO_VARSTR(sql, pTopic->sql);

64,819✔

981

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

64,819✔

982

  MND_TMQ_NULL_CHECK(pColInfo);

64,819✔

983

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

64,819✔

985

  taosMemoryFreeClear(sql);

64,819✔

987

  schemaJson = taosMemoryMalloc(TSDB_SHOW_SCHEMA_JSON_LEN + VARSTR_HEADER_SIZE);

64,819✔

988

  MND_TMQ_NULL_CHECK(schemaJson);

64,819✔

989

  if (pTopic->subType == TOPIC_SUB_TYPE__COLUMN) {

64,819✔

990

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

62,689✔

991

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

2,130✔

992

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

411✔

993

    if (pStb == NULL) {

411✔

997

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

411✔

998

      mndReleaseStb(pMnode, pStb);

411✔

1001

    STR_TO_VARSTR(schemaJson, "NULL");

1,719✔

1004

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

64,819✔

1005

  MND_TMQ_NULL_CHECK(pColInfo);

64,819✔

1006

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

64,819✔

1007

  taosMemoryFreeClear(schemaJson);

64,819✔

1009

  (*numOfRows)++;

64,819✔

1011

END:

64,819✔

1012

  taosRUnLockLatch(&pTopic->lock);

64,819✔

1014

  taosMemoryFreeClear(sql);

64,819✔

1015

  taosMemoryFreeClear(schemaJson);

64,819✔

1016

  return code;

64,819✔

1019

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

28,541✔

1020

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

28,541✔

1023

  SMnode      *pMnode = pReq->info.node;

28,541✔

1024

  SSdb        *pSdb = pMnode->pSdb;

28,541✔

1025

  int32_t      numOfRows = 0;

28,541✔

1026

  SMqTopicObj *pTopic = NULL;

28,541✔

1027

  SUserObj    *pOperUser = NULL;

28,541✔

1028

  int32_t      code = 0, lino = 0;

28,541✔

1029

  char        *sql = NULL;

28,541✔

1030

  char        *schemaJson = NULL;

28,541✔

1031

  char         objFName[TSDB_OBJ_FNAME_LEN + 1] = {0};

28,541✔

1032

  bool         showAll = false;

28,541✔

1034

  MND_TMQ_RETURN_CHECK(mndAcquireUser(pMnode, RPC_MSG_USER(pReq), &pOperUser));

28,541✔

1035

  (void)snprintf(objFName, sizeof(objFName), "%d.*", pOperUser->acctId);

28,541✔

1036

  int32_t objLevel = privObjGetLevel(PRIV_OBJ_TOPIC);

28,541✔

1037

  showAll =

28,541✔

1038

      (0 == mndCheckSysObjPrivilege(pMnode, pOperUser, RPC_MSG_TOKEN(pReq), PRIV_CM_SHOW, PRIV_OBJ_TOPIC, 0, objFName,

28,541✔

1041

  PRINT_LOG_START

28,541✔

1043

  while (numOfRows < rowsCapacity) {

93,912✔

1044

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

93,912✔

1045

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

93,912✔

1047

    if (!showAll) {

65,371✔

1048

      if (mndCheckObjPrivilegeRecF(pMnode, pOperUser, PRIV_CM_SHOW, PRIV_OBJ_TOPIC, pTopic->ownerId, pTopic->db,

1,472✔

1049

                                   objLevel == 0 ? NULL : mndGetDbStr(pTopic->name)) != 0) {  // 1.topic1

736✔

1050

        sdbRelease(pSdb, pTopic);

552✔

1051

        continue;

552✔

1055

    MND_TMQ_RETURN_CHECK(buildResult(pTopic, &numOfRows, pMnode, pBlock));

64,819✔

1056

    sdbRelease(pSdb, pTopic);

64,819✔

1057

    pTopic = NULL;

64,819✔

1059

  pShow->numOfRows += numOfRows;

28,541✔

1061

END:

28,541✔

1062

  sdbCancelFetch(pSdb, pShow->pIter);

28,541✔

1063

  sdbRelease(pSdb, pTopic);

28,541✔

1064

  mndReleaseUser(pMnode, pOperUser);

28,541✔

1065

  if (code != TSDB_CODE_SUCCESS) {

28,541✔

1069

    mDebug("%s retrieved %d rows successfully", __func__, numOfRows);

28,541✔

1070

    return numOfRows;

28,541✔

1080

bool mndTopicExistsForDb(SMnode *pMnode, SDbObj *pDb) {

847,449✔

1081

  if (pMnode == NULL || pDb == NULL) {

847,449✔

1084

  SSdb *       pSdb = pMnode->pSdb;

847,449✔

1085

  void *       pIter = NULL;

847,449✔

1086

  SMqTopicObj *pTopic = NULL;

847,449✔

1088

  while (1) {

64,435✔

1089

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

911,884✔

1090

    if (pIter == NULL) {

911,884✔

1091

      break;

847,025✔

1094

    taosRLockLatch(&pTopic->lock);

64,859✔

1095

    bool found = pTopic->dbUid == pDb->uid;

64,859✔

1096

    taosRUnLockLatch(&pTopic->lock);

64,859✔

1098

    if (found) {

64,859✔

1099

      sdbRelease(pSdb, pTopic);

424✔

1100

      sdbCancelFetch(pSdb, pIter);

424✔

1101

      return true;

424✔

1104

    sdbRelease(pSdb, pTopic);

64,435✔

1107

  return false;

847,025✔

taosdata / TDengine / #5054

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

taosdata / TDengine / #5054

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