mndTopic.c 33.8 KB
Newer Older
L
Liu Jicong 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

L
Liu Jicong 已提交
16
#include "mndTopic.h"
L
Liu Jicong 已提交
17
#include "mndConsumer.h"
L
Liu Jicong 已提交
18 19 20
#include "mndDb.h"
#include "mndDnode.h"
#include "mndMnode.h"
21
#include "mndPrivilege.h"
L
Liu Jicong 已提交
22
#include "mndShow.h"
S
Shengliang Guan 已提交
23
#include "mndStb.h"
L
Liu Jicong 已提交
24
#include "mndSubscribe.h"
L
Liu Jicong 已提交
25 26 27
#include "mndTrans.h"
#include "mndUser.h"
#include "mndVgroup.h"
L
Liu Jicong 已提交
28
#include "parser.h"
L
Liu Jicong 已提交
29 30
#include "tname.h"

wmmhello's avatar
wmmhello 已提交
31
#define MND_TOPIC_VER_NUMBER   3
S
Shengliang Guan 已提交
32
#define MND_TOPIC_RESERVE_SIZE 64
L
Liu Jicong 已提交
33

34 35 36
SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic);
SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw);

L
Liu Jicong 已提交
37 38
static int32_t mndTopicActionInsert(SSdb *pSdb, SMqTopicObj *pTopic);
static int32_t mndTopicActionDelete(SSdb *pSdb, SMqTopicObj *pTopic);
39
static int32_t mndTopicActionUpdate(SSdb *pSdb, SMqTopicObj *pOldTopic, SMqTopicObj *pNewTopic);
S
Shengliang Guan 已提交
40 41
static int32_t mndProcessCreateTopicReq(SRpcMsg *pReq);
static int32_t mndProcessDropTopicReq(SRpcMsg *pReq);
L
Liu Jicong 已提交
42

S
Shengliang Guan 已提交
43
static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
L
Liu Jicong 已提交
44
static void    mndCancelGetNextTopic(SMnode *pMnode, void *pIter);
L
Liu Jicong 已提交
45

L
Liu Jicong 已提交
46 47
static int32_t mndSetDropTopicCommitLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic);

L
Liu Jicong 已提交
48
int32_t mndInitTopic(SMnode *pMnode) {
49 50 51 52 53 54 55 56 57
  SSdbTable table = {
      .sdbType = SDB_TOPIC,
      .keyType = SDB_KEY_BINARY,
      .encodeFp = (SdbEncodeFp)mndTopicActionEncode,
      .decodeFp = (SdbDecodeFp)mndTopicActionDecode,
      .insertFp = (SdbInsertFp)mndTopicActionInsert,
      .updateFp = (SdbUpdateFp)mndTopicActionUpdate,
      .deleteFp = (SdbDeleteFp)mndTopicActionDelete,
  };
L
Liu Jicong 已提交
58

L
Liu Jicong 已提交
59 60 61 62
  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CREATE_TOPIC, mndProcessCreateTopicReq);
  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_TOPIC, mndProcessDropTopicReq);
  mndSetMsgHandle(pMnode, TDMT_VND_TMQ_ADD_CHECKINFO_RSP, mndTransProcessRsp);
  mndSetMsgHandle(pMnode, TDMT_VND_TMQ_DEL_CHECKINFO_RSP, mndTransProcessRsp);
L
Liu Jicong 已提交
63

L
Liu Jicong 已提交
64
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndRetrieveTopic);
65
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndCancelGetNextTopic);
S
Shengliang Guan 已提交
66

L
Liu Jicong 已提交
67 68 69 70 71
  return sdbSetTable(pMnode->pSdb, table);
}

void mndCleanupTopic(SMnode *pMnode) {}

L
Liu Jicong 已提交
72 73 74 75 76
const char *mndTopicGetShowName(const char topic[TSDB_TOPIC_FNAME_LEN]) {
  //
  return strchr(topic, '.') + 1;
}

L
Liu Jicong 已提交
77
SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {
L
Liu Jicong 已提交
78 79
  terrno = TSDB_CODE_OUT_OF_MEMORY;

L
Liu Jicong 已提交
80 81 82 83 84
  void   *swBuf = NULL;
  int32_t physicalPlanLen = 0;
  if (pTopic->physicalPlan) {
    physicalPlanLen = strlen(pTopic->physicalPlan) + 1;
  }
85

L
Liu Jicong 已提交
86 87
  int32_t schemaLen = 0;
  if (pTopic->schema.nCols) {
L
Liu Jicong 已提交
88
    schemaLen = taosEncodeSSchemaWrapper(NULL, &pTopic->schema);
L
Liu Jicong 已提交
89
  }
L
Liu Jicong 已提交
90 91 92 93
  int32_t ntbColLen = taosArrayGetSize(pTopic->ntbColIds) * sizeof(int16_t);

  int32_t size = sizeof(SMqTopicObj) + physicalPlanLen + pTopic->sqlLen + pTopic->astLen + schemaLen + ntbColLen +
                 MND_TOPIC_RESERVE_SIZE;
S
Shengliang Guan 已提交
94
  SSdbRaw *pRaw = sdbAllocRaw(SDB_TOPIC, MND_TOPIC_VER_NUMBER, size);
95 96 97
  if (pRaw == NULL) {
    goto TOPIC_ENCODE_OVER;
  }
L
Liu Jicong 已提交
98 99

  int32_t dataPos = 0;
L
Liu Jicong 已提交
100
  SDB_SET_BINARY(pRaw, dataPos, pTopic->name, TSDB_TOPIC_FNAME_LEN, TOPIC_ENCODE_OVER);
L
Liu Jicong 已提交
101
  SDB_SET_BINARY(pRaw, dataPos, pTopic->db, TSDB_DB_FNAME_LEN, TOPIC_ENCODE_OVER);
102
  SDB_SET_BINARY(pRaw, dataPos, pTopic->createUser, TSDB_USER_LEN, TOPIC_ENCODE_OVER);
L
Liu Jicong 已提交
103 104 105 106 107
  SDB_SET_INT64(pRaw, dataPos, pTopic->createTime, TOPIC_ENCODE_OVER);
  SDB_SET_INT64(pRaw, dataPos, pTopic->updateTime, TOPIC_ENCODE_OVER);
  SDB_SET_INT64(pRaw, dataPos, pTopic->uid, TOPIC_ENCODE_OVER);
  SDB_SET_INT64(pRaw, dataPos, pTopic->dbUid, TOPIC_ENCODE_OVER);
  SDB_SET_INT32(pRaw, dataPos, pTopic->version, TOPIC_ENCODE_OVER);
L
Liu Jicong 已提交
108
  SDB_SET_INT8(pRaw, dataPos, pTopic->subType, TOPIC_ENCODE_OVER);
L
Liu Jicong 已提交
109
  SDB_SET_INT8(pRaw, dataPos, pTopic->withMeta, TOPIC_ENCODE_OVER);
L
Liu Jicong 已提交
110

L
Liu Jicong 已提交
111
  SDB_SET_INT64(pRaw, dataPos, pTopic->stbUid, TOPIC_ENCODE_OVER);
112
  SDB_SET_BINARY(pRaw, dataPos, pTopic->stbName, TSDB_TABLE_FNAME_LEN, TOPIC_ENCODE_OVER);
L
Liu Jicong 已提交
113 114
  SDB_SET_INT32(pRaw, dataPos, pTopic->sqlLen, TOPIC_ENCODE_OVER);
  SDB_SET_BINARY(pRaw, dataPos, pTopic->sql, pTopic->sqlLen, TOPIC_ENCODE_OVER);
L
Liu Jicong 已提交
115
  SDB_SET_INT32(pRaw, dataPos, pTopic->astLen, TOPIC_ENCODE_OVER);
116

L
Liu Jicong 已提交
117 118 119
  if (pTopic->astLen) {
    SDB_SET_BINARY(pRaw, dataPos, pTopic->ast, pTopic->astLen, TOPIC_ENCODE_OVER);
  }
L
Liu Jicong 已提交
120
  SDB_SET_INT32(pRaw, dataPos, physicalPlanLen, TOPIC_ENCODE_OVER);
L
Liu Jicong 已提交
121 122
  if (physicalPlanLen) {
    SDB_SET_BINARY(pRaw, dataPos, pTopic->physicalPlan, physicalPlanLen, TOPIC_ENCODE_OVER);
L
Liu Jicong 已提交
123
  }
L
Liu Jicong 已提交
124
  SDB_SET_INT32(pRaw, dataPos, schemaLen, TOPIC_ENCODE_OVER);
L
Liu Jicong 已提交
125 126 127 128 129 130 131 132 133
  if (schemaLen) {
    swBuf = taosMemoryMalloc(schemaLen);
    if (swBuf == NULL) {
      goto TOPIC_ENCODE_OVER;
    }
    void *aswBuf = swBuf;
    taosEncodeSSchemaWrapper(&aswBuf, &pTopic->schema);
    SDB_SET_BINARY(pRaw, dataPos, swBuf, schemaLen, TOPIC_ENCODE_OVER);
  }
134

L
Liu Jicong 已提交
135 136 137 138 139 140 141 142 143
  SDB_SET_INT64(pRaw, dataPos, pTopic->ntbUid, TOPIC_ENCODE_OVER);
  if (pTopic->ntbUid != 0) {
    int32_t sz = taosArrayGetSize(pTopic->ntbColIds);
    SDB_SET_INT32(pRaw, dataPos, sz, TOPIC_ENCODE_OVER);
    for (int32_t i = 0; i < sz; i++) {
      int16_t colId = *(int16_t *)taosArrayGet(pTopic->ntbColIds, i);
      SDB_SET_INT16(pRaw, dataPos, colId, TOPIC_ENCODE_OVER);
    }
  }
144

L
Liu Jicong 已提交
145
  SDB_SET_INT64(pRaw, dataPos, pTopic->ctbStbUid, TOPIC_ENCODE_OVER);
L
Liu Jicong 已提交
146

L
Liu Jicong 已提交
147 148 149 150 151 152
  SDB_SET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_ENCODE_OVER);
  SDB_SET_DATALEN(pRaw, dataPos, TOPIC_ENCODE_OVER);

  terrno = TSDB_CODE_SUCCESS;

