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

53

int32_t mndInitTopic(SMnode *pMnode) {

484,191✔

54

  SSdbTable table = {

484,191✔

64

  if (pMnode == NULL) {

484,191✔

67

  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CREATE_TOPIC, mndProcessCreateTopicReq);

484,191✔

68

  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_TOPIC, mndProcessDropTopicReq);

484,191✔

70

  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndRetrieveTopic);

484,191✔

71

  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndCancelGetNextTopic);

484,191✔

73

  return sdbSetTable(pMnode->pSdb, table);

484,191✔

76

void mndCleanupTopic(SMnode *pMnode) {}

484,127✔

78

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

186,682✔

79

  if (fullTopic == NULL) {

186,682✔

82

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

186,682✔

83

  if (tmp == NULL) {

186,682✔

86

    tstrncpy(topic, tmp + 1, TSDB_TOPIC_FNAME_LEN);

186,682✔

89

SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {

335,886✔

90

  if (pTopic == NULL) {

335,886✔

93

  int32_t code = 0;

335,886✔

94

  int32_t lino = 0;

335,886✔

95

  terrno = TSDB_CODE_OUT_OF_MEMORY;

335,886✔

97

  void *  swBuf = NULL;

335,886✔

98

  int32_t physicalPlanLen = 0;

335,886✔

99

  if (pTopic->physicalPlan) {

335,886✔

100

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

335,886✔

103

  int32_t schemaLen = 0;

335,886✔

104

  if (pTopic->schema.nCols) {

335,886✔

105

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

517,428✔

108

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

335,886✔

109

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

335,886✔

110

  if (pRaw == NULL) {

335,886✔

114

  int32_t dataPos = 0;

335,886✔

115

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

335,886✔

116

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

335,886✔

117

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

335,886✔

118

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

335,886✔

119

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

335,886✔

120

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

335,886✔

121

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

335,886✔

122

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

335,886✔

123

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

335,886✔

124

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

335,886✔

126

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

335,886✔

127

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

335,886✔

128

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

335,886✔

129

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

335,886✔

130

  SDB_SET_INT32(pRaw, dataPos, physicalPlanLen, TOPIC_ENCODE_OVER);

335,886✔

131

  if (physicalPlanLen) {

335,886✔

132

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

335,886✔

134

  SDB_SET_INT32(pRaw, dataPos, schemaLen, TOPIC_ENCODE_OVER);

335,886✔

135

  if (schemaLen) {

335,886✔

136

    swBuf = taosMemoryMalloc(schemaLen);

258,714✔

137

    if (swBuf == NULL) {

258,714✔

140

    void *aswBuf = swBuf;

258,714✔

141

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

517,428✔

144

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

258,714✔

146

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

335,886✔

147

  SDB_SET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_ENCODE_OVER);

335,886✔

148

  SDB_SET_DATALEN(pRaw, dataPos, TOPIC_ENCODE_OVER);

335,886✔

150

  terrno = TSDB_CODE_SUCCESS;

335,886✔

152

TOPIC_ENCODE_OVER:

335,886✔

153

  if (swBuf) taosMemoryFree(swBuf);

335,886✔

154

  if (terrno != TSDB_CODE_SUCCESS) {

335,886✔

160

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

335,886✔

161

  return pRaw;

335,886✔

164

SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {

286,032✔

165

  if (pRaw == NULL) return NULL;

286,032✔

166

  int32_t code = 0;

286,032✔

167

  int32_t lino = 0;

286,032✔

168

  terrno = TSDB_CODE_OUT_OF_MEMORY;

286,032✔

169

  SSdbRow *    pRow = NULL;

286,032✔

170

  SMqTopicObj *pTopic = NULL;

286,032✔

171

  void *       buf = NULL;

286,032✔

172

  char*        ast = NULL;

286,032✔

174

  int8_t sver = 0;

286,032✔

175

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

286,032✔

177

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

286,032✔

182

  pRow = sdbAllocRow(sizeof(SMqTopicObj));

286,032✔

183

  if (pRow == NULL) goto TOPIC_DECODE_OVER;

286,032✔

185

  pTopic = sdbGetRowObj(pRow);

286,032✔

186

  if (pTopic == NULL) goto TOPIC_DECODE_OVER;

286,032✔

188

  int32_t len = 0;

286,032✔

189

  int32_t dataPos = 0;

286,032✔

190

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

286,032✔

191

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

286,032✔

192

  if (sver >= 2) {

286,032✔

193

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

286,032✔

195

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

286,032✔

196

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

286,032✔

197

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

286,032✔

198

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

286,032✔

199

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

286,032✔

200

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

286,032✔

201

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

286,032✔

203

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

286,032✔

204

  if (sver >= 3) {

286,032✔

205

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

286,032✔

207

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

286,032✔

208

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

286,032✔

209

  if (pTopic->sql == NULL) {

286,032✔

213

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

286,032✔

215

  if (sver < MND_TOPIC_VER_SUPPORT_OWNER) {

286,032✔

231

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

286,032✔

232

    if (len) {

286,032✔

233

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

286,032✔

234

      if (pTopic->physicalPlan == NULL) {

286,032✔

238

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

286,032✔

243

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

286,032✔

244

    if (len) {

286,032✔

245

      buf = taosMemoryMalloc(len);

222,557✔

246

      if (buf == NULL) {

222,557✔

250

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

222,557✔

251

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

445,114✔

255

      pTopic->schema.nCols = 0;

63,475✔

256

      pTopic->schema.version = 0;

63,475✔

257

      pTopic->schema.pSchema = NULL;

63,475✔

259

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

286,032✔

262

  SDB_GET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_DECODE_OVER);

286,032✔

263

  terrno = TSDB_CODE_SUCCESS;

286,032✔

265

TOPIC_DECODE_OVER:

286,032✔

266

  taosMemoryFreeClear(buf);

286,032✔

267

  taosMemoryFreeClear(ast);

286,032✔

269

  if (terrno != TSDB_CODE_SUCCESS) {

286,032✔

275

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

286,032✔

276

  return pRow;

286,032✔

279

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

170,025✔

280

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

170,025✔

281

  return 0;

170,025✔

284

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

286,032✔

285

  if (pTopic == NULL) return 0;

286,032✔

286

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

286,032✔

287

  taosMemoryFreeClear(pTopic->sql);

286,032✔

288

  taosMemoryFreeClear(pTopic->physicalPlan);

286,032✔

289

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

286,032✔

290

  return 0;

286,032✔

293

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

2,282✔

294

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

2,282✔

295

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

2,282✔

296

  taosWLockLatch(&pOldTopic->lock);

2,282✔

297

  SMqTopicObj tmpTopic = *pOldTopic;

2,282✔

298

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

2,282✔

299

  *pNewTopic = tmpTopic;

2,282✔

300

  taosWUnLockLatch(&pOldTopic->lock);

2,282✔

301

  return 0;

2,282✔

304

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

2,719,426✔

305

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

2,719,426✔

308

  SSdb *pSdb = pMnode->pSdb;

2,719,426✔

309

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

2,719,426✔

310

  if (*pTopic == NULL) {

2,719,426✔

311

    return TSDB_CODE_MND_TOPIC_NOT_EXIST;

195,468✔

313

  return TSDB_CODE_SUCCESS;

2,523,958✔

316

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

2,968,990✔

317

  if (pMnode == NULL) return;

2,968,990✔

318

  SSdb *pSdb = pMnode->pSdb;

2,968,990✔

319

  sdbRelease(pSdb, pTopic);

2,968,990✔

322

static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {

151,808✔

323

  if (pCreate == NULL) return TSDB_CODE_INVALID_PARA;

151,808✔

324

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

151,808✔

326

  if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) {

151,808✔

327

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

110,250✔

328

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

41,558✔

329

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

14,043✔

330

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

27,515✔

331

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

27,515✔

334

  return 0;

151,808✔

337

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

148,127✔

338

  SNode *     pAst = NULL;

148,127✔

339

  SQueryPlan *pPlan = NULL;

148,127✔

340

  int32_t     code = TSDB_CODE_SUCCESS;

148,127✔

341

  int32_t     lino = 0;

148,127✔

343

  PRINT_LOG_START

148,127✔

344

  if (ast == NULL) {

148,127✔

345

    topicObj->physicalPlan = taosStrdup("");

33,482✔

346

    goto END;

33,482✔

348

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

114,645✔

349

  MND_TMQ_RETURN_CHECK(nodesStringToNode(ast, &pAst));

114,645✔

350

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

114,645✔

352

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

114,645✔

353

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

114,645✔

354

  if (pPlan == NULL) {

114,645✔

358

  int32_t levelNum = LIST_LENGTH(pPlan->pSubplans);

114,645✔

359

  if (levelNum != 1) {

114,645✔

364

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

114,645✔

365

  MND_TMQ_NULL_CHECK(pNodeListNode);

114,645✔

366

  int32_t opNum = LIST_LENGTH(pNodeListNode->pNodeList);

114,645✔

367

  if (opNum != 1) {

114,645✔

372

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

114,645✔

374

END:

148,127✔

375

  nodesDestroyNode(pAst);

148,127✔

376

  qDestroyQueryPlan(pPlan);

148,127✔

377

  PRINT_LOG_END

148,127✔

378

  return code;

148,127✔

381

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

145,845✔

383

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

145,845✔

385

  STrans *    pTrans = NULL;

145,845✔

386

  int32_t     code = 0;

145,845✔

387

  int32_t     lino = 0;

145,845✔

388

  SMqTopicObj topicObj = {0};

145,845✔

390

  PRINT_LOG_START

145,845✔

391

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

145,845✔

392

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

145,845✔

393

  MND_TMQ_NULL_CHECK(pTrans);

145,845✔

394

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

145,845✔

395

  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

145,845✔

397

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

145,845✔

398

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

145,845✔

399

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

145,845✔

400

  topicObj.ownerId = pOperUser->uid;

145,845✔

403

  if (pDb) {

145,845✔

405

    MND_TMQ_RETURN_CHECK(

145,845✔

409

  topicObj.createTime = taosGetTimestampMs();

145,845✔

410

  topicObj.updateTime = topicObj.createTime;

145,845✔

411

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

145,845✔

412

  topicObj.dbUid = pDb->uid;

145,845✔

413

  topicObj.version = 1;

145,845✔

414

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

145,845✔

415

  MND_TMQ_NULL_CHECK(topicObj.sql);

145,845✔

416

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

145,845✔

417

  topicObj.subType = pCreate->subType;

145,845✔

418

  topicObj.withMeta = pCreate->withMeta;

145,845✔

419

  taosInitRWLatch(&topicObj.lock);

145,845✔

421

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

145,845✔

423

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

145,845✔

424

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

108,699✔

425

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

108,699✔

426

    MND_TMQ_NULL_CHECK(pStb);

108,699✔

427

    char stbName[TSDB_TABLE_NAME_LEN] = {0};

107,993✔

428

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

107,993✔

429

    MND_TMQ_RETURN_CHECK(

107,993✔

431

    MND_TMQ_RETURN_CHECK(

107,993✔

433

    topicObj.stbUid = pStb->uid;

107,993✔

434

    mndReleaseStb(pMnode, pStb);

107,993✔

437

  if (pCreate->subType == TOPIC_SUB_TYPE__DB) {

145,139✔

438

    MND_TMQ_RETURN_CHECK(

26,420✔

440

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

26,420✔

441

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

118,719✔

445

  SSdbRaw *pCommitRaw = mndTopicActionEncode(&topicObj);

145,139✔

446

  MND_TMQ_NULL_CHECK(pCommitRaw);

145,139✔

447

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

145,139✔

448

  if (code != 0) {

145,139✔

453

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));

145,139✔

454

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

145,139✔

456

END:

145,845✔

457

  PRINT_LOG_END

145,845✔

458

  taosMemoryFreeClear(topicObj.sql);

145,845✔

459

  taosMemoryFreeClear(topicObj.physicalPlan);

145,845✔

460

  if (topicObj.schema.nCols) {

145,845✔

461

    taosMemoryFreeClear(topicObj.schema.pSchema);

112,363✔

463

  mndTransDrop(pTrans);

145,845✔

464

  return code;

145,845✔

467

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

2,282✔

469

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

2,282✔

471

  STrans *    pTrans = NULL;

2,282✔

472

  int32_t     code = 0;

2,282✔

473

  int32_t     lino = 0;

2,282✔

474

  SMqTopicObj topicObj = {0};

2,282✔

476

  PRINT_LOG_START

2,282✔

477

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

2,282✔

478

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

2,282✔

479

  MND_TMQ_NULL_CHECK(pTrans);

2,282✔

480

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

2,282✔

481

  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

2,282✔

483

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

2,282✔

484

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

2,282✔

485

  tstrncpy(topicObj.createUser, userName, TSDB_USER_LEN);

2,282✔

487

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

2,282✔

489

  taosRLockLatch(&topicObjOri->lock);

2,282✔

490

  topicObj.createTime = topicObjOri->createTime;

2,282✔

491

  topicObj.updateTime = taosGetTimestampMs();

2,282✔

492

  topicObj.uid = topicObjOri->uid;

2,282✔

493

  topicObj.dbUid = pDb->uid;

2,282✔

494

  topicObj.version = topicObjOri->version + 1;

2,282✔

495

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

2,282✔

496

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

2,282✔

497

  topicObj.subType = pCreate->subType;

2,282✔

498

  topicObj.withMeta = pCreate->withMeta;

2,282✔

499

  taosInitRWLatch(&topicObj.lock);

2,282✔

500

  taosRUnLockLatch(&topicObjOri->lock);

2,282✔

502

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

2,282✔

504

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

2,282✔

505

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

1,304✔

506

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

1,304✔

507

    MND_TMQ_NULL_CHECK(pStb);

1,304✔

508

    topicObj.stbUid = pStb->uid;

1,304✔

509

    mndReleaseStb(pMnode, pStb);

1,304✔

512

  SSdbRaw *pCommitRaw = mndTopicActionEncode(&topicObj);

2,282✔

513

  MND_TMQ_NULL_CHECK(pCommitRaw);

2,282✔

514

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

2,282✔

515

  if (code != 0) {

2,282✔

520

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));

2,282✔

521

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

2,282✔

523

END:

2,282✔

524

  PRINT_LOG_END

2,282✔

525

  taosMemoryFreeClear(topicObj.sql);

2,282✔

526

  taosMemoryFreeClear(topicObj.physicalPlan);

2,282✔

527

  if (topicObj.schema.nCols) {

2,282✔

528

    taosMemoryFreeClear(topicObj.schema.pSchema);

2,282✔

530

  mndTransDrop(pTrans);

2,282✔

531

  return code;

2,282✔

534

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

148,874✔

535

  SMqTopicObj *pTopic = NULL;

148,874✔

536

  SDbObj *     pDb = NULL;

148,874✔

537

  int32_t      code = TSDB_CODE_SUCCESS;

148,874✔

538

  int32_t      lino = 0;

148,874✔

539

  SMnode *     pMnode = pReq->info.node;

148,874✔

540

  int64_t      tss = taosGetTimestampMs();

148,874✔

542

  PRINT_LOG_START

148,874✔

543

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

148,874✔

544

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

148,874✔

545

  if (code == TSDB_CODE_SUCCESS) {

148,874✔

546

    mndReleaseTopic(pMnode, pTopic);

770✔

547

    if (createTopicReq->igExists) {

770✔

548

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

770✔

552

    goto END;

770✔

553

  } else if (code != TSDB_CODE_MND_TOPIC_NOT_EXIST) {

148,104✔

557

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

148,104✔

558

  MND_TMQ_NULL_CHECK(pDb);

148,104✔

560

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

147,398✔

565

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

147,398✔

566

    code = TSDB_CODE_TMQ_TOPIC_OUT_OF_RANGE;

1,553✔

567

    goto END;

1,553✔

570

  MND_TMQ_RETURN_CHECK(grantCheck(TSDB_GRANT_SUBSCRIPTION));

145,845✔

571

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

145,845✔

572

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {

145,139✔

573

    int64_t tse = taosGetTimestampMs();

145,139✔

574

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

145,139✔

575

    duration = duration / 1000;

145,139✔

576

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

145,139✔

577

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

145,139✔

579

  code = TSDB_CODE_ACTION_IN_PROGRESS;

145,139✔

581

END:

148,874✔

582

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

148,874✔

583

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

2,965✔

585

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

145,909✔

587

  mndReleaseDb(pMnode, pDb);

148,874✔

588

  return code;

148,874✔

591

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

2,934✔

592

  SMnode *     pMnode = pReq->info.node;

2,934✔

593

  int32_t      code = TSDB_CODE_SUCCESS;

2,934✔

594

  int32_t      lino = 0;

2,934✔

595

  SDbObj *     pDb = NULL;

2,934✔

596

  SMqTopicObj *pTopic = NULL;

2,934✔

597

  int64_t      tss = taosGetTimestampMs();

2,934✔

599

  PRINT_LOG_START

2,934✔

600

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

2,934✔

601

  if (code != 0) {

2,934✔

602

    if (createTopicReq->igExists) {

652✔

603

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

326✔

604

      code = 0;

326✔

605

      goto END;

326✔

607

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

326✔

608

      goto END;

326✔

612

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

2,282✔

613

  MND_TMQ_NULL_CHECK(pDb);

2,282✔

615

  MND_TMQ_RETURN_CHECK(grantCheck(TSDB_GRANT_SUBSCRIPTION));

2,282✔

616

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

2,282✔

618

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {

2,282✔

619

    int64_t tse = taosGetTimestampMs();

2,282✔

620

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

2,282✔

621

    duration = duration / 1000;

2,282✔

622

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

2,282✔

623

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

2,282✔

626

  code = TSDB_CODE_ACTION_IN_PROGRESS;

2,282✔

628

  if (topicsToReload == NULL) {

2,282✔

629

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

326✔

630

    MND_TMQ_NULL_CHECK(topicsToReload);

326✔

632

  MND_TMQ_RETURN_CHECK(

2,282✔

634

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

2,282✔

636

END:

2,934✔

637

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

2,934✔

638

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

326✔

640

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

2,608✔

642

  mndReleaseTopic(pMnode, pTopic);

2,934✔

643

  mndReleaseDb(pMnode, pDb);

2,934✔

645

  return code;

2,934✔

648

static int32_t mndProcessCreateTopicReq(SRpcMsg *pReq) {

151,808✔

649

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

151,808✔

652

  SMnode *pMnode = pReq->info.node;

151,808✔

653

  SUserObj *pOperUser = NULL;

151,808✔

654

  int32_t code = TSDB_CODE_SUCCESS;

151,808✔

655

  int32_t lino = 0;

151,808✔

657

  SCMCreateTopicReq createTopicReq = {0};

151,808✔

659

  PRINT_LOG_START

151,808✔

660

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

151,808✔

662

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

151,808✔

664

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

151,808✔

666

  MND_TMQ_RETURN_CHECK(mndCheckCreateTopicReq(&createTopicReq));

151,808✔

668

  if (createTopicReq.reload) {

151,808✔

669

    MND_TMQ_RETURN_CHECK(reloadTopic(pReq, &createTopicReq));

2,934✔

671

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

148,874✔

674

END:

151,695✔

675

  tFreeSCMCreateTopicReq(&createTopicReq);

151,808✔

676

  mndReleaseUser(pMnode, pOperUser);

151,808✔

677

  return code;

151,808✔

680

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

112,811✔

681

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

112,811✔

684

  int32_t  code = 0;

112,811✔

685

  int32_t  lino = 0;

112,811✔

686

  SSdbRaw *pCommitRaw = NULL;

112,811✔

687

  PRINT_LOG_START

112,811✔

688

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

112,811✔

689

  mndTopicGetShowName(pTopic->name, topicFName);

112,811✔

690

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

112,811✔

691

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

112,811✔

692

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

112,811✔

693

  pCommitRaw = mndTopicActionEncode(pTopic);

112,811✔

694

  MND_TMQ_NULL_CHECK(pCommitRaw);

112,811✔

695

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

112,811✔

696

  if (code != 0) {

112,811✔

700

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));

112,811✔

701

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

112,811✔

703

END:

112,811✔

704

  PRINT_LOG_END

112,811✔

705

  return code;

112,811✔

708

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

108,942✔

709

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

108,942✔

712

  int32_t sz = taosArrayGetSize(topics);

108,942✔

713

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

109,633✔

714

    char *name = taosArrayGetP(topics, i);

1,790✔

715

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

1,790✔

716

      return true;

1,099✔

719

  return false;

107,843✔

722

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

35,663✔

723

  int32_t         code = 0;

35,663✔

724

  int32_t         lino = 0;

35,663✔

725

  SMqConsumerObj *pConsumerNew = NULL;

35,663✔

727

  taosRLockLatch(&pConsumer->lock);

35,663✔

728

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

35,663✔

729

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

35,663✔

730

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

35,663✔

731

  if (found1 || found2 || found3) {

35,663✔

732

    if (deleteConsumer) {

746✔

733

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

746✔

734

      MND_TMQ_RETURN_CHECK(mndSetConsumerDropLogs(pTrans, pConsumerNew));

746✔

735

      tDeleteSMqConsumerObj(pConsumerNew);

746✔

736

      pConsumerNew = NULL;

746✔

744

END:

35,663✔

745

  taosRUnLockLatch(&pConsumer->lock);

35,663✔

746

  tDeleteSMqConsumerObj(pConsumerNew);

35,663✔

747

  return code;

35,663✔

750

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

112,811✔

751

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

112,811✔

754

  int32_t         code = 0;

112,811✔

755

  int32_t         lino = 0;

112,811✔

756

  SSdb *          pSdb = pMnode->pSdb;

112,811✔

757

  void *          pIter = NULL;

112,811✔

758

  SMqConsumerObj *pConsumer = NULL;

112,811✔

760

  PRINT_LOG_START

112,811✔

762

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

148,474✔

763

    if (pIter == NULL) {

148,474✔

764

      break;

112,811✔

767

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

35,663✔

768

    sdbRelease(pSdb, pConsumer);

35,663✔

771

END:

112,811✔

772

  PRINT_LOG_END

112,811✔

773

  sdbRelease(pSdb, pConsumer);

112,811✔

774

  sdbCancelFetch(pSdb, pIter);

112,811✔

775

  return code;

112,811✔

778

static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {

121,589✔

779

  if (pReq == NULL) {

121,589✔

782

  SMnode *       pMnode = pReq->info.node;

121,589✔

783

  SMDropTopicReq dropReq = {0};

121,589✔

784

  int32_t        code = 0;

121,589✔

785

  int32_t        lino = 0;

121,589✔

786

  SMqTopicObj *  pTopic = NULL;

121,589✔

787

  STrans *       pTrans = NULL;

121,589✔

788

  SUserObj      *pOperUser = NULL;

121,589✔

789

  int64_t        tss = taosGetTimestampMs();

121,589✔

791

  PRINT_LOG_START

121,589✔

792

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

121,589✔

794

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

121,589✔

795

  if (code != 0) {

121,589✔

796

    if (dropReq.igNotExists) {

8,303✔

797

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

8,303✔

798

      code = 0;

8,303✔

800

    goto END;

8,303✔

802

  taosRLockLatch(&pTopic->lock);

113,286✔

804

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

113,286✔

805

  MND_TMQ_NULL_CHECK(pTrans);

113,286✔

807

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

113,286✔

808

  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

113,286✔

809

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

113,286✔

811

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

113,286✔

815

  MND_TMQ_RETURN_CHECK(

113,286✔

817

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

113,140✔

819

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

112,811✔

820

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

112,811✔

821

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

112,811✔

822

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {

112,811✔

823

    int64_t tse = taosGetTimestampMs();

112,811✔

824

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

112,811✔

825

    duration = duration / 1000;

112,811✔

826

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

112,811✔

830

  code = TSDB_CODE_ACTION_IN_PROGRESS;

112,811✔

832

END:

121,589✔

833

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

121,589✔

834

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

475✔

836

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

121,114✔

838

  if (pTopic != NULL) {

121,589✔

839

    taosRUnLockLatch(&pTopic->lock);

113,286✔

841

  mndReleaseTopic(pMnode, pTopic);

121,589✔

842

  mndReleaseUser(pMnode, pOperUser);

121,589✔

843

  mndTransDrop(pTrans);

121,589✔

844

  tFreeSMDropTopicReq(&dropReq);

121,589✔

845

  return code;

121,589✔

848

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

122,862✔

849

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

122,862✔

852

  *pNumOfTopics = 0;

122,862✔

854

  SSdb *  pSdb = pMnode->pSdb;

122,862✔

855

  SDbObj *pDb = mndAcquireDb(pMnode, dbName);

122,862✔

856

  if (pDb == NULL) {

122,862✔

860

  int32_t numOfTopics = 0;

122,862✔

861

  void *  pIter = NULL;

122,862✔

862

  while (1) {

210✔

863

    SMqTopicObj *pTopic = NULL;

123,072✔

864

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

123,072✔

865

    if (pIter == NULL) {

123,072✔

866

      break;

122,862✔

868

    taosRLockLatch(&pTopic->lock);

210✔

869

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

210✔

870

      numOfTopics++;

210✔

872

    taosRUnLockLatch(&pTopic->lock);

210✔

874

    sdbRelease(pSdb, pTopic);

210✔

877

  *pNumOfTopics = numOfTopics;

122,862✔

878

  mndReleaseDb(pMnode, pDb);

122,862✔

879

  return 0;

122,862✔

882

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

54,041✔

883

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

54,041✔

886

  char *  string = NULL;

54,041✔

887

  int32_t code = 0;

54,041✔

888

  int32_t lino = 0;

54,041✔

890

  cJSON *cbytes = NULL;

54,041✔

891

  cJSON *ctype = NULL;

54,041✔

892

  cJSON *cname = NULL;

54,041✔

893

  cJSON *column = NULL;

54,041✔

894

  cJSON *columns = cJSON_CreateArray();

54,041✔

895

  MND_TMQ_NULL_CHECK(columns);

54,041✔

896

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

606,301✔

897

    column = cJSON_CreateObject();

552,260✔

898

    MND_TMQ_NULL_CHECK(column);

552,260✔

899

    SSchema *s = schema + i;

552,260✔

900

    cname = cJSON_CreateString(s->name);

552,260✔

901

    MND_TMQ_NULL_CHECK(cname);

552,260✔

902

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

552,260✔

903

    cname = NULL;  // ownership transferred to column object

552,260✔

905

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

552,260✔

906

    MND_TMQ_NULL_CHECK(ctype);

552,260✔

907

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

552,260✔

908

    ctype = NULL;  // ownership transferred to column object

552,260✔

910

    int32_t length = 0;

552,260✔

911

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

552,260✔

912

      length = s->bytes - VARSTR_HEADER_SIZE;

103,462✔

913

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

448,798✔

914

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

69,452✔

916

      length = s->bytes;

379,346✔

918

    cbytes = cJSON_CreateNumber(length);

552,260✔

919

    MND_TMQ_NULL_CHECK(cbytes);

552,260✔

920

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

552,260✔

921

    cbytes = NULL;  // ownership transferred to column object

552,260✔

923

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

552,260✔

924

    column = NULL;  // ownership transferred to columns array

552,260✔

927

END:

54,041✔

928

  string = cJSON_PrintUnformatted(columns);

54,041✔

929

  cJSON_Delete(columns);

54,041✔

930

  cJSON_Delete(column);

54,041✔

931

  cJSON_Delete(cname);

54,041✔

932

  cJSON_Delete(ctype);

54,041✔

933

  cJSON_Delete(cbytes);

54,041✔

935

  size_t len = strlen(string);

54,041✔

936

  if (string && len <= TSDB_SHOW_SCHEMA_JSON_LEN) {

54,041✔

937

    STR_TO_VARSTR(schemaJson, string);

54,041✔

942

  taosMemoryFree(string);

54,041✔

945

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

55,586✔

946

  SColumnInfoData *pColInfo = NULL;

55,586✔

947

  SName            n = {0};

55,586✔

948

  int32_t          cols = 0;

55,586✔

949

  char *           schemaJson = NULL;

55,586✔

950

  char *           sql = NULL;

55,586✔

951

  int32_t          code = 0;

55,586✔

952

  int32_t          lino = 0;

55,586✔

954

  taosRLockLatch(&pTopic->lock);

55,586✔

956

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

55,586✔

957

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

55,586✔

958

  STR_TO_VARSTR(topicName, pName);

55,586✔

960

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

55,586✔

961

  MND_TMQ_NULL_CHECK(pColInfo);

55,586✔

962

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

55,586✔

964

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

55,586✔

965

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

55,586✔

966

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

55,586✔

967

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

55,586✔

969

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

55,586✔

970

  MND_TMQ_NULL_CHECK(pColInfo);

55,586✔

971

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

55,586✔

973

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

55,586✔

974

  MND_TMQ_NULL_CHECK(pColInfo);

55,586✔

975

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

55,586✔

977

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

55,586✔

978

  MND_TMQ_NULL_CHECK(sql);

55,586✔

979

  STR_TO_VARSTR(sql, pTopic->sql);

55,586✔

981

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

55,586✔

982

  MND_TMQ_NULL_CHECK(pColInfo);

55,586✔

983

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

55,586✔

985

  taosMemoryFreeClear(sql);

55,586✔

987

  schemaJson = taosMemoryMalloc(TSDB_SHOW_SCHEMA_JSON_LEN + VARSTR_HEADER_SIZE);

55,586✔

988

  MND_TMQ_NULL_CHECK(schemaJson);

55,586✔

989

  if (pTopic->subType == TOPIC_SUB_TYPE__COLUMN) {

55,586✔

990

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

53,675✔

991

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

1,911✔

992

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

366✔

993

    if (pStb == NULL) {

366✔

997

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

366✔

998

      mndReleaseStb(pMnode, pStb);

366✔

1001

    STR_TO_VARSTR(schemaJson, "NULL");

1,545✔

1004

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

55,586✔

1005

  MND_TMQ_NULL_CHECK(pColInfo);

55,586✔

1006

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

55,586✔

1007

  taosMemoryFreeClear(schemaJson);

55,586✔

1009

  (*numOfRows)++;

55,586✔

1011

END:

55,586✔

1012

  taosRUnLockLatch(&pTopic->lock);

55,586✔

1014

  taosMemoryFreeClear(sql);

55,586✔

1015

  taosMemoryFreeClear(schemaJson);

55,586✔

1016

  return code;

55,586✔

1019

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

20,513✔

1020

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

20,513✔

1023

  SMnode      *pMnode = pReq->info.node;

20,513✔

1024

  SSdb        *pSdb = pMnode->pSdb;

20,513✔

1025

  int32_t      numOfRows = 0;

20,513✔

1026

  SMqTopicObj *pTopic = NULL;

20,513✔

1027

  SUserObj    *pOperUser = NULL;

20,513✔

1028

  int32_t      code = 0, lino = 0;

20,513✔

1029

  char        *sql = NULL;

20,513✔

1030

  char        *schemaJson = NULL;

20,513✔

1031

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

20,513✔

1032

  bool         showAll = false;

20,513✔

1034

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

20,513✔

1035

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

20,513✔

1036

  int32_t objLevel = privObjGetLevel(PRIV_OBJ_TOPIC);

20,513✔

1037

  showAll =

20,513✔

1038

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

20,513✔

1041

  PRINT_LOG_START

20,513✔

1043

  while (numOfRows < rowsCapacity) {

76,537✔

1044

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

76,537✔

1045

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

76,537✔

1047

    if (!showAll) {

56,024✔

1048

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

1,168✔

1049

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

584✔

1050

        sdbRelease(pSdb, pTopic);

438✔

1051

        continue;

438✔

1055

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

55,586✔

1056

    sdbRelease(pSdb, pTopic);

55,586✔

1057

    pTopic = NULL;

55,586✔

1059

  pShow->numOfRows += numOfRows;

20,513✔

1061

END:

20,513✔

1062

  sdbCancelFetch(pSdb, pShow->pIter);

20,513✔

1063

  sdbRelease(pSdb, pTopic);

20,513✔

1064

  mndReleaseUser(pMnode, pOperUser);

20,513✔

1065

  if (code != TSDB_CODE_SUCCESS) {

20,513✔

1069

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

20,513✔

1070

    return numOfRows;

20,513✔

1080

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

762,138✔

1081

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

762,138✔

1084

  SSdb *       pSdb = pMnode->pSdb;

762,138✔

1085

  void *       pIter = NULL;

762,138✔

1086

  SMqTopicObj *pTopic = NULL;

762,138✔

1088

  while (1) {

56,799✔

1089

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

818,937✔

1090

    if (pIter == NULL) {

818,937✔

1091

      break;

761,750✔

1094

    taosRLockLatch(&pTopic->lock);

57,187✔

1095

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

57,187✔

1096

    taosRUnLockLatch(&pTopic->lock);

57,187✔

1098

    if (found) {

57,187✔

1099

      sdbRelease(pSdb, pTopic);

388✔

1100

      sdbCancelFetch(pSdb, pIter);

388✔

1101

      return true;

388✔

1104

    sdbRelease(pSdb, pTopic);

56,799✔

1107

  return false;

761,750✔

taosdata / TDengine / #5032

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

taosdata / TDengine / #5032

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