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

53

int32_t mndInitTopic(SMnode *pMnode) {

487,164✔

54

  SSdbTable table = {

487,164✔

64

  if (pMnode == NULL) {

487,164✔

67

  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CREATE_TOPIC, mndProcessCreateTopicReq);

487,164✔

68

  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_TOPIC, mndProcessDropTopicReq);

487,164✔

70

  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndRetrieveTopic);

487,164✔

71

  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndCancelGetNextTopic);

487,164✔

73

  return sdbSetTable(pMnode->pSdb, table);

487,164✔

76

void mndCleanupTopic(SMnode *pMnode) {}

487,133✔

78

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

161,477✔

79

  if (fullTopic == NULL) {

161,477✔

82

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

161,477✔

83

  if (tmp == NULL) {

161,477✔

86

    tstrncpy(topic, tmp + 1, TSDB_TOPIC_FNAME_LEN);

161,477✔

89

SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {

336,532✔

90

  if (pTopic == NULL) {

336,532✔

93

  int32_t code = 0;

336,532✔

94

  int32_t lino = 0;

336,532✔

95

  terrno = TSDB_CODE_OUT_OF_MEMORY;

336,532✔

97

  void *  swBuf = NULL;

336,532✔

98

  int32_t physicalPlanLen = 0;

336,532✔

99

  if (pTopic->physicalPlan) {

336,532✔

100

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

336,532✔

103

  int32_t schemaLen = 0;

336,532✔

104

  if (pTopic->schema.nCols) {

336,532✔

105

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

518,882✔

108

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

336,532✔

109

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

336,532✔

110

  if (pRaw == NULL) {

336,532✔

114

  int32_t dataPos = 0;

336,532✔

115

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

336,532✔

116

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

336,532✔

117

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

336,532✔

118

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

336,532✔

119

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

336,532✔

120

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

336,532✔

121

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

336,532✔

122

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

336,532✔

123

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

336,532✔

124

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

336,532✔

126

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

336,532✔

127

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

336,532✔

128

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

336,532✔

129

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

336,532✔

130

  SDB_SET_INT32(pRaw, dataPos, physicalPlanLen, TOPIC_ENCODE_OVER);

336,532✔

131

  if (physicalPlanLen) {

336,532✔

132

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

336,532✔

134

  SDB_SET_INT32(pRaw, dataPos, schemaLen, TOPIC_ENCODE_OVER);

336,532✔

135

  if (schemaLen) {

336,532✔

136

    swBuf = taosMemoryMalloc(schemaLen);

259,441✔

137

    if (swBuf == NULL) {

259,441✔

140

    void *aswBuf = swBuf;

259,441✔

141

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

518,882✔

144

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

259,441✔

146

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

336,532✔

147

  SDB_SET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_ENCODE_OVER);

336,532✔

148

  SDB_SET_DATALEN(pRaw, dataPos, TOPIC_ENCODE_OVER);

336,532✔

150

  terrno = TSDB_CODE_SUCCESS;

336,532✔

152

TOPIC_ENCODE_OVER:

336,532✔

153

  if (swBuf) taosMemoryFree(swBuf);

336,532✔

154

  if (terrno != TSDB_CODE_SUCCESS) {

336,532✔

160

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

336,532✔

161

  return pRaw;

336,532✔

164

SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {

285,255✔

165

  if (pRaw == NULL) return NULL;

285,255✔

166

  int32_t code = 0;

285,255✔

167

  int32_t lino = 0;

285,255✔

168

  terrno = TSDB_CODE_OUT_OF_MEMORY;

285,255✔

169

  SSdbRow *    pRow = NULL;

285,255✔

170

  SMqTopicObj *pTopic = NULL;

285,255✔

171

  void *       buf = NULL;

285,255✔

172

  char*        ast = NULL;

285,255✔

174

  int8_t sver = 0;

285,255✔

175

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

285,255✔

177

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

285,255✔

182

  pRow = sdbAllocRow(sizeof(SMqTopicObj));

285,255✔

183

  if (pRow == NULL) goto TOPIC_DECODE_OVER;

285,255✔

185

  pTopic = sdbGetRowObj(pRow);

285,255✔

186

  if (pTopic == NULL) goto TOPIC_DECODE_OVER;

285,255✔

188

  int32_t len = 0;

285,255✔

189

  int32_t dataPos = 0;

285,255✔

190

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

285,255✔

191

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

285,255✔

192

  if (sver >= 2) {

285,255✔

193

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

285,255✔

195

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

285,255✔

196

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

285,255✔

197

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

285,255✔

198

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

285,255✔

199

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

285,255✔

200

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

285,255✔

201

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

285,255✔

203

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

285,255✔

204

  if (sver >= 3) {

285,255✔

205

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

285,255✔

207

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

285,255✔

208

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

285,255✔

209

  if (pTopic->sql == NULL) {

285,255✔

213

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

285,255✔

215

  if (sver < MND_TOPIC_VER_SUPPORT_OWNER) {

285,255✔

231

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

285,255✔

232

    if (len) {

285,255✔

233

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

285,255✔

234

      if (pTopic->physicalPlan == NULL) {

285,255✔

238

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

285,255✔

243

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

285,255✔

244

    if (len) {

285,255✔

245

      buf = taosMemoryMalloc(len);

221,829✔

246

      if (buf == NULL) {

221,829✔

250

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

221,829✔

251

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

443,658✔

255

      pTopic->schema.nCols = 0;

63,426✔

256

      pTopic->schema.version = 0;

63,426✔

257

      pTopic->schema.pSchema = NULL;

63,426✔

259

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

285,255✔

262

  SDB_GET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_DECODE_OVER);

285,255✔

263

  terrno = TSDB_CODE_SUCCESS;

285,255✔

265

TOPIC_DECODE_OVER:

285,255✔

266

  taosMemoryFreeClear(buf);

285,255✔

267

  taosMemoryFreeClear(ast);

285,255✔

269

  if (terrno != TSDB_CODE_SUCCESS) {

285,255✔

275

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

285,255✔

276

  return pRow;

285,255✔

279

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

169,945✔

280

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

169,945✔

281

  return 0;

169,945✔

284

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

285,255✔

285

  if (pTopic == NULL) return 0;

285,255✔

286

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

285,255✔

287

  taosMemoryFreeClear(pTopic->sql);

285,255✔

288

  taosMemoryFreeClear(pTopic->physicalPlan);

285,255✔

289

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

285,255✔

290

  return 0;

285,255✔

293

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

2,240✔

294

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

2,240✔

295

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

2,240✔

296

  taosWLockLatch(&pOldTopic->lock);

2,240✔

297

  SMqTopicObj tmpTopic = *pOldTopic;

2,240✔

298

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

2,240✔

299

  *pNewTopic = tmpTopic;

2,240✔

300

  taosWUnLockLatch(&pOldTopic->lock);

2,240✔

301

  return 0;

2,240✔

304

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

2,484,712✔

305

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

2,484,712✔

308

  SSdb *pSdb = pMnode->pSdb;

2,484,712✔

309

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

2,484,712✔

310

  if (*pTopic == NULL) {

2,484,712✔

311

    return TSDB_CODE_MND_TOPIC_NOT_EXIST;

194,940✔

313

  return TSDB_CODE_SUCCESS;

2,289,772✔

316

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

2,653,150✔

317

  if (pMnode == NULL) return;

2,653,150✔

318

  SSdb *pSdb = pMnode->pSdb;

2,653,150✔

319

  sdbRelease(pSdb, pTopic);

2,653,150✔

322

static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {

151,496✔

323

  if (pCreate == NULL) return TSDB_CODE_INVALID_PARA;

151,496✔

324

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

151,496✔

326

  if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) {

151,496✔

327

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

110,021✔

328

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

41,475✔

329

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

14,013✔

330

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

27,462✔

331

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

27,462✔

334

  return 0;

151,496✔

337

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

147,862✔

338

  SNode *     pAst = NULL;

147,862✔

339

  SQueryPlan *pPlan = NULL;

147,862✔

340

  int32_t     code = TSDB_CODE_SUCCESS;

147,862✔

341

  int32_t     lino = 0;

147,862✔

343

  PRINT_LOG_START

147,862✔

344

  if (ast == NULL) {

147,862✔

345

    topicObj->physicalPlan = taosStrdup("");

33,402✔

346

    goto END;

33,402✔

348

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

114,460✔

349

  MND_TMQ_RETURN_CHECK(nodesStringToNode(ast, &pAst));

114,460✔

350

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

114,460✔

352

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

114,460✔

353

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

114,460✔

354

  if (pPlan == NULL) {

114,460✔

358

  int32_t levelNum = LIST_LENGTH(pPlan->pSubplans);

114,460✔

359

  if (levelNum != 1) {

114,460✔

364

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

114,460✔

365

  MND_TMQ_NULL_CHECK(pNodeListNode);

114,460✔

366

  int32_t opNum = LIST_LENGTH(pNodeListNode->pNodeList);

114,460✔

367

  if (opNum != 1) {

114,460✔

372

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

114,460✔

374

END:

147,862✔

375

  nodesDestroyNode(pAst);

147,862✔

376

  qDestroyQueryPlan(pPlan);

147,862✔

377

  PRINT_LOG_END

147,862✔

378

  return code;

147,862✔

381

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

145,622✔

383

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

145,622✔

385

  STrans *    pTrans = NULL;

145,622✔

386

  int32_t     code = 0;

145,622✔

387

  int32_t     lino = 0;

145,622✔

388

  SMqTopicObj topicObj = {0};

145,622✔

390

  PRINT_LOG_START

145,622✔

391

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

145,622✔

392

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

145,622✔

393

  MND_TMQ_NULL_CHECK(pTrans);

145,622✔

394

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

145,622✔

395

  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

145,622✔

397

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

145,622✔

398

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

145,622✔

399

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

145,622✔

400

  topicObj.ownerId = pOperUser->uid;

145,622✔

403

  if (pDb) {

145,622✔

405

    MND_TMQ_RETURN_CHECK(

145,622✔

409

  topicObj.createTime = taosGetTimestampMs();

145,622✔

410

  topicObj.updateTime = topicObj.createTime;

145,622✔

411

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

145,622✔

412

  topicObj.dbUid = pDb->uid;

145,622✔

413

  topicObj.version = 1;

145,622✔

414

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

145,622✔

415

  MND_TMQ_NULL_CHECK(topicObj.sql);

145,622✔

416

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

145,622✔

417

  topicObj.subType = pCreate->subType;

145,622✔

418

  topicObj.withMeta = pCreate->withMeta;

145,622✔

419

  taosInitRWLatch(&topicObj.lock);

145,622✔

421

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

145,622✔

423

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

145,622✔

424

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

107,899✔

425

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

107,899✔

426

    MND_TMQ_NULL_CHECK(pStb);

107,899✔

427

    char stbName[TSDB_TABLE_NAME_LEN] = {0};

107,189✔

428

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

107,189✔

429

    MND_TMQ_RETURN_CHECK(

107,189✔

431

    MND_TMQ_RETURN_CHECK(

107,189✔

433

    topicObj.stbUid = pStb->uid;

107,189✔

434

    mndReleaseStb(pMnode, pStb);

107,189✔

437

  if (pCreate->subType == TOPIC_SUB_TYPE__DB) {

144,912✔

438

    MND_TMQ_RETURN_CHECK(

26,370✔

440

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

26,370✔

441

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

118,542✔

445

  SSdbRaw *pCommitRaw = mndTopicActionEncode(&topicObj);

144,912✔

446

  MND_TMQ_NULL_CHECK(pCommitRaw);

144,912✔

447

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

144,912✔

448

  if (code != 0) {

144,912✔

453

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));

144,912✔

454

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

144,912✔

456

END:

145,622✔

457

  PRINT_LOG_END

145,622✔

458

  taosMemoryFreeClear(topicObj.sql);

145,622✔

459

  taosMemoryFreeClear(topicObj.physicalPlan);

145,622✔

460

  if (topicObj.schema.nCols) {

145,622✔

461

    taosMemoryFreeClear(topicObj.schema.pSchema);

112,220✔

463

  mndTransDrop(pTrans);

145,622✔

464

  return code;

145,622✔

467

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

2,240✔

469

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

2,240✔

471

  STrans *    pTrans = NULL;

2,240✔

472

  int32_t     code = 0;

2,240✔

473

  int32_t     lino = 0;

2,240✔

474

  SMqTopicObj topicObj = {0};

2,240✔

476

  PRINT_LOG_START

2,240✔

477

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

2,240✔

478

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

2,240✔

479

  MND_TMQ_NULL_CHECK(pTrans);

2,240✔

480

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

2,240✔

481

  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

2,240✔

483

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

2,240✔

484

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

2,240✔

485

  tstrncpy(topicObj.createUser, userName, TSDB_USER_LEN);

2,240✔

487

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

2,240✔

489

  taosRLockLatch(&topicObjOri->lock);

2,240✔

490

  topicObj.createTime = topicObjOri->createTime;

2,240✔

491

  topicObj.updateTime = taosGetTimestampMs();

2,240✔

492

  topicObj.uid = topicObjOri->uid;

2,240✔

493

  topicObj.dbUid = pDb->uid;

2,240✔

494

  topicObj.version = topicObjOri->version + 1;

2,240✔

495

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

2,240✔

496

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

2,240✔

497

  topicObj.subType = pCreate->subType;

2,240✔

498

  topicObj.withMeta = pCreate->withMeta;

2,240✔

499

  taosInitRWLatch(&topicObj.lock);

2,240✔

500

  taosRUnLockLatch(&topicObjOri->lock);

2,240✔

502

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

2,240✔

504

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

2,240✔

505

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

1,280✔

506

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

1,280✔

507

    MND_TMQ_NULL_CHECK(pStb);

1,280✔

508

    topicObj.stbUid = pStb->uid;

1,280✔

509

    mndReleaseStb(pMnode, pStb);

1,280✔

512

  SSdbRaw *pCommitRaw = mndTopicActionEncode(&topicObj);

2,240✔

513

  MND_TMQ_NULL_CHECK(pCommitRaw);

2,240✔

514

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

2,240✔

515

  if (code != 0) {

2,240✔

520

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));

2,240✔

521

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

2,240✔

523

END:

2,240✔

524

  PRINT_LOG_END

2,240✔

525

  taosMemoryFreeClear(topicObj.sql);

2,240✔

526

  taosMemoryFreeClear(topicObj.physicalPlan);

2,240✔

527

  if (topicObj.schema.nCols) {

2,240✔

528

    taosMemoryFreeClear(topicObj.schema.pSchema);

2,240✔

530

  mndTransDrop(pTrans);

2,240✔

531

  return code;

2,240✔

534

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

148,616✔

535

  SMqTopicObj *pTopic = NULL;

148,616✔

536

  SDbObj *     pDb = NULL;

148,616✔

537

  int32_t      code = TSDB_CODE_SUCCESS;

148,616✔

538

  int32_t      lino = 0;

148,616✔

539

  SMnode *     pMnode = pReq->info.node;

148,616✔

540

  int64_t      tss = taosGetTimestampMs();

148,616✔

542

  PRINT_LOG_START

148,616✔

543

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

148,616✔

544

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

148,616✔

545

  if (code == TSDB_CODE_SUCCESS) {

148,616✔

546

    mndReleaseTopic(pMnode, pTopic);

759✔

547

    if (createTopicReq->igExists) {

759✔

548

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

759✔

552

    goto END;

759✔

553

  } else if (code != TSDB_CODE_MND_TOPIC_NOT_EXIST) {

147,857✔

557

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

147,857✔

558

  MND_TMQ_NULL_CHECK(pDb);

147,857✔

560

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

147,147✔

565

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

147,147✔

566

    code = TSDB_CODE_TMQ_TOPIC_OUT_OF_RANGE;

1,525✔

567

    goto END;

1,525✔

570

  MND_TMQ_RETURN_CHECK(grantCheck(TSDB_GRANT_SUBSCRIPTION));

145,622✔

571

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

145,622✔

572

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {

144,912✔

573

    int64_t tse = taosGetTimestampMs();

144,912✔

574

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

144,912✔

575

    duration = duration / 1000;

144,912✔

576

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

144,912✔

577

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

144,912✔

579

  code = TSDB_CODE_ACTION_IN_PROGRESS;

144,912✔

581

END:

148,616✔

582

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

148,616✔

583

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

2,945✔

585

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

145,671✔

587

  mndReleaseDb(pMnode, pDb);

148,616✔

588

  return code;

148,616✔

591

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

2,880✔

592

  SMnode *     pMnode = pReq->info.node;

2,880✔

593

  int32_t      code = TSDB_CODE_SUCCESS;

2,880✔

594

  int32_t      lino = 0;

2,880✔

595

  SDbObj *     pDb = NULL;

2,880✔

596

  SMqTopicObj *pTopic = NULL;

2,880✔

597

  int64_t      tss = taosGetTimestampMs();

2,880✔

599

  PRINT_LOG_START

2,880✔

600

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

2,880✔

601

  if (code != 0) {

2,880✔

602

    if (createTopicReq->igExists) {

640✔

603

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

320✔

604

      code = 0;

320✔

605

      goto END;

320✔

607

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

320✔

608

      goto END;

320✔

612

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

2,240✔

613

  MND_TMQ_NULL_CHECK(pDb);

2,240✔

615

  MND_TMQ_RETURN_CHECK(grantCheck(TSDB_GRANT_SUBSCRIPTION));

2,240✔

616

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

2,240✔

618

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {

2,240✔

619

    int64_t tse = taosGetTimestampMs();

2,240✔

620

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

2,240✔

621

    duration = duration / 1000;

2,240✔

622

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

2,240✔

623

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

2,240✔

626

  code = TSDB_CODE_ACTION_IN_PROGRESS;

2,240✔

628

  if (topicsToReload == NULL) {

2,240✔

629

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

320✔

630

    MND_TMQ_NULL_CHECK(topicsToReload);

320✔

632

  MND_TMQ_RETURN_CHECK(

2,240✔

634

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

2,240✔

636

END:

2,880✔

637

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

2,880✔

638

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

320✔

640

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

2,560✔

642

  mndReleaseTopic(pMnode, pTopic);

2,880✔

643

  mndReleaseDb(pMnode, pDb);

2,880✔

645

  return code;

2,880✔

648

static int32_t mndProcessCreateTopicReq(SRpcMsg *pReq) {

151,496✔

649

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

151,496✔

652

  SMnode *pMnode = pReq->info.node;

151,496✔

653

  SUserObj *pOperUser = NULL;

151,496✔

654

  int32_t code = TSDB_CODE_SUCCESS;

151,496✔

655

  int32_t lino = 0;

151,496✔

657

  SCMCreateTopicReq createTopicReq = {0};

151,496✔

659

  PRINT_LOG_START

151,496✔

660

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

151,496✔

662

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

151,496✔

664

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

151,496✔

666

  MND_TMQ_RETURN_CHECK(mndCheckCreateTopicReq(&createTopicReq));

151,496✔

668

  if (createTopicReq.reload) {

151,496✔

669

    MND_TMQ_RETURN_CHECK(reloadTopic(pReq, &createTopicReq));

2,880✔

671

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

148,616✔

674

END:

151,383✔

675

  tFreeSCMCreateTopicReq(&createTopicReq);

151,496✔

676

  mndReleaseUser(pMnode, pOperUser);

151,496✔

677

  return code;

151,496✔

680

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

112,761✔

681

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

112,761✔

684

  int32_t  code = 0;

112,761✔

685

  int32_t  lino = 0;

112,761✔

686

  SSdbRaw *pCommitRaw = NULL;

112,761✔

687

  PRINT_LOG_START

112,761✔

688

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

112,761✔

689

  mndTopicGetShowName(pTopic->name, topicFName);

112,761✔

690

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

112,761✔

691

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

112,761✔

692

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

112,761✔

693

  pCommitRaw = mndTopicActionEncode(pTopic);

112,761✔

694

  MND_TMQ_NULL_CHECK(pCommitRaw);

112,761✔

695

  code = mndTransAppendCommitlog(pTrans, pCommitRaw);

112,761✔

696

  if (code != 0) {

112,761✔

700

  MND_TMQ_RETURN_CHECK(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));

112,761✔

701

  MND_TMQ_RETURN_CHECK(mndTransPrepare(pMnode, pTrans));

112,761✔

703

END:

112,761✔

704

  PRINT_LOG_END

112,761✔

705

  return code;

112,761✔

708

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

117,423✔

709

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

117,423✔

712

  int32_t sz = taosArrayGetSize(topics);

117,423✔

713

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

118,118✔

714

    char *name = taosArrayGetP(topics, i);

1,801✔

715

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

1,801✔

716

      return true;

1,106✔

719

  return false;

116,317✔

722

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

38,487✔

723

  int32_t         code = 0;

38,487✔

724

  int32_t         lino = 0;

38,487✔

725

  SMqConsumerObj *pConsumerNew = NULL;

38,487✔

727

  taosRLockLatch(&pConsumer->lock);

38,487✔

728

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

38,487✔

729

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

38,487✔

730

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

38,487✔

731

  if (found1 || found2 || found3) {

38,487✔

732

    if (deleteConsumer) {

751✔

733

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

751✔

734

      MND_TMQ_RETURN_CHECK(mndSetConsumerDropLogs(pTrans, pConsumerNew));

751✔

735

      tDeleteSMqConsumerObj(pConsumerNew);

751✔

736

      pConsumerNew = NULL;

751✔

744

END:

38,487✔

745

  taosRUnLockLatch(&pConsumer->lock);

38,487✔

746

  tDeleteSMqConsumerObj(pConsumerNew);

38,487✔

747

  return code;

38,487✔

750

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

112,761✔

751

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

112,761✔

754

  int32_t         code = 0;

112,761✔

755

  int32_t         lino = 0;

112,761✔

756

  SSdb *          pSdb = pMnode->pSdb;

112,761✔

757

  void *          pIter = NULL;

112,761✔

758

  SMqConsumerObj *pConsumer = NULL;

112,761✔

760

  PRINT_LOG_START

112,761✔

762

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

151,248✔

763

    if (pIter == NULL) {

151,248✔

764

      break;

112,761✔

767

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

38,487✔

768

    sdbRelease(pSdb, pConsumer);

38,487✔

771

END:

112,761✔

772

  PRINT_LOG_END

112,761✔

773

  sdbRelease(pSdb, pConsumer);

112,761✔

774

  sdbCancelFetch(pSdb, pIter);

112,761✔

775

  return code;

112,761✔

778

static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {

121,788✔

779

  if (pReq == NULL) {

121,788✔

782

  SMnode *       pMnode = pReq->info.node;

121,788✔

783

  SMDropTopicReq dropReq = {0};

121,788✔

784

  int32_t        code = 0;

121,788✔

785

  int32_t        lino = 0;

121,788✔

786

  SMqTopicObj *  pTopic = NULL;

121,788✔

787

  STrans *       pTrans = NULL;

121,788✔

788

  SUserObj      *pOperUser = NULL;

121,788✔

789

  int64_t        tss = taosGetTimestampMs();

121,788✔

791

  PRINT_LOG_START

121,788✔

792

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

121,788✔

794

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

121,788✔

795

  if (code != 0) {

121,788✔

796

    if (dropReq.igNotExists) {

8,559✔

797

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

8,559✔

798

      code = 0;

8,559✔

800

    goto END;

8,559✔

802

  taosRLockLatch(&pTopic->lock);

113,229✔

804

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

113,229✔

805

  MND_TMQ_NULL_CHECK(pTrans);

113,229✔

807

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

113,229✔

808

  MND_TMQ_RETURN_CHECK(mndTransCheckConflict(pMnode, pTrans));

113,229✔

809

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

113,229✔

811

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

113,229✔

815

  MND_TMQ_RETURN_CHECK(

113,229✔

817

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

113,086✔

819

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

112,761✔

820

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

112,761✔

821

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

112,761✔

822

  if (tsAuditLevel >= AUDIT_LEVEL_DATABASE) {

112,761✔

823

    int64_t tse = taosGetTimestampMs();

112,761✔

824

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

112,761✔

825

    duration = duration / 1000;

112,761✔

826

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

112,761✔

830

  code = TSDB_CODE_ACTION_IN_PROGRESS;

112,761✔

832

END:

121,788✔

833

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

121,788✔

834

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

468✔

836

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

121,320✔

838

  if (pTopic != NULL) {

121,788✔

839

    taosRUnLockLatch(&pTopic->lock);

113,229✔

841

  mndReleaseTopic(pMnode, pTopic);

121,788✔

842

  mndReleaseUser(pMnode, pOperUser);

121,788✔

843

  mndTransDrop(pTrans);

121,788✔

844

  tFreeSMDropTopicReq(&dropReq);

121,788✔

845

  return code;

121,788✔

848

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

124,401✔

849

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

124,401✔

852

  *pNumOfTopics = 0;

124,401✔

854

  SSdb *  pSdb = pMnode->pSdb;

124,401✔

855

  SDbObj *pDb = mndAcquireDb(pMnode, dbName);

124,401✔

856

  if (pDb == NULL) {

124,401✔

860

  int32_t numOfTopics = 0;

124,401✔

861

  void *  pIter = NULL;

124,401✔

862

  while (1) {

263✔

863

    SMqTopicObj *pTopic = NULL;

124,664✔

864

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

124,664✔

865

    if (pIter == NULL) {

124,664✔

866

      break;

124,401✔

868

    taosRLockLatch(&pTopic->lock);

263✔

869

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

263✔

870

      numOfTopics++;

263✔

872

    taosRUnLockLatch(&pTopic->lock);

263✔

874

    sdbRelease(pSdb, pTopic);

263✔

877

  *pNumOfTopics = numOfTopics;

124,401✔

878

  mndReleaseDb(pMnode, pDb);

124,401✔

879

  return 0;

124,401✔

882

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

53,417✔

883

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

53,417✔

886

  char *  string = NULL;

53,417✔

887

  int32_t code = 0;

53,417✔

888

  int32_t lino = 0;

53,417✔

890

  cJSON *cbytes = NULL;

53,417✔

891

  cJSON *ctype = NULL;

53,417✔

892

  cJSON *cname = NULL;

53,417✔

893

  cJSON *column = NULL;

53,417✔

894

  cJSON *columns = cJSON_CreateArray();

53,417✔

895

  MND_TMQ_NULL_CHECK(columns);

53,417✔

896

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

598,207✔

897

    column = cJSON_CreateObject();

544,790✔

898

    MND_TMQ_NULL_CHECK(column);

544,790✔

899

    SSchema *s = schema + i;

544,790✔

900

    cname = cJSON_CreateString(s->name);

544,790✔

901

    MND_TMQ_NULL_CHECK(cname);

544,790✔

902

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

544,790✔

903

    cname = NULL;  // ownership transferred to column object

544,790✔

905

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

544,790✔

906

    MND_TMQ_NULL_CHECK(ctype);

544,790✔

907

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

544,790✔

908

    ctype = NULL;  // ownership transferred to column object

544,790✔

910

    int32_t length = 0;

544,790✔

911

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

544,790✔

912

      length = s->bytes - VARSTR_HEADER_SIZE;

102,172✔

913

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

442,618✔

914

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

68,278✔

916

      length = s->bytes;

374,340✔

918

    cbytes = cJSON_CreateNumber(length);

544,790✔

919

    MND_TMQ_NULL_CHECK(cbytes);

544,790✔

920

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

544,790✔

921

    cbytes = NULL;  // ownership transferred to column object

544,790✔

923

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

544,790✔

924

    column = NULL;  // ownership transferred to columns array

544,790✔

927

END:

53,417✔

928

  string = cJSON_PrintUnformatted(columns);

53,417✔

929

  cJSON_Delete(columns);

53,417✔

930

  cJSON_Delete(column);

53,417✔

931

  cJSON_Delete(cname);

53,417✔

932

  cJSON_Delete(ctype);

53,417✔

933

  cJSON_Delete(cbytes);

53,417✔

935

  size_t len = strlen(string);

53,417✔

936

  if (string && len <= TSDB_SHOW_SCHEMA_JSON_LEN) {

53,417✔

937

    STR_TO_VARSTR(schemaJson, string);

53,417✔

942

  taosMemoryFree(string);

53,417✔

945

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

54,984✔

946

  SColumnInfoData *pColInfo = NULL;

54,984✔

947

  SName            n = {0};

54,984✔

948

  int32_t          cols = 0;

54,984✔

949

  char *           schemaJson = NULL;

54,984✔

950

  char *           sql = NULL;

54,984✔

951

  int32_t          code = 0;

54,984✔

952

  int32_t          lino = 0;

54,984✔

954

  taosRLockLatch(&pTopic->lock);

54,984✔

956

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

54,984✔

957

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

54,984✔

958

  STR_TO_VARSTR(topicName, pName);

54,984✔

960

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

54,984✔

961

  MND_TMQ_NULL_CHECK(pColInfo);

54,984✔

962

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

54,984✔

964

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

54,984✔

965

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

54,984✔

966

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

54,984✔

967

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

54,984✔

969

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

54,984✔

970

  MND_TMQ_NULL_CHECK(pColInfo);

54,984✔

971

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

54,984✔

973

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

54,984✔

974

  MND_TMQ_NULL_CHECK(pColInfo);

54,984✔

975

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

54,984✔

977

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

54,984✔

978

  MND_TMQ_NULL_CHECK(sql);

54,984✔

979

  STR_TO_VARSTR(sql, pTopic->sql);

54,984✔

981

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

54,984✔

982

  MND_TMQ_NULL_CHECK(pColInfo);

54,984✔

983

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

54,984✔

985

  taosMemoryFreeClear(sql);

54,984✔

987

  schemaJson = taosMemoryMalloc(TSDB_SHOW_SCHEMA_JSON_LEN + VARSTR_HEADER_SIZE);

54,984✔

988

  MND_TMQ_NULL_CHECK(schemaJson);

54,984✔

989

  if (pTopic->subType == TOPIC_SUB_TYPE__COLUMN) {

54,984✔

990

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

53,038✔

991

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

1,946✔

992

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

379✔

993

    if (pStb == NULL) {

379✔

997

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

379✔

998

      mndReleaseStb(pMnode, pStb);

379✔

1001

    STR_TO_VARSTR(schemaJson, "NULL");

1,567✔

1004

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

54,984✔

1005

  MND_TMQ_NULL_CHECK(pColInfo);

54,984✔

1006

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

54,984✔

1007

  taosMemoryFreeClear(schemaJson);

54,984✔

1009

  (*numOfRows)++;

54,984✔

1011

END:

54,984✔

1012

  taosRUnLockLatch(&pTopic->lock);

54,984✔

1014

  taosMemoryFreeClear(sql);

54,984✔

1015

  taosMemoryFreeClear(schemaJson);

54,984✔

1016

  return code;

54,984✔

1019

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

20,383✔

1020

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

20,383✔

1023

  SMnode      *pMnode = pReq->info.node;

20,383✔

1024

  SSdb        *pSdb = pMnode->pSdb;

20,383✔

1025

  int32_t      numOfRows = 0;

20,383✔

1026

  SMqTopicObj *pTopic = NULL;

20,383✔

1027

  SUserObj    *pOperUser = NULL;

20,383✔

1028

  int32_t      code = 0, lino = 0;

20,383✔

1029

  char        *sql = NULL;

20,383✔

1030

  char        *schemaJson = NULL;

20,383✔

1031

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

20,383✔

1032

  bool         showAll = false;

20,383✔

1034

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

20,383✔

1035

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

20,383✔

1036

  int32_t objLevel = privObjGetLevel(PRIV_OBJ_TOPIC);

20,383✔

1037

  showAll =

20,383✔

1038

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

20,383✔

1041

  PRINT_LOG_START

20,383✔

1043

  while (numOfRows < rowsCapacity) {

75,796✔

1044

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

75,796✔

1045

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

75,796✔

1047

    if (!showAll) {

55,413✔

1048

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

1,144✔

1049

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

572✔

1050

        sdbRelease(pSdb, pTopic);

429✔

1051

        continue;

429✔

1055

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

54,984✔

1056

    sdbRelease(pSdb, pTopic);

54,984✔

1057

    pTopic = NULL;

54,984✔

1059

  pShow->numOfRows += numOfRows;

20,383✔

1061

END:

20,383✔

1062

  sdbCancelFetch(pSdb, pShow->pIter);

20,383✔

1063

  sdbRelease(pSdb, pTopic);

20,383✔

1064

  mndReleaseUser(pMnode, pOperUser);

20,383✔

1065

  if (code != TSDB_CODE_SUCCESS) {

20,383✔

1069

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

20,383✔

1070

    return numOfRows;

20,383✔

1080

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

767,151✔

1081

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

767,151✔

1084

  SSdb *       pSdb = pMnode->pSdb;

767,151✔

1085

  void *       pIter = NULL;

767,151✔

1086

  SMqTopicObj *pTopic = NULL;

767,151✔

1088

  while (1) {

57,349✔

1089

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

824,500✔

1090

    if (pIter == NULL) {

824,500✔

1091

      break;

766,770✔

1094

    taosRLockLatch(&pTopic->lock);

57,730✔

1095

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

57,730✔

1096

    taosRUnLockLatch(&pTopic->lock);

57,730✔

1098

    if (found) {

57,730✔

1099

      sdbRelease(pSdb, pTopic);

381✔

1100

      sdbCancelFetch(pSdb, pIter);

381✔

1101

      return true;

381✔

1104

    sdbRelease(pSdb, pTopic);

57,349✔

1107

  return false;

766,770✔

taosdata / TDengine / #5042

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

taosdata / TDengine / #5042

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