TOPIC_ENCODE_OVER:
L
Liu Jicong 已提交
153
  if (swBuf) taosMemoryFree(swBuf);
L
Liu Jicong 已提交
154 155 156 157 158 159 160
  if (terrno != TSDB_CODE_SUCCESS) {
    mError("topic:%s, failed to encode to raw:%p since %s", pTopic->name, pRaw, terrstr());
    sdbFreeRaw(pRaw);
    return NULL;
  }

  mTrace("topic:%s, encode to raw:%p, row:%p", pTopic->name, pRaw, pTopic);
L
Liu Jicong 已提交
161 162 163
  return pRaw;
}

L
Liu Jicong 已提交
164
SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {
165
  terrno = TSDB_CODE_OUT_OF_MEMORY;
166 167 168
  SSdbRow     *pRow = NULL;
  SMqTopicObj *pTopic = NULL;
  void        *buf = NULL;
169

L
Liu Jicong 已提交
170
  int8_t sver = 0;
171
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto TOPIC_DECODE_OVER;
L
Liu Jicong 已提交
172

wmmhello's avatar
wmmhello 已提交
173
  if (sver < 1 || sver > MND_TOPIC_VER_NUMBER) {
L
Liu Jicong 已提交
174
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
175
    goto TOPIC_DECODE_OVER;
L
Liu Jicong 已提交
176 177
  }

178
  pRow = sdbAllocRow(sizeof(SMqTopicObj));
179 180
  if (pRow == NULL) goto TOPIC_DECODE_OVER;

181
  pTopic = sdbGetRowObj(pRow);
182
  if (pTopic == NULL) goto TOPIC_DECODE_OVER;
L
Liu Jicong 已提交
183

L
Liu Jicong 已提交
184
  int32_t len;
L
Liu Jicong 已提交
185
  int32_t dataPos = 0;
L
Liu Jicong 已提交
186
  SDB_GET_BINARY(pRaw, dataPos, pTopic->name, TSDB_TOPIC_FNAME_LEN, TOPIC_DECODE_OVER);
L
Liu Jicong 已提交
187
  SDB_GET_BINARY(pRaw, dataPos, pTopic->db, TSDB_DB_FNAME_LEN, TOPIC_DECODE_OVER);
188 189 190
  if (sver >= 2) {
    SDB_GET_BINARY(pRaw, dataPos, pTopic->createUser, TSDB_USER_LEN, TOPIC_DECODE_OVER);
  }
L
Liu Jicong 已提交
191 192 193 194 195
  SDB_GET_INT64(pRaw, dataPos, &pTopic->createTime, TOPIC_DECODE_OVER);
  SDB_GET_INT64(pRaw, dataPos, &pTopic->updateTime, TOPIC_DECODE_OVER);
  SDB_GET_INT64(pRaw, dataPos, &pTopic->uid, TOPIC_DECODE_OVER);
  SDB_GET_INT64(pRaw, dataPos, &pTopic->dbUid, TOPIC_DECODE_OVER);
  SDB_GET_INT32(pRaw, dataPos, &pTopic->version, TOPIC_DECODE_OVER);
L
Liu Jicong 已提交
196
  SDB_GET_INT8(pRaw, dataPos, &pTopic->subType, TOPIC_DECODE_OVER);
L
Liu Jicong 已提交
197
  SDB_GET_INT8(pRaw, dataPos, &pTopic->withMeta, TOPIC_DECODE_OVER);
L
Liu Jicong 已提交
198

L
Liu Jicong 已提交
199
  SDB_GET_INT64(pRaw, dataPos, &pTopic->stbUid, TOPIC_DECODE_OVER);
wmmhello's avatar
wmmhello 已提交
200 201 202
  if (sver >= 3) {
    SDB_GET_BINARY(pRaw, dataPos, pTopic->stbName, TSDB_TABLE_FNAME_LEN, TOPIC_DECODE_OVER);
  }
L
Liu Jicong 已提交
203 204 205 206 207 208
  SDB_GET_INT32(pRaw, dataPos, &pTopic->sqlLen, TOPIC_DECODE_OVER);
  pTopic->sql = taosMemoryCalloc(pTopic->sqlLen, sizeof(char));
  if (pTopic->sql == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    goto TOPIC_DECODE_OVER;
  }
L
Liu Jicong 已提交
209
  SDB_GET_BINARY(pRaw, dataPos, pTopic->sql, pTopic->sqlLen, TOPIC_DECODE_OVER);
210

L
Liu Jicong 已提交
211
  SDB_GET_INT32(pRaw, dataPos, &pTopic->astLen, TOPIC_DECODE_OVER);
L
Liu Jicong 已提交
212 213 214 215 216 217 218 219
  if (pTopic->astLen) {
    pTopic->ast = taosMemoryCalloc(pTopic->astLen, sizeof(char));
    if (pTopic->ast == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      goto TOPIC_DECODE_OVER;
    }
  } else {
    pTopic->ast = NULL;
220
  }
L
Liu Jicong 已提交
221
  SDB_GET_BINARY(pRaw, dataPos, pTopic->ast, pTopic->astLen, TOPIC_DECODE_OVER);
222
  SDB_GET_INT32(pRaw, dataPos, &len, TOPIC_DECODE_OVER);
L
Liu Jicong 已提交
223 224 225 226 227 228 229 230 231
  if (len) {
    pTopic->physicalPlan = taosMemoryCalloc(len, sizeof(char));
    if (pTopic->physicalPlan == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      goto TOPIC_DECODE_OVER;
    }
    SDB_GET_BINARY(pRaw, dataPos, pTopic->physicalPlan, len, TOPIC_DECODE_OVER);
  } else {
    pTopic->physicalPlan = NULL;
232
  }
233

L
Liu Jicong 已提交
234
  SDB_GET_INT32(pRaw, dataPos, &len, TOPIC_DECODE_OVER);
L
Liu Jicong 已提交
235
  if (len) {
236
    buf = taosMemoryMalloc(len);
L
Liu Jicong 已提交
237 238 239 240 241 242 243 244 245 246
    if (buf == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      goto TOPIC_DECODE_OVER;
    }
    SDB_GET_BINARY(pRaw, dataPos, buf, len, TOPIC_DECODE_OVER);
    if (taosDecodeSSchemaWrapper(buf, &pTopic->schema) == NULL) {
      goto TOPIC_DECODE_OVER;
    }
  } else {
    pTopic->schema.nCols = 0;
247
    pTopic->schema.version = 0;
L
Liu Jicong 已提交
248
    pTopic->schema.pSchema = NULL;
L
Liu Jicong 已提交
249
  }
L
Liu Jicong 已提交
250 251 252 253 254 255 256 257 258 259 260 261 262
  SDB_GET_INT64(pRaw, dataPos, &pTopic->ntbUid, TOPIC_DECODE_OVER);
  if (pTopic->ntbUid != 0) {
    int32_t ntbColNum;
    SDB_GET_INT32(pRaw, dataPos, &ntbColNum, TOPIC_DECODE_OVER);
    pTopic->ntbColIds = taosArrayInit(ntbColNum, sizeof(int16_t));
    if (pTopic->ntbColIds == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      goto TOPIC_DECODE_OVER;
    }
    int16_t colId;
    SDB_GET_INT16(pRaw, dataPos, &colId, TOPIC_DECODE_OVER);
    taosArrayPush(pTopic->ntbColIds, &colId);
  }
L
Liu Jicong 已提交
263

264
  SDB_GET_INT64(pRaw, dataPos, &pTopic->ctbStbUid, TOPIC_DECODE_OVER);
L
Liu Jicong 已提交
265
  SDB_GET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_DECODE_OVER);
