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

53

int32_t mndInitTopic(SMnode *pMnode) {

525,949✔

54

  SSdbTable table = {

525,949✔

64

  if (pMnode == NULL) {

525,949✔

67

  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CREATE_TOPIC, mndProcessCreateTopicReq);

525,949✔

68

  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_TOPIC, mndProcessDropTopicReq);

525,949✔

70

  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndRetrieveTopic);

525,949✔

71

  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndCancelGetNextTopic);

525,949✔

73

  return sdbSetTable(pMnode->pSdb, table);

525,949✔

76

void mndCleanupTopic(SMnode *pMnode) {}

525,887✔

78

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

172,645✔

79

  if (fullTopic == NULL) {

172,645✔

82

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

172,645✔

83

  if (tmp == NULL) {

172,645✔

86

    tstrncpy(topic, tmp + 1, TSDB_TOPIC_FNAME_LEN);

172,645✔

89

SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {

359,349✔

90

  if (pTopic == NULL) {

359,349✔

93

  int32_t code = 0;

359,349✔

94

  int32_t lino = 0;

359,349✔

95

  terrno = TSDB_CODE_OUT_OF_MEMORY;

359,349✔

97

  void *  swBuf = NULL;

359,349✔

98

  int32_t physicalPlanLen = 0;

359,349✔

99

  if (pTopic->physicalPlan) {

359,349✔

100

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

359,349✔

103

  int32_t schemaLen = 0;

359,349✔

104

  if (pTopic->schema.nCols) {

359,349✔

105

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

553,674✔

108

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

359,349✔

109

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

359,349✔

110

  if (pRaw == NULL) {

359,349✔

114

  int32_t dataPos = 0;

359,349✔

115

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

359,349✔

116

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

359,349✔

117

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

359,349✔

118

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

359,349✔

119

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

359,349✔

120

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

359,349✔

121

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

359,349✔

122

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

359,349✔

123

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

359,349✔

124

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

359,349✔

126

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

359,349✔

127

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

359,349✔

128

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

359,349✔

129

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

359,349✔

130

  SDB_SET_INT32(pRaw, dataPos, physicalPlanLen, TOPIC_ENCODE_OVER);

359,349✔

131

  if (physicalPlanLen) {

359,349✔

132

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

359,349✔

134

  SDB_SET_INT32(pRaw, dataPos, schemaLen, TOPIC_ENCODE_OVER);

359,349✔

135

  if (schemaLen) {

359,349✔

136

    swBuf = taosMemoryMalloc(schemaLen);

276,837✔

137

    if (swBuf == NULL) {

276,837✔

140

    void *aswBuf = swBuf;

276,837✔

141

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

553,674✔

144

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

276,837✔

146

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

359,349✔

147

  SDB_SET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_ENCODE_OVER);

359,349✔

148

  SDB_SET_DATALEN(pRaw, dataPos, TOPIC_ENCODE_OVER);

359,349✔

150

  terrno = TSDB_CODE_SUCCESS;

359,349✔

152

TOPIC_ENCODE_OVER:

359,349✔

153

  if (swBuf) taosMemoryFree(swBuf);

359,349✔

154

  if (terrno != TSDB_CODE_SUCCESS) {

359,349✔

160

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

359,349✔

161

  return pRaw;

359,349✔

164

SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {

304,626✔

165

  if (pRaw == NULL) return NULL;

304,626✔

166

  int32_t code = 0;

304,626✔

167

  int32_t lino = 0;

304,626✔

168

  terrno = TSDB_CODE_OUT_OF_MEMORY;

304,626✔

169

  SSdbRow *    pRow = NULL;

304,626✔

170

  SMqTopicObj *pTopic = NULL;

304,626✔

171

  void *       buf = NULL;

304,626✔

172

  char*        ast = NULL;

304,626✔

174

  int8_t sver = 0;

304,626✔

175

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

304,626✔

177

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

304,626✔

182

  pRow = sdbAllocRow(sizeof(SMqTopicObj));

304,626✔

183

  if (pRow == NULL) goto TOPIC_DECODE_OVER;

304,626✔

185

  pTopic = sdbGetRowObj(pRow);

304,626✔

186

  if (pTopic == NULL) goto TOPIC_DECODE_OVER;

304,626✔

188

  int32_t len = 0;

304,626✔

189

  int32_t dataPos = 0;

304,626✔

190

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

304,626✔

191

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

304,626✔

192

  if (sver >= 2) {

304,626✔

193

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

304,626✔

195

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

304,626✔

196

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

304,626✔

197

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

304,626✔

198

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

304,626✔

199

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

304,626✔

200

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

304,626✔

201

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

304,626✔

203

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

304,626✔

204

  if (sver >= 3) {

304,626✔

205

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

304,626✔

207

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

304,626✔

208

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

304,626✔

209

  if (pTopic->sql == NULL) {

304,626✔

213

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

304,626✔

215

  if (sver < MND_TOPIC_VER_SUPPORT_OWNER) {

304,626✔

231

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

304,626✔

232

    if (len) {

304,626✔

233

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

304,626✔

234

      if (pTopic->physicalPlan == NULL) {

304,626✔

238

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

304,626✔

243

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

304,626✔

244

    if (len) {

304,626✔

245

      buf = taosMemoryMalloc(len);

236,822✔

246

      if (buf == NULL) {

236,822✔

250

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

236,822✔

251

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

473,644✔

255

      pTopic->schema.nCols = 0;

67,804✔

256

      pTopic->schema.version = 0;

67,804✔

257

      pTopic->schema.pSchema = NULL;

67,804✔

259

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

304,626✔

262

  SDB_GET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_DECODE_OVER);

304,626✔

263

  terrno = TSDB_CODE_SUCCESS;

304,626✔

265

TOPIC_DECODE_OVER:

304,626✔

266

  taosMemoryFreeClear(buf);

304,626✔

267

  taosMemoryFreeClear(ast);

304,626✔

269

  if (terrno != TSDB_CODE_SUCCESS) {

304,626✔

275

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

304,626✔

276

  return pRow;

304,626✔

279

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

181,513✔

280

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

181,513✔

281

  return 0;

181,513✔

284

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

304,626✔

285

  if (pTopic == NULL) return 0;

304,626✔

286

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

304,626✔

287

  taosMemoryFreeClear(pTopic->sql);

304,626✔

288

  taosMemoryFreeClear(pTopic->physicalPlan);

304,626✔

289

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

304,626✔

290

  return 0;

304,626✔

293

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

2,471✔

294

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

2,471✔

295

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

2,471✔

296

  taosWLockLatch(&pOldTopic->lock);

2,471✔

297

  SMqTopicObj tmpTopic = *pOldTopic;

2,471✔

298

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

2,471✔

299

  *pNewTopic = tmpTopic;

2,471✔

300

  taosWUnLockLatch(&pOldTopic->lock);

2,471✔

301

  return 0;

2,471✔

304

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

2,865,222✔

305

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

2,865,222✔

308

  SSdb *pSdb = pMnode->pSdb;

2,865,222✔

309

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

2,865,222✔

310

  if (*pTopic == NULL) {

2,865,222✔

311

    return TSDB_CODE_MND_TOPIC_NOT_EXIST;

208,328✔

313

  return TSDB_CODE_SUCCESS;

2,656,894✔

316

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

3,128,224✔

317

  if (pMnode == NULL) return;

3,128,224✔

318

  SSdb *pSdb = pMnode->pSdb;

3,128,224✔

319

  sdbRelease(pSdb, pTopic);

3,128,224✔

322

static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {

162,081✔

323

  if (pCreate == NULL) return TSDB_CODE_INVALID_PARA;

162,081✔

324

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

162,081✔

326

  if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) {

162,081✔

327

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

117,647✔

328

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

44,434✔

329

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

15,113✔

330

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

29,321✔

331

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

29,321✔

334

  return 0;

162,081✔

337

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

158,193✔

338

  SNode *     pAst = NULL;

158,193✔

339

  SQueryPlan *pPlan = NULL;

158,193✔

340

  int32_t     code = TSDB_CODE_SUCCESS;

158,193✔

341

  int32_t     lino = 0;

158,193✔

343

  PRINT_LOG_START

158,193✔

344

  if (ast == NULL) {

158,193✔

345

    topicObj->physicalPlan = taosStrdup("");

35,748✔

346

    goto END;

35,748✔

348

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

122,445✔

349

  MND_TMQ_RETURN_CHECK(nodesStringToNode(ast, &pAst));

122,445✔

350

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

122,445✔

352

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

122,445✔

353

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

122,445✔

354

  if (pPlan == NULL) {

122,445✔

358

  int32_t levelNum = LIST_LENGTH(pPlan->pSubplans);

122,445✔

359

  if (levelNum != 1) {

122,445✔

364

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

122,445✔

365

  MND_TMQ_NULL_CHECK(pNodeListNode);

122,445✔

366

  int32_t opNum = LIST_LENGTH(pNodeListNode->pNodeList);

122,445✔

367

  if (opNum != 1) {

122,445✔

372

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

122,445✔

374

END:

158,193✔

375

  nodesDestroyNode(pAst);

158,193✔

376

  qDestroyQueryPlan(pPlan);

158,193✔

377

  PRINT_LOG_END

158,193✔

378

  return code;

158,193✔

381

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

155,722✔

383

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

155,722✔

385

  STrans *    pTrans = NULL;

155,722✔

386

  int32_t     code = 0;

155,722✔

387

  int32_t     lino = 0;

155,722✔

388

  SMqTopicObj topicObj = {0};

155,722✔

390

  PRINT_LOG_START

155,722✔

391

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

155,722✔

392

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

155,722✔

393

  MND_TMQ_NULL_CHECK(pTrans);

155,722✔

394

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

155,722✔

395

  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

155,722✔

397

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

155,722✔

398

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

155,722✔

399

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

155,722✔

400

  topicObj.ownerId = pOperUser->uid;

155,722✔

403

  if (pDb) {

155,722✔

405

    MND_TMQ_RETURN_CHECK(

155,722✔

409

  topicObj.createTime = taosGetTimestampMs();

155,722✔

410

  topicObj.updateTime = topicObj.createTime;

155,722✔

411

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

155,722✔

412

  topicObj.dbUid = pDb->uid;

155,722✔

413

  topicObj.version = 1;

155,722✔

414

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

155,722✔

415

  MND_TMQ_NULL_CHECK(topicObj.sql);

155,722✔

416

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

155,722✔

417

  topicObj.subType = pCreate->subType;

155,722✔

418

  topicObj.withMeta = pCreate->withMeta;

155,722✔

419

  taosInitRWLatch(&topicObj.lock);

155,722✔

421

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

155,722✔

423

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

155,722✔

424

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

116,106✔

425

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

116,106✔

426

    MND_TMQ_NULL_CHECK(pStb);

116,106✔

427

    char stbName[TSDB_TABLE_NAME_LEN] = {0};

115,360✔

428

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

115,360✔

429

    MND_TMQ_RETURN_CHECK(

115,360✔

431

    MND_TMQ_RETURN_CHECK(

115,360✔

433

    topicObj.stbUid = pStb->uid;

115,360✔

434

    mndReleaseStb(pMnode, pStb);

115,360✔

437

  if (pCreate->subType == TOPIC_SUB_TYPE__DB) {

154,976✔

438

    MND_TMQ_RETURN_CHECK(

28,169✔

440

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

28,169✔

441

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

126,807✔

445

  SSdbRaw *pCommitRaw = mndTopicActionEncode(&topicObj);

154,976✔

446

  MND_TMQ_NULL_CHECK(pCommitRaw);

154,976✔

447

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

154,976✔

448

  if (code != 0) {

154,976✔

453

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));

154,976✔

454

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

154,976✔

456

END:

155,722✔

457

  PRINT_LOG_END

155,722✔

458

  taosMemoryFreeClear(topicObj.sql);

155,722✔

459

  taosMemoryFreeClear(topicObj.physicalPlan);

155,722✔

460

  if (topicObj.schema.nCols) {

155,722✔

461

    taosMemoryFreeClear(topicObj.schema.pSchema);

119,974✔

463

  mndTransDrop(pTrans);

155,722✔

464

  return code;

155,722✔

467

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

2,471✔

469

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

2,471✔

471

  STrans *    pTrans = NULL;

2,471✔

472

  int32_t     code = 0;

2,471✔

473

  int32_t     lino = 0;

2,471✔

474

  SMqTopicObj topicObj = {0};

2,471✔

476

  PRINT_LOG_START

2,471✔

477

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

2,471✔

478

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

2,471✔

479

  MND_TMQ_NULL_CHECK(pTrans);

2,471✔

480

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

2,471✔

481

  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

2,471✔

483

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

2,471✔

484

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

2,471✔

485

  tstrncpy(topicObj.createUser, userName, TSDB_USER_LEN);

2,471✔

487

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

2,471✔

489

  taosRLockLatch(&topicObjOri->lock);

2,471✔

490

  topicObj.createTime = topicObjOri->createTime;

2,471✔

491

  topicObj.updateTime = taosGetTimestampMs();

2,471✔

492

  topicObj.uid = topicObjOri->uid;

2,471✔

493

  topicObj.dbUid = pDb->uid;

2,471✔

494

  topicObj.version = topicObjOri->version + 1;

2,471✔

495

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

2,471✔

496

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

2,471✔

497

  topicObj.subType = pCreate->subType;

2,471✔

498

  topicObj.withMeta = pCreate->withMeta;

2,471✔

499

  taosInitRWLatch(&topicObj.lock);

2,471✔

500

  taosRUnLockLatch(&topicObjOri->lock);

2,471✔

502

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

2,471✔

504

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

2,471✔

505

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

1,412✔

506

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

1,412✔

507

    MND_TMQ_NULL_CHECK(pStb);

1,412✔

508

    topicObj.stbUid = pStb->uid;

1,412✔

509

    mndReleaseStb(pMnode, pStb);

1,412✔

512

  SSdbRaw *pCommitRaw = mndTopicActionEncode(&topicObj);

2,471✔

513

  MND_TMQ_NULL_CHECK(pCommitRaw);

2,471✔

514

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

2,471✔

515

  if (code != 0) {

2,471✔

520

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));

2,471✔

521

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

2,471✔

523

END:

2,471✔

524

  PRINT_LOG_END

2,471✔

525

  taosMemoryFreeClear(topicObj.sql);

2,471✔

526

  taosMemoryFreeClear(topicObj.physicalPlan);

2,471✔

527

  if (topicObj.schema.nCols) {

2,471✔

528

    taosMemoryFreeClear(topicObj.schema.pSchema);

2,471✔

530

  mndTransDrop(pTrans);

2,471✔

531

  return code;

2,471✔

534

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

158,904✔

535

  SMqTopicObj *pTopic = NULL;

158,904✔

536

  SDbObj *     pDb = NULL;

158,904✔

537

  int32_t      code = TSDB_CODE_SUCCESS;

158,904✔

538

  int32_t      lino = 0;

158,904✔

539

  SMnode *     pMnode = pReq->info.node;

158,904✔

540

  int64_t      tss = taosGetTimestampMs();

158,904✔

542

  PRINT_LOG_START

158,904✔

543

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

158,904✔

544

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

158,904✔

545

  if (code == TSDB_CODE_SUCCESS) {

158,904✔

546

    mndReleaseTopic(pMnode, pTopic);

812✔

547

    if (createTopicReq->igExists) {

812✔

548

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

812✔

552

    goto END;

812✔

553

  } else if (code != TSDB_CODE_MND_TOPIC_NOT_EXIST) {

158,092✔

557

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

158,092✔

558

  MND_TMQ_NULL_CHECK(pDb);

158,092✔

560

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

157,346✔

565

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

157,346✔

566

    code = TSDB_CODE_TMQ_TOPIC_OUT_OF_RANGE;

1,624✔

567

    goto END;

1,624✔

570

  MND_TMQ_RETURN_CHECK(grantCheck(TSDB_GRANT_SUBSCRIPTION));

155,722✔

571

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

155,722✔

572

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {

154,976✔

573

    int64_t tse = taosGetTimestampMs();

154,976✔

574

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

154,976✔

575

    duration = duration / 1000;

154,976✔

576

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

154,976✔

577

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

154,976✔

579

  code = TSDB_CODE_ACTION_IN_PROGRESS;

154,976✔

581

END:

158,904✔

582

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

158,904✔

583

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

3,116✔

585

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

155,788✔

587

  mndReleaseDb(pMnode, pDb);

158,904✔

588

  return code;

158,904✔

591

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

3,177✔

592

  SMnode *     pMnode = pReq->info.node;

3,177✔

593

  int32_t      code = TSDB_CODE_SUCCESS;

3,177✔

594

  int32_t      lino = 0;

3,177✔

595

  SDbObj *     pDb = NULL;

3,177✔

596

  SMqTopicObj *pTopic = NULL;

3,177✔

597

  int64_t      tss = taosGetTimestampMs();

3,177✔

599

  PRINT_LOG_START

3,177✔

600

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

3,177✔

601

  if (code != 0) {

3,177✔

602

    if (createTopicReq->igExists) {

706✔

603

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

353✔

604

      code = 0;

353✔

605

      goto END;

353✔

607

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

353✔

608

      goto END;

353✔

612

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

2,471✔

613

  MND_TMQ_NULL_CHECK(pDb);

2,471✔

615

  MND_TMQ_RETURN_CHECK(grantCheck(TSDB_GRANT_SUBSCRIPTION));

2,471✔

616

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

2,471✔

618

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {

2,471✔

619

    int64_t tse = taosGetTimestampMs();

2,471✔

620

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

2,471✔

621

    duration = duration / 1000;

2,471✔

622

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

2,471✔

623

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

2,471✔

626

  code = TSDB_CODE_ACTION_IN_PROGRESS;

2,471✔

628

  if (topicsToReload == NULL) {

2,471✔

629

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

353✔

630

    MND_TMQ_NULL_CHECK(topicsToReload);

353✔

632

  MND_TMQ_RETURN_CHECK(

2,471✔

634

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

2,471✔

636

END:

3,177✔

637

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

3,177✔

638

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

353✔

640

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

2,824✔

642

  mndReleaseTopic(pMnode, pTopic);

3,177✔

643

  mndReleaseDb(pMnode, pDb);

3,177✔

645

  return code;

3,177✔

648

static int32_t mndProcessCreateTopicReq(SRpcMsg *pReq) {

162,081✔

649

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

162,081✔

652

  SMnode *pMnode = pReq->info.node;

162,081✔

653

  SUserObj *pOperUser = NULL;

162,081✔

654

  int32_t code = TSDB_CODE_SUCCESS;

162,081✔

655

  int32_t lino = 0;

162,081✔

657

  SCMCreateTopicReq createTopicReq = {0};

162,081✔

659

  PRINT_LOG_START

162,081✔

660

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

162,081✔

662

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

162,081✔

664

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

162,081✔

666

  MND_TMQ_RETURN_CHECK(mndCheckCreateTopicReq(&createTopicReq));

162,081✔

668

  if (createTopicReq.reload) {

162,081✔

669

    MND_TMQ_RETURN_CHECK(reloadTopic(pReq, &createTopicReq));

3,177✔

671

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

158,904✔

674

END:

161,967✔

675

  tFreeSCMCreateTopicReq(&createTopicReq);

162,081✔

676

  mndReleaseUser(pMnode, pOperUser);

162,081✔

677

  return code;

162,081✔

680

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

120,310✔

681

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

120,310✔

684

  int32_t  code = 0;

120,310✔

685

  int32_t  lino = 0;

120,310✔

686

  SSdbRaw *pCommitRaw = NULL;

120,310✔

687

  PRINT_LOG_START

120,310✔

688

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

120,310✔

689

  mndTopicGetShowName(pTopic->name, topicFName);

120,310✔

690

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

120,310✔

691

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

120,310✔

692

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

120,310✔

693

  pCommitRaw = mndTopicActionEncode(pTopic);

120,310✔

694

  MND_TMQ_NULL_CHECK(pCommitRaw);

120,310✔

695

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

120,310✔

696

  if (code != 0) {

120,310✔

700

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));

120,310✔

701

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

120,310✔

703

END:

120,310✔

704

  PRINT_LOG_END

120,310✔

705

  return code;

120,310✔

708

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

118,923✔

709

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

118,923✔

712

  int32_t sz = taosArrayGetSize(topics);

118,923✔

713

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

119,671✔

714

    char *name = taosArrayGetP(topics, i);

1,915✔

715

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

1,915✔

716

      return true;

1,167✔

719

  return false;

117,756✔

722

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

38,941✔

723

  int32_t         code = 0;

38,941✔

724

  int32_t         lino = 0;

38,941✔

725

  SMqConsumerObj *pConsumerNew = NULL;

38,941✔

727

  taosRLockLatch(&pConsumer->lock);

38,941✔

728

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

38,941✔

729

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

38,941✔

730

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

38,941✔

731

  if (found1 || found2 || found3) {

38,941✔

732

    if (deleteConsumer) {

794✔

733

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

794✔

734

      MND_TMQ_RETURN_CHECK(mndSetConsumerDropLogs(pTrans, pConsumerNew));

794✔

735

      tDeleteSMqConsumerObj(pConsumerNew);

794✔

736

      pConsumerNew = NULL;

794✔

744

END:

38,941✔

745

  taosRUnLockLatch(&pConsumer->lock);

38,941✔

746

  tDeleteSMqConsumerObj(pConsumerNew);

38,941✔

747

  return code;

38,941✔

750

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

120,310✔

751

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

120,310✔

754

  int32_t         code = 0;

120,310✔

755

  int32_t         lino = 0;

120,310✔

756

  SSdb *          pSdb = pMnode->pSdb;

120,310✔

757

  void *          pIter = NULL;

120,310✔

758

  SMqConsumerObj *pConsumer = NULL;

120,310✔

760

  PRINT_LOG_START

120,310✔

762

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

159,251✔

763

    if (pIter == NULL) {

159,251✔

764

      break;

120,310✔

767

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

38,941✔

768

    sdbRelease(pSdb, pConsumer);

38,941✔

771

END:

120,310✔

772

  PRINT_LOG_END

120,310✔

773

  sdbRelease(pSdb, pConsumer);

120,310✔

774

  sdbCancelFetch(pSdb, pIter);

120,310✔

775

  return code;

120,310✔

778

static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {

130,067✔

779

  if (pReq == NULL) {

130,067✔

782

  SMnode *       pMnode = pReq->info.node;

130,067✔

783

  SMDropTopicReq dropReq = {0};

130,067✔

784

  int32_t        code = 0;

130,067✔

785

  int32_t        lino = 0;

130,067✔

786

  SMqTopicObj *  pTopic = NULL;

130,067✔

787

  STrans *       pTrans = NULL;

130,067✔

788

  SUserObj      *pOperUser = NULL;

130,067✔

789

  int64_t        tss = taosGetTimestampMs();

130,067✔

791

  PRINT_LOG_START

130,067✔

792

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

130,067✔

794

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

130,067✔

795

  if (code != 0) {

130,067✔

796

    if (dropReq.igNotExists) {

9,225✔

797

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

9,225✔

798

      code = 0;

9,225✔

800

    goto END;

9,225✔

802

  taosRLockLatch(&pTopic->lock);

120,842✔

804

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

120,842✔

805

  MND_TMQ_NULL_CHECK(pTrans);

120,842✔

807

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

120,842✔

808

  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

120,842✔

809

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

120,842✔

811

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

120,842✔

815

  MND_TMQ_RETURN_CHECK(

120,842✔

817

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

120,678✔

819

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

120,310✔

820

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

120,310✔

821

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

120,310✔

822

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {

120,310✔

823

    int64_t tse = taosGetTimestampMs();

120,310✔

824

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

120,310✔

825

    duration = duration / 1000;

120,310✔

826

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

120,310✔

830

  code = TSDB_CODE_ACTION_IN_PROGRESS;

120,310✔

832

END:

130,067✔

833

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

130,067✔

834

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

532✔

836

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

129,535✔

838

  if (pTopic != NULL) {

130,067✔

839

    taosRUnLockLatch(&pTopic->lock);

120,842✔

841

  mndReleaseTopic(pMnode, pTopic);

130,067✔

842

  mndReleaseUser(pMnode, pOperUser);

130,067✔

843

  mndTransDrop(pTrans);

130,067✔

844

  tFreeSMDropTopicReq(&dropReq);

130,067✔

845

  return code;

130,067✔

848

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

129,422✔

849

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

129,422✔

852

  *pNumOfTopics = 0;

129,422✔

854

  SSdb *  pSdb = pMnode->pSdb;

129,422✔

855

  SDbObj *pDb = mndAcquireDb(pMnode, dbName);

129,422✔

856

  if (pDb == NULL) {

129,422✔

860

  int32_t numOfTopics = 0;

129,422✔

861

  void *  pIter = NULL;

129,422✔

862

  while (1) {

226✔

863

    SMqTopicObj *pTopic = NULL;

129,648✔

864

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

129,648✔

865

    if (pIter == NULL) {

129,648✔

866

      break;

129,422✔

868

    taosRLockLatch(&pTopic->lock);

226✔

869

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

226✔

870

      numOfTopics++;

226✔

872

    taosRUnLockLatch(&pTopic->lock);

226✔

874

    sdbRelease(pSdb, pTopic);

226✔

877

  *pNumOfTopics = numOfTopics;

129,422✔

878

  mndReleaseDb(pMnode, pDb);

129,422✔

879

  return 0;

129,422✔

882

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

57,164✔

883

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

57,164✔

886

  char *  string = NULL;

57,164✔

887

  int32_t code = 0;

57,164✔

888

  int32_t lino = 0;

57,164✔

890

  cJSON *cbytes = NULL;

57,164✔

891

  cJSON *ctype = NULL;

57,164✔

892

  cJSON *cname = NULL;

57,164✔

893

  cJSON *column = NULL;

57,164✔

894

  cJSON *columns = cJSON_CreateArray();

57,164✔

895

  MND_TMQ_NULL_CHECK(columns);

57,164✔

896

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

641,561✔

897

    column = cJSON_CreateObject();

584,397✔

898

    MND_TMQ_NULL_CHECK(column);

584,397✔

899

    SSchema *s = schema + i;

584,397✔

900

    cname = cJSON_CreateString(s->name);

584,397✔

901

    MND_TMQ_NULL_CHECK(cname);

584,397✔

902

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

584,397✔

903

    cname = NULL;  // ownership transferred to column object

584,397✔

905

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

584,397✔

906

    MND_TMQ_NULL_CHECK(ctype);

584,397✔

907

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

584,397✔

908

    ctype = NULL;  // ownership transferred to column object

584,397✔

910

    int32_t length = 0;

584,397✔

911

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

584,397✔

912

      length = s->bytes - VARSTR_HEADER_SIZE;

109,779✔

913

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

474,618✔

914

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

72,946✔

916

      length = s->bytes;

401,672✔

918

    cbytes = cJSON_CreateNumber(length);

584,397✔

919

    MND_TMQ_NULL_CHECK(cbytes);

584,397✔

920

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

584,397✔

921

    cbytes = NULL;  // ownership transferred to column object

584,397✔

923

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

584,397✔

924

    column = NULL;  // ownership transferred to columns array

584,397✔

927

END:

57,164✔

928

  string = cJSON_PrintUnformatted(columns);

57,164✔

929

  cJSON_Delete(columns);

57,164✔

930

  cJSON_Delete(column);

57,164✔

931

  cJSON_Delete(cname);

57,164✔

932

  cJSON_Delete(ctype);

57,164✔

933

  cJSON_Delete(cbytes);

57,164✔

935

  size_t len = strlen(string);

57,164✔

936

  if (string && len <= TSDB_SHOW_SCHEMA_JSON_LEN) {

57,164✔

937

    STR_TO_VARSTR(schemaJson, string);

57,164✔

942

  taosMemoryFree(string);

57,164✔

945

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

58,835✔

946

  SColumnInfoData *pColInfo = NULL;

58,835✔

947

  SName            n = {0};

58,835✔

948

  int32_t          cols = 0;

58,835✔

949

  char *           schemaJson = NULL;

58,835✔

950

  char *           sql = NULL;

58,835✔

951

  int32_t          code = 0;

58,835✔

952

  int32_t          lino = 0;

58,835✔

954

  taosRLockLatch(&pTopic->lock);

58,835✔

956

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

58,835✔

957

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

58,835✔

958

  STR_TO_VARSTR(topicName, pName);

58,835✔

960

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

58,835✔

961

  MND_TMQ_NULL_CHECK(pColInfo);

58,835✔

962

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

58,835✔

964

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

58,835✔

965

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

58,835✔

966

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

58,835✔

967

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

58,835✔

969

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

58,835✔

970

  MND_TMQ_NULL_CHECK(pColInfo);

58,835✔

971

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

58,835✔

973

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

58,835✔

974

  MND_TMQ_NULL_CHECK(pColInfo);

58,835✔

975

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

58,835✔

977

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

58,835✔

978

  MND_TMQ_NULL_CHECK(sql);

58,835✔

979

  STR_TO_VARSTR(sql, pTopic->sql);

58,835✔

981

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

58,835✔

982

  MND_TMQ_NULL_CHECK(pColInfo);

58,835✔

983

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

58,835✔

985

  taosMemoryFreeClear(sql);

58,835✔

987

  schemaJson = taosMemoryMalloc(TSDB_SHOW_SCHEMA_JSON_LEN + VARSTR_HEADER_SIZE);

58,835✔

988

  MND_TMQ_NULL_CHECK(schemaJson);

58,835✔

989

  if (pTopic->subType == TOPIC_SUB_TYPE__COLUMN) {

58,835✔

990

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

56,756✔

991

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

2,079✔

992

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

408✔

993

    if (pStb == NULL) {

408✔

997

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

408✔

998

      mndReleaseStb(pMnode, pStb);

408✔

1001

    STR_TO_VARSTR(schemaJson, "NULL");

1,671✔

1004

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

58,835✔

1005

  MND_TMQ_NULL_CHECK(pColInfo);

58,835✔

1006

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

58,835✔

1007

  taosMemoryFreeClear(schemaJson);

58,835✔

1009

  (*numOfRows)++;

58,835✔

1011

END:

58,835✔

1012

  taosRUnLockLatch(&pTopic->lock);

58,835✔

1014

  taosMemoryFreeClear(sql);

58,835✔

1015

  taosMemoryFreeClear(schemaJson);

58,835✔

1016

  return code;

58,835✔

1019

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

22,313✔

1020

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

22,313✔

1023

  SMnode      *pMnode = pReq->info.node;

22,313✔

1024

  SSdb        *pSdb = pMnode->pSdb;

22,313✔

1025

  int32_t      numOfRows = 0;

22,313✔

1026

  SMqTopicObj *pTopic = NULL;

22,313✔

1027

  SUserObj    *pOperUser = NULL;

22,313✔

1028

  int32_t      code = 0, lino = 0;

22,313✔

1029

  char        *sql = NULL;

22,313✔

1030

  char        *schemaJson = NULL;

22,313✔

1031

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

22,313✔

1032

  bool         showAll = false;

22,313✔

1034

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

22,313✔

1035

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

22,313✔

1036

  int32_t objLevel = privObjGetLevel(PRIV_OBJ_TOPIC);

22,313✔

1037

  showAll =

22,313✔

1038

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

22,313✔

1041

  PRINT_LOG_START

22,313✔

1043

  while (numOfRows < rowsCapacity) {

81,640✔

1044

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

81,640✔

1045

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

81,640✔

1047

    if (!showAll) {

59,327✔

1048

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

1,312✔

1049

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

656✔

1050

        sdbRelease(pSdb, pTopic);

492✔

1051

        continue;

492✔

1055

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

58,835✔

1056

    sdbRelease(pSdb, pTopic);

58,835✔

1057

    pTopic = NULL;

58,835✔

1059

  pShow->numOfRows += numOfRows;

22,313✔

1061

END:

22,313✔

1062

  sdbCancelFetch(pSdb, pShow->pIter);

22,313✔

1063

  sdbRelease(pSdb, pTopic);

22,313✔

1064

  mndReleaseUser(pMnode, pOperUser);

22,313✔

1065

  if (code != TSDB_CODE_SUCCESS) {

22,313✔

1069

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

22,313✔

1070

    return numOfRows;

22,313✔

1080

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

817,540✔

1081

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

817,540✔

1084

  SSdb *       pSdb = pMnode->pSdb;

817,540✔

1085

  void *       pIter = NULL;

817,540✔

1086

  SMqTopicObj *pTopic = NULL;

817,540✔

1088

  while (1) {

61,735✔

1089

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

879,275✔

1090

    if (pIter == NULL) {

879,275✔

1091

      break;

817,134✔

1094

    taosRLockLatch(&pTopic->lock);

62,141✔

1095

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

62,141✔

1096

    taosRUnLockLatch(&pTopic->lock);

62,141✔

1098

    if (found) {

62,141✔

1099

      sdbRelease(pSdb, pTopic);

406✔

1100

      sdbCancelFetch(pSdb, pIter);

406✔

1101

      return true;

406✔

1104

    sdbRelease(pSdb, pTopic);

61,735✔

1107

  return false;

817,134✔

taosdata / TDengine / #5052

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

taosdata / TDengine / #5052

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