L
Liu Jicong 已提交
266
  terrno = TSDB_CODE_SUCCESS;
267 268

TOPIC_DECODE_OVER:
269
  taosMemoryFreeClear(buf);
L
Liu Jicong 已提交
270
  if (terrno != TSDB_CODE_SUCCESS) {
271
    mError("topic:%s, failed to decode from raw:%p since %s", pTopic == NULL ? "null" : pTopic->name, pRaw, terrstr());
wafwerar's avatar
wafwerar 已提交
272
    taosMemoryFreeClear(pRow);
273 274
    return NULL;
  }
L
Liu Jicong 已提交
275

276
  mTrace("topic:%s, decode from raw:%p, row:%p", pTopic->name, pRaw, pTopic);
L
Liu Jicong 已提交
277 278 279
  return pRow;
}

L
Liu Jicong 已提交
280
static int32_t mndTopicActionInsert(SSdb *pSdb, SMqTopicObj *pTopic) {
281
  mTrace("topic:%s perform insert action", pTopic->name);
L
Liu Jicong 已提交
282 283 284
  return 0;
}

L
Liu Jicong 已提交
285
static int32_t mndTopicActionDelete(SSdb *pSdb, SMqTopicObj *pTopic) {
286
  mTrace("topic:%s perform delete action", pTopic->name);
L
Liu Jicong 已提交
287 288 289 290 291
  taosMemoryFreeClear(pTopic->sql);
  taosMemoryFreeClear(pTopic->ast);
  taosMemoryFreeClear(pTopic->physicalPlan);
  if (pTopic->schema.nCols) taosMemoryFreeClear(pTopic->schema.pSchema);
  taosArrayDestroy(pTopic->ntbColIds);
L
Liu Jicong 已提交
292 293 294
  return 0;
}

L
Liu Jicong 已提交
295
static int32_t mndTopicActionUpdate(SSdb *pSdb, SMqTopicObj *pOldTopic, SMqTopicObj *pNewTopic) {
296
  mTrace("topic:%s perform update action", pOldTopic->name);
wafwerar's avatar
wafwerar 已提交
297
  atomic_exchange_64(&pOldTopic->updateTime, pNewTopic->updateTime);
L
Liu Jicong 已提交
298 299
  atomic_exchange_32(&pOldTopic->version, pNewTopic->version);

L
Liu Jicong 已提交
300
  /*taosWLockLatch(&pOldTopic->lock);*/
L
Liu Jicong 已提交
301 302

  // TODO handle update
L
Liu Jicong 已提交
303

L
Liu Jicong 已提交
304
  /*taosWUnLockLatch(&pOldTopic->lock);*/
L
Liu Jicong 已提交
305 306 307
  return 0;
}

S
Shengliang Guan 已提交
308
SMqTopicObj *mndAcquireTopic(SMnode *pMnode, const char *topicName) {
L
Liu Jicong 已提交
309 310
  SSdb        *pSdb = pMnode->pSdb;
  SMqTopicObj *pTopic = sdbAcquire(pSdb, SDB_TOPIC, topicName);
S
Shengliang Guan 已提交
311
  if (pTopic == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
L
Liu Jicong 已提交
312 313 314 315 316
    terrno = TSDB_CODE_MND_TOPIC_NOT_EXIST;
  }
  return pTopic;
}

L
Liu Jicong 已提交
317
void mndReleaseTopic(SMnode *pMnode, SMqTopicObj *pTopic) {
L
Liu Jicong 已提交
318 319 320 321
  SSdb *pSdb = pMnode->pSdb;
  sdbRelease(pSdb, pTopic);
}

S
Shengliang Guan 已提交
322 323
static SDDropTopicReq *mndBuildDropTopicMsg(SMnode *pMnode, SVgObj *pVgroup, SMqTopicObj *pTopic) {
  int32_t contLen = sizeof(SDDropTopicReq);
L
Liu Jicong 已提交
324

wafwerar's avatar
wafwerar 已提交
325
  SDDropTopicReq *pDrop = taosMemoryCalloc(1, contLen);
L
Liu Jicong 已提交
326 327 328 329 330 331 332
  if (pDrop == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

  pDrop->head.contLen = htonl(contLen);
  pDrop->head.vgId = htonl(pVgroup->vgId);
L
Liu Jicong 已提交
333
  memcpy(pDrop->name, pTopic->name, TSDB_TOPIC_FNAME_LEN);
L
Liu Jicong 已提交
334 335 336 337 338
  pDrop->tuid = htobe64(pTopic->uid);

  return pDrop;
}

L
Liu Jicong 已提交
339
static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {
L
Liu Jicong 已提交
340 341 342 343 344 345 346 347 348 349
  terrno = TSDB_CODE_MND_INVALID_TOPIC;

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

  if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) {
    if (pCreate->ast == NULL || pCreate->ast[0] == 0) return -1;
  } else if (pCreate->subType == TOPIC_SUB_TYPE__TABLE) {
    if (pCreate->subStbName[0] == 0) return -1;
  } else if (pCreate->subType == TOPIC_SUB_TYPE__DB) {
    if (pCreate->subDbName[0] == 0) return -1;
S
Shengliang Guan 已提交
350
  }
351

L
Liu Jicong 已提交
352
  terrno = TSDB_CODE_SUCCESS;
L
Liu Jicong 已提交
353
  return 0;
354 355
}

L
Liu Jicong 已提交
356
static int32_t extractTopicTbInfo(SNode *pAst, SMqTopicObj *pTopic) {
L
Liu Jicong 已提交
357 358
  SNodeList *pNodeList = NULL;
  nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList);
L
Liu Jicong 已提交
359 360 361 362 363 364 365 366 367 368 369 370 371
  int64_t suid = ((SRealTableNode *)((SSelectStmt *)pAst)->pFromTable)->pMeta->suid;
  int8_t  tableType = ((SRealTableNode *)((SSelectStmt *)pAst)->pFromTable)->pMeta->tableType;
  if (tableType == TSDB_CHILD_TABLE) {
    pTopic->ctbStbUid = suid;
  } else if (tableType == TSDB_NORMAL_TABLE) {
    SNode *pNode = NULL;
    FOREACH(pNode, pNodeList) {
      SColumnNode *pCol = (SColumnNode *)pNode;
      if (pCol->tableType == TSDB_NORMAL_TABLE) {
        pTopic->ntbUid = pCol->tableId;
        taosArrayPush(pTopic->ntbColIds, &pCol->colId);
      }
    }
L
Liu Jicong 已提交
372
  }
L
Liu Jicong 已提交
373
  nodesDestroyList(pNodeList);
L
Liu Jicong 已提交
374 375 376
  return 0;
}

377 378
static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *pCreate, SDbObj *pDb,
                              const char *userName) {
379
  mInfo("start to create topic:%s", pCreate->name);
380 381 382 383
  STrans *pTrans = NULL;
  int32_t code = -1;
  SNode  *pAst = NULL;
  SQueryPlan *pPlan = NULL;
384

L
Liu Jicong 已提交
385
  SMqTopicObj topicObj = {0};
386
  tstrncpy(topicObj.name, pCreate->name, TSDB_TOPIC_FNAME_LEN);
387
  tstrncpy(topicObj.db, pDb->name, TSDB_DB_FNAME_LEN);
388
  tstrncpy(topicObj.createUser, userName, TSDB_USER_LEN);
389 390 391 392 393

  if (mndCheckTopicPrivilege(pMnode, pReq->info.conn.user, MND_OPER_CREATE_TOPIC, &topicObj) != 0) {
    return -1;
  }

L
Liu Jicong 已提交
394 395
  topicObj.createTime = taosGetTimestampMs();
  topicObj.updateTime = topicObj.createTime;
L
Liu Jicong 已提交
396
  topicObj.uid = mndGenerateUid(pCreate->name, strlen(pCreate->name));
L
Liu Jicong 已提交
397 398
  topicObj.dbUid = pDb->uid;
  topicObj.version = 1;
399
  topicObj.sql = taosStrdup(pCreate->sql);
L
Liu Jicong 已提交
400
  topicObj.sqlLen = strlen(pCreate->sql) + 1;
L
Liu Jicong 已提交
401
  topicObj.subType = pCreate->subType;
L
Liu Jicong 已提交
402
  topicObj.withMeta = pCreate->withMeta;
403

L
Liu Jicong 已提交
404
  if (pCreate->subType == TOPIC_SUB_TYPE__COLUMN) {
405
    if (pCreate->withMeta) {
S
Shengliang Guan 已提交
406 407
      terrno = TSDB_CODE_MND_INVALID_TOPIC_OPTION;
      mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
408
      goto _OUT;
S
Shengliang Guan 已提交
409
    }
L
Liu Jicong 已提交
410

411
    topicObj.ast = taosStrdup(pCreate->ast);
L
Liu Jicong 已提交
412
    topicObj.astLen = strlen(pCreate->ast) + 1;
L
Liu Jicong 已提交
413

414
    qDebugL("topic:%s ast %s", topicObj.name, topicObj.ast);
L
Liu Jicong 已提交
415

L
Liu Jicong 已提交
416 417
    if (nodesStringToNode(pCreate->ast, &pAst) != 0) {
      mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
418
      goto _OUT;
L
Liu Jicong 已提交
419
    }
L
Liu Jicong 已提交
420

L
Liu Jicong 已提交
421 422
    SPlanContext cxt = {.pAstRoot = pAst, .topicQuery = true};
    if (qCreateQueryPlan(&cxt, &pPlan, NULL) != 0) {
423
      mError("failed to create topic:%s since %s", pCreate->name, terrstr());
424
      goto _OUT;
L
Liu Jicong 已提交
425 426
    }

L
Liu Jicong 已提交
427 428
    topicObj.ntbColIds = taosArrayInit(0, sizeof(int16_t));
    if (topicObj.ntbColIds == NULL) {
L
Liu Jicong 已提交
429
      terrno = TSDB_CODE_OUT_OF_MEMORY;
430
      goto _OUT;
L
Liu Jicong 已提交
431
    }
432

L
Liu Jicong 已提交
433 434 435 436 437
    extractTopicTbInfo(pAst, &topicObj);

    if (topicObj.ntbUid == 0) {
      taosArrayDestroy(topicObj.ntbColIds);
      topicObj.ntbColIds = NULL;
L
Liu Jicong 已提交
438 439
    }

L
Liu Jicong 已提交
440 441
    if (qExtractResultSchema(pAst, &topicObj.schema.nCols, &topicObj.schema.pSchema) != 0) {
      mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
442
      goto _OUT;
L
Liu Jicong 已提交
443 444
    }

445
    if (nodesNodeToString((SNode *)pPlan, false, &topicObj.physicalPlan, NULL) != 0) {
L
Liu Jicong 已提交
446
      mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
447
      goto _OUT;
L
Liu Jicong 已提交
448
    }
L
Liu Jicong 已提交
449
  } else if (pCreate->subType == TOPIC_SUB_TYPE__TABLE) {
L
Liu Jicong 已提交
450
    SStbObj *pStb = mndAcquireStb(pMnode, pCreate->subStbName);
451 452
    if (pStb == NULL) {
      terrno = TSDB_CODE_MND_STB_NOT_EXIST;
453
      goto _OUT;
454
    }
455

456
    strcpy(topicObj.stbName, pCreate->subStbName);
L
Liu Jicong 已提交
457
    topicObj.stbUid = pStb->uid;
S
Shengliang Guan 已提交
458
    mndReleaseStb(pMnode, pStb);
459 460 461 462 463
    if(pCreate->ast != NULL){
      qDebugL("topic:%s ast %s", topicObj.name, pCreate->ast);
      topicObj.ast = taosStrdup(pCreate->ast);
      topicObj.astLen = strlen(pCreate->ast) + 1;
    }
L
Liu Jicong 已提交
464
  }
L
Liu Jicong 已提交
465 466 467 468 469 470
  /*} else if (pCreate->subType == TOPIC_SUB_TYPE__DB) {*/
  /*topicObj.ast = NULL;*/
  /*topicObj.astLen = 0;*/
  /*topicObj.physicalPlan = NULL;*/
  /*topicObj.withTbName = 1;*/
  /*topicObj.withSchema = 1;*/
L
Liu Jicong 已提交
471

472
  pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pReq, "create-topic");
S
Shengliang Guan 已提交
473 474
  if (pTrans == NULL) {
    mError("topic:%s, failed to create since %s", pCreate->name, terrstr());
475
    goto _OUT;
S
Shengliang Guan 已提交
476
  }
477

478
  mndTransSetDbName(pTrans, pDb->name, NULL);
479 480 481
  if (mndTransCheckConflict(pMnode, pTrans) != 0) {
      goto _OUT;
  }
482
  mInfo("trans:%d to create topic:%s", pTrans->id, pCreate->name);
S
Shengliang Guan 已提交
483

484 485 486
  SSdbRaw *pCommitRaw = mndTopicActionEncode(&topicObj);
  if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
    mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
487
    goto _OUT;
S
Shengliang Guan 已提交
488
  }
489

S
Shengliang Guan 已提交
490
  (void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
S
Shengliang Guan 已提交
491

L
Liu Jicong 已提交
492
  if (topicObj.ntbUid != 0) {
493
    STqCheckInfo info;
L
Liu Jicong 已提交
494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512
    memcpy(info.topic, topicObj.name, TSDB_TOPIC_FNAME_LEN);
    info.ntbUid = topicObj.ntbUid;
    info.colIdList = topicObj.ntbColIds;
    // broadcast forbid alter info
    void   *pIter = NULL;
    SSdb   *pSdb = pMnode->pSdb;
    SVgObj *pVgroup = NULL;
    while (1) {
      // iterate vg
      pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
      if (pIter == NULL) break;
      if (!mndVgroupInDb(pVgroup, topicObj.dbUid)) {
        sdbRelease(pSdb, pVgroup);
        continue;
      }

      // encoder check alter info
      int32_t len;
      int32_t code;
513
      tEncodeSize(tEncodeSTqCheckInfo, &info, len, code);
L
Liu Jicong 已提交
514 515
      if (code < 0) {
        sdbRelease(pSdb, pVgroup);
516
        sdbCancelFetch(pSdb, pIter);
517
        goto _OUT;
L
Liu Jicong 已提交
518 519 520 521 522
      }
      void    *buf = taosMemoryCalloc(1, sizeof(SMsgHead) + len);
      void    *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
      SEncoder encoder;
      tEncoderInit(&encoder, abuf, len);
523
      if (tEncodeSTqCheckInfo(&encoder, &info) < 0) {
524
        taosMemoryFree(buf);
L
Liu Jicong 已提交
525
        sdbRelease(pSdb, pVgroup);
526
        sdbCancelFetch(pSdb, pIter);
527
        goto _OUT;
L
Liu Jicong 已提交
528 529
      }
      tEncoderClear(&encoder);
L
Liu Jicong 已提交
530
      ((SMsgHead *)buf)->vgId = htonl(pVgroup->vgId);
L
Liu Jicong 已提交
531 532 533 534 535
      // add redo action
      STransAction action = {0};
      action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
      action.pCont = buf;
      action.contLen = sizeof(SMsgHead) + len;
L
Liu Jicong 已提交
536
      action.msgType = TDMT_VND_TMQ_ADD_CHECKINFO;
L
Liu Jicong 已提交
537 538 539
      if (mndTransAppendRedoAction(pTrans, &action) != 0) {
        taosMemoryFree(buf);
        sdbRelease(pSdb, pVgroup);
540
        sdbCancelFetch(pSdb, pIter);
541
        goto _OUT;
L
Liu Jicong 已提交
542
      }
543
      buf = NULL;
S
Shengliang Guan 已提交
544
      sdbRelease(pSdb, pVgroup);
L
Liu Jicong 已提交
545 546 547
    }
  }

S
Shengliang Guan 已提交
548 549
  if (mndTransPrepare(pMnode, pTrans) != 0) {
    mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
550
    goto _OUT;
S
Shengliang Guan 已提交
551 552
  }

553 554 555
  code = TSDB_CODE_ACTION_IN_PROGRESS;

_OUT:
L
Liu Jicong 已提交
556
  taosMemoryFreeClear(topicObj.physicalPlan);
L
Liu Jicong 已提交
557 558 559
  taosMemoryFreeClear(topicObj.sql);
  taosMemoryFreeClear(topicObj.ast);
  taosArrayDestroy(topicObj.ntbColIds);
560 561 562
  if (topicObj.schema.nCols) {
    taosMemoryFreeClear(topicObj.schema.pSchema);
  }
563 564
  nodesDestroyNode(pAst);
  nodesDestroyNode((SNode *)pPlan);
S
Shengliang Guan 已提交
565
  mndTransDrop(pTrans);
566
  return code;
L
Liu Jicong 已提交
567 568
}

S
Shengliang Guan 已提交
569 570
static int32_t mndProcessCreateTopicReq(SRpcMsg *pReq) {
  SMnode           *pMnode = pReq->info.node;
L
Liu Jicong 已提交
571 572 573 574
  int32_t           code = -1;
  SMqTopicObj      *pTopic = NULL;
  SDbObj           *pDb = NULL;
  SCMCreateTopicReq createTopicReq = {0};
wmmhello's avatar
wmmhello 已提交
575 576 577 578 579
  if (sdbGetSize(pMnode->pSdb, SDB_TOPIC) >= tmqMaxTopicNum){
    terrno = TSDB_CODE_TMQ_TOPIC_OUT_OF_RANGE;
    mError("topic num out of range");
    return code;
  }
L
Liu Jicong 已提交
580

S
Shengliang Guan 已提交
581
  if (tDeserializeSCMCreateTopicReq(pReq->pCont, pReq->contLen, &createTopicReq) != 0) {
S
Shengliang Guan 已提交
582
    terrno = TSDB_CODE_INVALID_MSG;
583
    goto _OVER;
S
Shengliang Guan 已提交
584
  }
585

H
Haojun Liao 已提交
586
  mInfo("topic:%s start to create, sql:%s", createTopicReq.name, createTopicReq.sql);
L
Liu Jicong 已提交
587

S
Shengliang Guan 已提交
588
  if (mndCheckCreateTopicReq(&createTopicReq) != 0) {
H
Haojun Liao 已提交
589
    mError("topic:%s failed to create since %s", createTopicReq.name, terrstr());
590
    goto _OVER;
L
Liu Jicong 已提交
591 592
  }

S
Shengliang Guan 已提交
593
  pTopic = mndAcquireTopic(pMnode, createTopicReq.name);
L
Liu Jicong 已提交
594
  if (pTopic != NULL) {
595
    if (createTopicReq.igExists) {
H
Haojun Liao 已提交
596
      mInfo("topic:%s already exist, ignore exist is set", createTopicReq.name);
S
Shengliang Guan 已提交
597
      code = 0;
598
      goto _OVER;
L
Liu Jicong 已提交
599 600
    } else {
      terrno = TSDB_CODE_MND_TOPIC_ALREADY_EXIST;
601
      goto _OVER;
L
Liu Jicong 已提交
602
    }
S
Shengliang Guan 已提交
603
  } else if (terrno != TSDB_CODE_MND_TOPIC_NOT_EXIST) {
604
    goto _OVER;
L
Liu Jicong 已提交
605 606
  }

L
Liu Jicong 已提交
607
  pDb = mndAcquireDb(pMnode, createTopicReq.subDbName);
L
Liu Jicong 已提交
608 609
  if (pDb == NULL) {
    terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
610
    goto _OVER;
S
Shengliang Guan 已提交
611 612
  }

613
  if (pDb->cfg.walRetentionPeriod == 0) {
614
    terrno = TSDB_CODE_MND_DB_RETENTION_PERIOD_ZERO;
615 616 617 618
    mError("db:%s, not allowed to create topic when WAL_RETENTION_PERIOD is zero", pDb->name);
    goto _OVER;
  }

619
  code = mndCreateTopic(pMnode, pReq, &createTopicReq, pDb, pReq->info.conn.user);
620 621 622
  if (code == 0) {
    code = TSDB_CODE_ACTION_IN_PROGRESS;
  }
S
Shengliang Guan 已提交
623

624
_OVER:
S
Shengliang Guan 已提交
625
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
626
    mError("failed to create topic:%s since %s", createTopicReq.name, terrstr());
L
Liu Jicong 已提交
627 628
  }

S
Shengliang Guan 已提交
629
  mndReleaseTopic(pMnode, pTopic);
L
Liu Jicong 已提交
630 631
  mndReleaseDb(pMnode, pDb);

L
Liu Jicong 已提交
632
  tFreeSCMCreateTopicReq(&createTopicReq);
S
Shengliang Guan 已提交
633 634 635
  return code;
}

636
static int32_t mndDropTopic(SMnode *pMnode, STrans *pTrans, SRpcMsg *pReq, SMqTopicObj *pTopic) {
637
  int32_t code = -1;
638 639 640
  if (mndUserRemoveTopic(pMnode, pTrans, pTopic->name) != 0) {
    goto _OVER;
  }
641

642
  SSdbRaw *pCommitRaw = mndTopicActionEncode(pTopic);
643
  if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) goto _OVER;
S
Shengliang Guan 已提交
644
  (void)sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED);
S
Shengliang Guan 已提交
645

646 647 648 649
  if (mndTransPrepare(pMnode, pTrans) != 0) {
    goto _OVER;
  }

650
  code = 0;
L
Liu Jicong 已提交
651

652 653
_OVER:
  return code;
L
Liu Jicong 已提交
654 655
}

S
Shengliang Guan 已提交
656
static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
L
Liu Jicong 已提交
657 658
  SMnode        *pMnode = pReq->info.node;
  SSdb          *pSdb = pMnode->pSdb;
S
Shengliang Guan 已提交
659
  SMDropTopicReq dropReq = {0};
L
Liu Jicong 已提交
660

S
Shengliang Guan 已提交
661
  if (tDeserializeSMDropTopicReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
S
Shengliang Guan 已提交
662
    terrno = TSDB_CODE_INVALID_MSG;
663
    return -1;
L
Liu Jicong 已提交
664 665
  }

666
  SMqTopicObj *pTopic = mndAcquireTopic(pMnode, dropReq.name);
L
Liu Jicong 已提交
667 668
  if (pTopic == NULL) {
    if (dropReq.igNotExists) {
669
      mInfo("topic:%s, not exist, ignore not exist is set", dropReq.name);
L
Liu Jicong 已提交
670 671 672 673 674 675 676
      return 0;
    } else {
      terrno = TSDB_CODE_MND_TOPIC_NOT_EXIST;
      mError("topic:%s, failed to drop since %s", dropReq.name, terrstr());
      return -1;
    }
  }
S
Shengliang Guan 已提交
677

678
  if (mndCheckTopicPrivilege(pMnode, pReq->info.conn.user, MND_OPER_DROP_TOPIC, pTopic) != 0) {
S
Shengliang Guan 已提交
679 680 681 682
    mndReleaseTopic(pMnode, pTopic);
    return -1;
  }

L
Liu Jicong 已提交
683 684 685 686
  void           *pIter = NULL;
  SMqConsumerObj *pConsumer;
  while (1) {
    pIter = sdbFetch(pSdb, SDB_CONSUMER, pIter, (void **)&pConsumer);
687 688 689
    if (pIter == NULL) {
      break;
    }
L
Liu Jicong 已提交
690

691 692
    if (pConsumer->status == MQ_CONSUMER_STATUS_LOST){
      mndDropConsumerFromSdb(pMnode, pConsumer->consumerId);
wmmhello's avatar
wmmhello 已提交
693
      mndReleaseConsumer(pMnode, pConsumer);
694 695
      continue;
    }
696

L
Liu Jicong 已提交
697 698 699 700 701 702
    int32_t sz = taosArrayGetSize(pConsumer->assignedTopics);
    for (int32_t i = 0; i < sz; i++) {
      char *name = taosArrayGetP(pConsumer->assignedTopics, i);
      if (strcmp(name, pTopic->name) == 0) {
        mndReleaseConsumer(pMnode, pConsumer);
        mndReleaseTopic(pMnode, pTopic);
703
        sdbCancelFetch(pSdb, pIter);
L
Liu Jicong 已提交
704
        terrno = TSDB_CODE_MND_TOPIC_SUBSCRIBED;
X
Xiaoyu Wang 已提交
705 706
        mError("topic:%s, failed to drop since subscribed by consumer:0x%" PRIx64 ", in consumer group %s",
               dropReq.name, pConsumer->consumerId, pConsumer->cgroup);
L
Liu Jicong 已提交
707 708 709
        return -1;
      }
    }
710 711 712 713 714 715 716

    sz = taosArrayGetSize(pConsumer->rebNewTopics);
    for (int32_t i = 0; i < sz; i++) {
      char *name = taosArrayGetP(pConsumer->rebNewTopics, i);
      if (strcmp(name, pTopic->name) == 0) {
        mndReleaseConsumer(pMnode, pConsumer);
        mndReleaseTopic(pMnode, pTopic);
717
        sdbCancelFetch(pSdb, pIter);
718 719 720 721 722 723 724 725 726 727 728 729 730
        terrno = TSDB_CODE_MND_TOPIC_SUBSCRIBED;
        mError("topic:%s, failed to drop since subscribed by consumer:%" PRId64 ", in consumer group %s (reb new)",
               dropReq.name, pConsumer->consumerId, pConsumer->cgroup);
        return -1;
      }
    }

    sz = taosArrayGetSize(pConsumer->rebRemovedTopics);
    for (int32_t i = 0; i < sz; i++) {
      char *name = taosArrayGetP(pConsumer->rebRemovedTopics, i);
      if (strcmp(name, pTopic->name) == 0) {
        mndReleaseConsumer(pMnode, pConsumer);
        mndReleaseTopic(pMnode, pTopic);
731
        sdbCancelFetch(pSdb, pIter);
732 733 734 735 736 737 738
        terrno = TSDB_CODE_MND_TOPIC_SUBSCRIBED;
        mError("topic:%s, failed to drop since subscribed by consumer:%" PRId64 ", in consumer group %s (reb remove)",
               dropReq.name, pConsumer->consumerId, pConsumer->cgroup);
        return -1;
      }
    }

L
Liu Jicong 已提交
739 740 741
    sdbRelease(pSdb, pConsumer);
  }

742
  if (mndCheckDbPrivilegeByName(pMnode, pReq->info.conn.user, MND_OPER_READ_DB, pTopic->db) != 0) {
743
    mndReleaseTopic(pMnode, pTopic);
S
Shengliang Guan 已提交
744 745 746
    return -1;
  }

747
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_DB_INSIDE, pReq, "drop-topic");
L
Liu Jicong 已提交
748 749
  if (pTrans == NULL) {
    mError("topic:%s, failed to drop since %s", pTopic->name, terrstr());
750
    mndReleaseTopic(pMnode, pTopic);
L
Liu Jicong 已提交
751 752
    return -1;
  }
L
Liu Jicong 已提交
753

754
  mndTransSetDbName(pTrans, pTopic->db, NULL);
755
  if (mndTransCheckConflict(pMnode, pTrans) != 0) {
756 757 758 759 760
    mndReleaseTopic(pMnode, pTopic);
    mndTransDrop(pTrans);
    return -1;
  }

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

763
  // TODO check if rebalancing
L
Liu Jicong 已提交
764
  if (mndDropSubByTopic(pMnode, pTrans, dropReq.name) < 0) {
765 766 767
    mError("topic:%s, failed to drop since %s", pTopic->name, terrstr());
    mndTransDrop(pTrans);
    mndReleaseTopic(pMnode, pTopic);
L
Liu Jicong 已提交
768 769 770
    return -1;
  }

771 772
  if (pTopic->ntbUid != 0) {
    // broadcast to all vnode
H
Haojun Liao 已提交
773
    pIter = NULL;
774
    SVgObj *pVgroup = NULL;
H
Haojun Liao 已提交
775

776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792
    while (1) {
      pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
      if (pIter == NULL) break;
      if (!mndVgroupInDb(pVgroup, pTopic->dbUid)) {
        sdbRelease(pSdb, pVgroup);
        continue;
      }

      void *buf = taosMemoryCalloc(1, sizeof(SMsgHead) + TSDB_TOPIC_FNAME_LEN);
      void *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
      ((SMsgHead *)buf)->vgId = htonl(pVgroup->vgId);
      memcpy(abuf, pTopic->name, TSDB_TOPIC_FNAME_LEN);

      STransAction action = {0};
      action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
      action.pCont = buf;
      action.contLen = sizeof(SMsgHead) + TSDB_TOPIC_FNAME_LEN;
L
Liu Jicong 已提交
793
      action.msgType = TDMT_VND_TMQ_DEL_CHECKINFO;
794 795 796
      if (mndTransAppendRedoAction(pTrans, &action) != 0) {
        taosMemoryFree(buf);
        sdbRelease(pSdb, pVgroup);
797
        mndReleaseTopic(pMnode, pTopic);
798
        sdbCancelFetch(pSdb, pIter);
799 800 801
        mndTransDrop(pTrans);
        return -1;
      }
802
      sdbRelease(pSdb, pVgroup);
803 804 805
    }
  }

L
Liu Jicong 已提交
806
  int32_t code = mndDropTopic(pMnode, pTrans, pReq, pTopic);
L
Liu Jicong 已提交
807
  mndReleaseTopic(pMnode, pTopic);
808
  mndTransDrop(pTrans);
L
Liu Jicong 已提交
809 810

  if (code != 0) {
S
Shengliang Guan 已提交
811
    mError("topic:%s, failed to drop since %s", dropReq.name, terrstr());
L
Liu Jicong 已提交
812 813 814
    return -1;
  }

S
Shengliang Guan 已提交
815
  return TSDB_CODE_ACTION_IN_PROGRESS;
L
Liu Jicong 已提交
816 817
}

818
int32_t mndGetNumOfTopics(SMnode *pMnode, char *dbName, int32_t *pNumOfTopics) {
819 820
  *pNumOfTopics = 0;

L
Liu Jicong 已提交
821 822 823 824 825 826 827 828 829 830 831 832
  SSdb   *pSdb = pMnode->pSdb;
  SDbObj *pDb = mndAcquireDb(pMnode, dbName);
  if (pDb == NULL) {
    terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
    return -1;
  }

  int32_t numOfTopics = 0;
  void   *pIter = NULL;
  while (1) {
    SMqTopicObj *pTopic = NULL;
    pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);
833 834 835
    if (pIter == NULL) {
      break;
    }
L
Liu Jicong 已提交
836 837 838 839 840 841 842 843 844 845 846 847 848

    if (pTopic->dbUid == pDb->uid) {
      numOfTopics++;
    }

    sdbRelease(pSdb, pTopic);
  }

  *pNumOfTopics = numOfTopics;
  mndReleaseDb(pMnode, pDb);
  return 0;
}

849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885
static void schemaToJson(SSchema *schema, int32_t nCols, char *schemaJson){
  char*  string = NULL;
  cJSON* columns = cJSON_CreateArray();
  if (columns == NULL) {
    return;
  }
  for (int i = 0; i < nCols; i++) {
    cJSON*   column = cJSON_CreateObject();
    SSchema* s = schema + i;
    cJSON*   cname = cJSON_CreateString(s->name);
    cJSON_AddItemToObject(column, "name", cname);
    cJSON* ctype = cJSON_CreateString(tDataTypes[s->type].name);
    cJSON_AddItemToObject(column, "type", ctype);
    int32_t length = 0;
    if (s->type == TSDB_DATA_TYPE_BINARY) {
      length = s->bytes - VARSTR_HEADER_SIZE;
    } else if (s->type == TSDB_DATA_TYPE_NCHAR || s->type == TSDB_DATA_TYPE_JSON) {
      length = (s->bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
    } else{
      length = s->bytes;
    }
    cJSON*  cbytes = cJSON_CreateNumber(length);
    cJSON_AddItemToObject(column, "length", cbytes);
    cJSON_AddItemToArray(columns, column);
  }
  string = cJSON_PrintUnformatted(columns);
  cJSON_Delete(columns);

  size_t len = strlen(string);
  if(string && len <= TSDB_SHOW_SCHEMA_JSON_LEN){
    STR_TO_VARSTR(schemaJson, string);
  }else{
    mError("mndRetrieveTopic build schema error json:%p, json len:%zu", string, len);
  }
  taosMemoryFree(string);
}

S
Shengliang Guan 已提交
886 887
static int32_t mndRetrieveTopic(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) {
  SMnode      *pMnode = pReq->info.node;
L
Liu Jicong 已提交
888 889 890
  SSdb        *pSdb = pMnode->pSdb;
  int32_t      numOfRows = 0;
  SMqTopicObj *pTopic = NULL;
S
Shengliang Guan 已提交
891

L
Liu Jicong 已提交
892
  while (numOfRows < rowsCapacity) {
L
Liu Jicong 已提交
893 894 895
    pShow->pIter = sdbFetch(pSdb, SDB_TOPIC, pShow->pIter, (void **)&pTopic);
    if (pShow->pIter == NULL) break;

L
Liu Jicong 已提交
896 897 898
    SColumnInfoData *pColInfo;
    SName            n;
    int32_t          cols = 0;
S
Shengliang Guan 已提交
899

X
Xiaoyu Wang 已提交
900 901
    char        topicName[TSDB_TOPIC_NAME_LEN + VARSTR_HEADER_SIZE + 5] = {0};
    const char *pName = mndGetDbStr(pTopic->name);
902 903
    STR_TO_VARSTR(topicName, pName);

L
Liu Jicong 已提交
904
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
905
    colDataSetVal(pColInfo, numOfRows, (const char *)topicName, false);
L
Liu Jicong 已提交
906

L
Liu Jicong 已提交
907 908 909 910
    char dbName[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
    tNameFromString(&n, pTopic->db, T_NAME_ACCT | T_NAME_DB);
    tNameGetDbName(&n, varDataVal(dbName));
    varDataSetLen(dbName, strlen(varDataVal(dbName)));
911

L
Liu Jicong 已提交
912
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
913
    colDataSetVal(pColInfo, numOfRows, (const char *)dbName, false);
L
Liu Jicong 已提交
914

L
Liu Jicong 已提交
915
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
916
    colDataSetVal(pColInfo, numOfRows, (const char *)&pTopic->createTime, false);
L
Liu Jicong 已提交
917

D
dmchen 已提交
918
    char *sql = taosMemoryMalloc(strlen(pTopic->sql) + VARSTR_HEADER_SIZE);
919 920
    STR_TO_VARSTR(sql, pTopic->sql);

L
Liu Jicong 已提交
921
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
922
    colDataSetVal(pColInfo, numOfRows, (const char *)sql, false);
L
Liu Jicong 已提交
923

D
dmchen 已提交
924 925
    taosMemoryFree(sql);

926
    char *schemaJson = taosMemoryMalloc(TSDB_SHOW_SCHEMA_JSON_LEN + VARSTR_HEADER_SIZE);
927 928 929 930 931
    if(pTopic->subType == TOPIC_SUB_TYPE__COLUMN){
      schemaToJson(pTopic->schema.pSchema, pTopic->schema.nCols, schemaJson);
    }else if(pTopic->subType == TOPIC_SUB_TYPE__TABLE){
      SStbObj *pStb = mndAcquireStb(pMnode, pTopic->stbName);
      if (pStb == NULL) {
wmmhello's avatar
wmmhello 已提交
932 933 934 935 936
        STR_TO_VARSTR(schemaJson, "NULL");
        mError("mndRetrieveTopic mndAcquireStb null stbName:%s", pTopic->stbName);
      }else{
        schemaToJson(pStb->pColumns, pStb->numOfColumns, schemaJson);
        mndReleaseStb(pMnode, pStb);
937 938 939 940 941 942 943
      }
    }else{
      STR_TO_VARSTR(schemaJson, "NULL");
    }

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataSetVal(pColInfo, numOfRows, (const char *)schemaJson, false);
944
    taosMemoryFree(schemaJson);
945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967

    char mete[4 + VARSTR_HEADER_SIZE] = {0};
    if(pTopic->withMeta){
      STR_TO_VARSTR(mete, "yes");
    }else{
      STR_TO_VARSTR(mete, "no");
    }

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataSetVal(pColInfo, numOfRows, (const char *)mete, false);

    char type[8 + VARSTR_HEADER_SIZE] = {0};
    if(pTopic->subType == TOPIC_SUB_TYPE__COLUMN){
      STR_TO_VARSTR(type, "column");
    }else if(pTopic->subType == TOPIC_SUB_TYPE__TABLE){
      STR_TO_VARSTR(type, "stable");
    }else{
      STR_TO_VARSTR(type, "db");
    }

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataSetVal(pColInfo, numOfRows, (const char *)type, false);

L
Liu Jicong 已提交
968 969 970 971
    numOfRows++;
    sdbRelease(pSdb, pTopic);
  }

972
  pShow->numOfRows += numOfRows;
L
Liu Jicong 已提交
973 974 975
  return numOfRows;
}

976 977 978 979 980
int32_t mndSetTopicCommitLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic) {
  SSdbRaw *pCommitRaw = mndTopicActionEncode(pTopic);
  if (pCommitRaw == NULL) return -1;
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) return -1;
L
Liu Jicong 已提交
981 982 983 984

  return 0;
}

L
Liu Jicong 已提交
985 986 987 988 989 990 991 992 993
static int32_t mndSetDropTopicCommitLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic) {
  SSdbRaw *pCommitRaw = mndTopicActionEncode(pTopic);
  if (pCommitRaw == NULL) return -1;
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED) != 0) return -1;

  return 0;
}

L
Liu Jicong 已提交
994 995 996 997
static void mndCancelGetNextTopic(SMnode *pMnode, void *pIter) {
  SSdb *pSdb = pMnode->pSdb;
  sdbCancelFetch(pSdb, pIter);
}
L
Liu Jicong 已提交
998

999 1000
bool mndTopicExistsForDb(SMnode *pMnode, SDbObj *pDb) {
  SSdb        *pSdb = pMnode->pSdb;
1001 1002
  void        *pIter = NULL;
  SMqTopicObj *pTopic = NULL;
1003

1004 1005
  while (1) {
    pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);
1006 1007 1008
    if (pIter == NULL) {
      break;
    }
1009 1010 1011

    if (pTopic->dbUid == pDb->uid) {
      sdbRelease(pSdb, pTopic);
1012
      sdbCancelFetch(pSdb, pIter);
1013
      return true;
1014 1015 1016 1017
    }

    sdbRelease(pSdb, pTopic);
  }
1018 1019

  return false;
1020 1021
}

1022
#if 0
L
Liu Jicong 已提交
1023
int32_t mndDropTopicByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
1024
  int32_t code = 0;
L
Liu Jicong 已提交
1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038
  SSdb   *pSdb = pMnode->pSdb;

  void        *pIter = NULL;
  SMqTopicObj *pTopic = NULL;
  while (1) {
    pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic);
    if (pIter == NULL) break;

    if (pTopic->dbUid != pDb->uid) {
      sdbRelease(pSdb, pTopic);
      continue;
    }

    if (mndSetDropTopicCommitLogs(pMnode, pTrans, pTopic) < 0) {
1039 1040 1041 1042
      sdbRelease(pSdb, pTopic);
      sdbCancelFetch(pSdb, pIter);
      code = -1;
      break;
L
Liu Jicong 已提交
1043
    }
1044 1045

    sdbRelease(pSdb, pTopic);
L
Liu Jicong 已提交
1046 1047 1048 1049
  }

  return code;
}
1050
#endif