Merge pull request #11893 from taosdata/feature/tq

feat(tmq): support show

example/src/tmq.c

@@ -18,7 +18,6 @@
 #include <string.h>
 #include <time.h>
 #include "taos.h"
-#include "osSleep.h"
 
 static int  running = 1;
 static void msg_process(TAOS_RES* msg) {

include/common/tcommon.h

@@ -66,13 +66,13 @@ typedef struct SDataBlockInfo {
   int32_t     rows;
   int32_t     rowSize;
   union {
-    int64_t uid;        // from which table of uid, comes from this data block
+    int64_t uid;  // from which table of uid, comes from this data block
     int64_t blockId;
   };
-  uint64_t    groupId;  // no need to serialize
-  int16_t     numOfCols;
-  int16_t     hasVarCol;
-  int16_t     capacity;
+  uint64_t groupId;  // no need to serialize
+  int16_t  numOfCols;
+  int16_t  hasVarCol;
+  int16_t  capacity;
 } SDataBlockInfo;
 
 typedef struct SSDataBlock {
@@ -131,59 +131,6 @@ static FORCE_INLINE void blockDestroyInner(SSDataBlock* pBlock) {
 
 static FORCE_INLINE void tDeleteSSDataBlock(SSDataBlock* pBlock) { blockDestroyInner(pBlock); }
 
-static FORCE_INLINE int32_t tEncodeSMqPollRsp(void** buf, const SMqPollRsp* pRsp) {
-  int32_t tlen = 0;
-  int32_t sz = 0;
-  // tlen += taosEncodeFixedI64(buf, pRsp->consumerId);
-  tlen += taosEncodeFixedI64(buf, pRsp->reqOffset);
-  tlen += taosEncodeFixedI64(buf, pRsp->rspOffset);
-  tlen += taosEncodeFixedI32(buf, pRsp->skipLogNum);
-  tlen += taosEncodeFixedI32(buf, pRsp->numOfTopics);
-  if (pRsp->numOfTopics == 0) return tlen;
-  tlen += taosEncodeSSchemaWrapper(buf, pRsp->schema);
-  if (pRsp->pBlockData) {
-    sz = taosArrayGetSize(pRsp->pBlockData);
-  }
-  tlen += taosEncodeFixedI32(buf, sz);
-  for (int32_t i = 0; i < sz; i++) {
-    SSDataBlock* pBlock = (SSDataBlock*)taosArrayGet(pRsp->pBlockData, i);
-    tlen += tEncodeDataBlock(buf, pBlock);
-  }
-  return tlen;
-}
-
-static FORCE_INLINE void* tDecodeSMqPollRsp(void* buf, SMqPollRsp* pRsp) {
-  int32_t sz;
-  // buf = taosDecodeFixedI64(buf, &pRsp->consumerId);
-  buf = taosDecodeFixedI64(buf, &pRsp->reqOffset);
-  buf = taosDecodeFixedI64(buf, &pRsp->rspOffset);
-  buf = taosDecodeFixedI32(buf, &pRsp->skipLogNum);
-  buf = taosDecodeFixedI32(buf, &pRsp->numOfTopics);
-  if (pRsp->numOfTopics == 0) return buf;
-  pRsp->schema = (SSchemaWrapper*)taosMemoryCalloc(1, sizeof(SSchemaWrapper));
-  if (pRsp->schema == NULL) return NULL;
-  buf = taosDecodeSSchemaWrapper(buf, pRsp->schema);
-  buf = taosDecodeFixedI32(buf, &sz);
-  pRsp->pBlockData = taosArrayInit(sz, sizeof(SSDataBlock));
-  for (int32_t i = 0; i < sz; i++) {
-    SSDataBlock block = {0};
-    tDecodeDataBlock(buf, &block);
-    taosArrayPush(pRsp->pBlockData, &block);
-  }
-  return buf;
-}
-
-static FORCE_INLINE void tDeleteSMqConsumeRsp(SMqPollRsp* pRsp) {
-  if (pRsp->schema) {
-    if (pRsp->schema->nCols) {
-      taosMemoryFreeClear(pRsp->schema->pSchema);
-    }
-    taosMemoryFree(pRsp->schema);
-  }
-  taosArrayDestroyEx(pRsp->pBlockData, (void (*)(void*))blockDestroyInner);
-  pRsp->pBlockData = NULL;
-}
-
 //======================================================================================================================
 // the following structure shared by parser and executor
 typedef struct SColumn {
@@ -204,22 +151,22 @@ typedef struct SColumn {
 } SColumn;
 
 typedef struct STableBlockDistInfo {
-  uint16_t  rowSize;
-  uint16_t  numOfFiles;
-  uint32_t  numOfTables;
-  uint64_t  totalSize;
-  uint64_t  totalRows;
-  int32_t   maxRows;
-  int32_t   minRows;
-  int32_t   firstSeekTimeUs;
-  uint32_t  numOfRowsInMemTable;
-  uint32_t  numOfSmallBlocks;
-  SArray   *dataBlockInfos;
+  uint16_t rowSize;
+  uint16_t numOfFiles;
+  uint32_t numOfTables;
+  uint64_t totalSize;
+  uint64_t totalRows;
+  int32_t  maxRows;
+  int32_t  minRows;
+  int32_t  firstSeekTimeUs;
+  uint32_t numOfRowsInMemTable;
+  uint32_t numOfSmallBlocks;
+  SArray*  dataBlockInfos;
 } STableBlockDistInfo;
 
 enum {
   FUNC_PARAM_TYPE_VALUE = 0x1,
-  FUNC_PARAM_TYPE_COLUMN= 0x2,
+  FUNC_PARAM_TYPE_COLUMN = 0x2,
 };
 
 typedef struct SFunctParam {
@@ -249,7 +196,7 @@ typedef struct SExprInfo {
   struct tExprNode*     pExpr;
 } SExprInfo;
 
-#define QUERY_ASC_FORWARD_STEP 1
+#define QUERY_ASC_FORWARD_STEP  1
 #define QUERY_DESC_FORWARD_STEP -1
 
 #define GET_FORWARD_DIRECTION_FACTOR(ord) (((ord) == TSDB_ORDER_ASC) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP)

include/common/tmsg.h

@@ -1650,103 +1650,11 @@ typedef struct {
   char    data[];
 } SVShowTablesFetchRsp;
 
-typedef struct SMqCMGetSubEpReq {
+typedef struct {
   int64_t consumerId;
   int32_t epoch;
   char    cgroup[TSDB_CGROUP_LEN];
-} SMqCMGetSubEpReq;
-
-static FORCE_INLINE int32_t tEncodeSMsgHead(void** buf, const SMsgHead* pMsg) {
-  int32_t tlen = 0;
-  tlen += taosEncodeFixedI32(buf, pMsg->contLen);
-  tlen += taosEncodeFixedI32(buf, pMsg->vgId);
-  return tlen;
-}
-
-typedef struct SMqHbRsp {
-  int8_t status;  // idle or not
-  int8_t vnodeChanged;
-  int8_t epChanged;  // should use new epset
-  int8_t reserved;
-  SEpSet epSet;
-} SMqHbRsp;
-
-static FORCE_INLINE int32_t taosEncodeSMqHbRsp(void** buf, const SMqHbRsp* pRsp) {
-  int32_t tlen = 0;
-  tlen += taosEncodeFixedI8(buf, pRsp->status);
-  tlen += taosEncodeFixedI8(buf, pRsp->vnodeChanged);
-  tlen += taosEncodeFixedI8(buf, pRsp->epChanged);
-  tlen += taosEncodeSEpSet(buf, &pRsp->epSet);
-  return tlen;
-}
-
-static FORCE_INLINE void* taosDecodeSMqHbRsp(void* buf, SMqHbRsp* pRsp) {
-  buf = taosDecodeFixedI8(buf, &pRsp->status);
-  buf = taosDecodeFixedI8(buf, &pRsp->vnodeChanged);
-  buf = taosDecodeFixedI8(buf, &pRsp->epChanged);
-  buf = taosDecodeSEpSet(buf, &pRsp->epSet);
-  return buf;
-}
-
-typedef struct SMqHbOneTopicBatchRsp {
-  char    topicName[TSDB_TOPIC_FNAME_LEN];
-  SArray* rsps;  // SArray<SMqHbRsp>
-} SMqHbOneTopicBatchRsp;
-
-static FORCE_INLINE int32_t taosEncodeSMqHbOneTopicBatchRsp(void** buf, const SMqHbOneTopicBatchRsp* pBatchRsp) {
-  int32_t tlen = 0;
-  tlen += taosEncodeString(buf, pBatchRsp->topicName);
-  int32_t sz = taosArrayGetSize(pBatchRsp->rsps);
-  tlen += taosEncodeFixedI32(buf, sz);
-  for (int32_t i = 0; i < sz; i++) {
-    SMqHbRsp* pRsp = (SMqHbRsp*)taosArrayGet(pBatchRsp->rsps, i);
-    tlen += taosEncodeSMqHbRsp(buf, pRsp);
-  }
-  return tlen;
-}
-
-static FORCE_INLINE void* taosDecodeSMqHbOneTopicBatchRsp(void* buf, SMqHbOneTopicBatchRsp* pBatchRsp) {
-  int32_t sz;
-  buf = taosDecodeStringTo(buf, pBatchRsp->topicName);
-  buf = taosDecodeFixedI32(buf, &sz);
-  pBatchRsp->rsps = taosArrayInit(sz, sizeof(SMqHbRsp));
-  for (int32_t i = 0; i < sz; i++) {
-    SMqHbRsp rsp;
-    buf = taosDecodeSMqHbRsp(buf, &rsp);
-    buf = taosArrayPush(pBatchRsp->rsps, &rsp);
-  }
-  return buf;
-}
-
-typedef struct SMqHbBatchRsp {
-  int64_t consumerId;
-  SArray* batchRsps;  // SArray<SMqHbOneTopicBatchRsp>
-} SMqHbBatchRsp;
-
-static FORCE_INLINE int32_t taosEncodeSMqHbBatchRsp(void** buf, const SMqHbBatchRsp* pBatchRsp) {
-  int32_t tlen = 0;
-  tlen += taosEncodeFixedI64(buf, pBatchRsp->consumerId);
-  int32_t sz;
-  tlen += taosEncodeFixedI32(buf, sz);
-  for (int32_t i = 0; i < sz; i++) {
-    SMqHbOneTopicBatchRsp* pRsp = (SMqHbOneTopicBatchRsp*)taosArrayGet(pBatchRsp->batchRsps, i);
-    tlen += taosEncodeSMqHbOneTopicBatchRsp(buf, pRsp);
-  }
-  return tlen;
-}
-
-static FORCE_INLINE void* taosDecodeSMqHbBatchRsp(void* buf, SMqHbBatchRsp* pBatchRsp) {
-  buf = taosDecodeFixedI64(buf, &pBatchRsp->consumerId);
-  int32_t sz;
-  buf = taosDecodeFixedI32(buf, &sz);
-  pBatchRsp->batchRsps = taosArrayInit(sz, sizeof(SMqHbOneTopicBatchRsp));
-  for (int32_t i = 0; i < sz; i++) {
-    SMqHbOneTopicBatchRsp rsp;
-    buf = taosDecodeSMqHbOneTopicBatchRsp(buf, &rsp);
-    buf = taosArrayPush(pBatchRsp->batchRsps, &rsp);
-  }
-  return buf;
-}
+} SMqAskEpReq;
 
 typedef struct {
   int32_t key;
@@ -2470,22 +2378,6 @@ typedef struct {
   int64_t consumerId;
 } SMqRspHead;
 
-#if 0
-typedef struct {
-  SMsgHead head;
-
-  int64_t consumerId;
-  int64_t blockingTime;
-  int32_t epoch;
-  int8_t  withSchema;
-  char    cgroup[TSDB_CGROUP_LEN];
-
-  int64_t  currentOffset;
-  uint64_t reqId;
-  char     topic[TSDB_TOPIC_FNAME_LEN];
-} SMqPollReq;
-#endif
-
 typedef struct {
   SMsgHead head;
   char     subKey[TSDB_SUBSCRIBE_KEY_LEN];
@@ -2509,18 +2401,6 @@ typedef struct {
   SSchemaWrapper schema;
 } SMqSubTopicEp;
 
-typedef struct {
-  SMqRspHead head;
-  int64_t    reqOffset;
-  int64_t    rspOffset;
-  int32_t    skipLogNum;
-  // TODO: replace with topic name
-  int32_t numOfTopics;
-  // TODO: remove from msg
-  SSchemaWrapper* schema;
-  SArray*         pBlockData;  // SArray<SSDataBlock>
-} SMqPollRsp;
-
 typedef struct {
   SMqRspHead head;
   int64_t    reqOffset;
@@ -2644,7 +2524,7 @@ typedef struct {
   SMqRspHead head;
   char       cgroup[TSDB_CGROUP_LEN];
   SArray*    topics;  // SArray<SMqSubTopicEp>
-} SMqCMGetSubEpRsp;
+} SMqAskEpRsp;
 
 static FORCE_INLINE void tDeleteSMqSubTopicEp(SMqSubTopicEp* pSubTopicEp) {
   // taosMemoryFree(pSubTopicEp->schema.pSchema);
@@ -2666,10 +2546,6 @@ static FORCE_INLINE void* tDecodeSMqSubVgEp(void* buf, SMqSubVgEp* pVgEp) {
   return buf;
 }
 
-static FORCE_INLINE void tDeleteSMqCMGetSubEpRsp(SMqCMGetSubEpRsp* pRsp) {
-  taosArrayDestroyEx(pRsp->topics, (void (*)(void*))tDeleteSMqSubTopicEp);
-}
-
 static FORCE_INLINE int32_t tEncodeSMqSubTopicEp(void** buf, const SMqSubTopicEp* pTopicEp) {
   int32_t tlen = 0;
   tlen += taosEncodeString(buf, pTopicEp->topic);
@@ -2702,7 +2578,7 @@ static FORCE_INLINE void* tDecodeSMqSubTopicEp(void* buf, SMqSubTopicEp* pTopicE
   return buf;
 }
 
-static FORCE_INLINE int32_t tEncodeSMqCMGetSubEpRsp(void** buf, const SMqCMGetSubEpRsp* pRsp) {
+static FORCE_INLINE int32_t tEncodeSMqAskEpRsp(void** buf, const SMqAskEpRsp* pRsp) {
   int32_t tlen = 0;
   // tlen += taosEncodeString(buf, pRsp->cgroup);
   int32_t sz = taosArrayGetSize(pRsp->topics);
@@ -2714,7 +2590,7 @@ static FORCE_INLINE int32_t tEncodeSMqCMGetSubEpRsp(void** buf, const SMqCMGetSu
   return tlen;
 }
 
-static FORCE_INLINE void* tDecodeSMqCMGetSubEpRsp(void* buf, SMqCMGetSubEpRsp* pRsp) {
+static FORCE_INLINE void* tDecodeSMqAskEpRsp(void* buf, SMqAskEpRsp* pRsp) {
   // buf = taosDecodeStringTo(buf, pRsp->cgroup);
   int32_t sz;
   buf = taosDecodeFixedI32(buf, &sz);
@@ -2730,6 +2606,10 @@ static FORCE_INLINE void* tDecodeSMqCMGetSubEpRsp(void* buf, SMqCMGetSubEpRsp* p
   return buf;
 }
 
+static FORCE_INLINE void tDeleteSMqAskEpRsp(SMqAskEpRsp* pRsp) {
+  taosArrayDestroyEx(pRsp->topics, (void (*)(void*))tDeleteSMqSubTopicEp);
+}
+
 #pragma pack(pop)
 
 #ifdef __cplusplus

include/common/tmsgdef.h

@@ -145,7 +145,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_ALTER_TOPIC, "mnode-alter-topic", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_DROP_TOPIC, "mnode-drop-topic", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_SUBSCRIBE, "mnode-subscribe", SCMSubscribeReq, SCMSubscribeRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_GET_SUB_EP, "mnode-mq-ask-ep", SMqCMGetSubEpReq, SMqCMGetSubEpRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_ASK_EP, "mnode-mq-ask-ep", SMqAskEpReq, SMqAskEpReq)
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mnode-mq-tmr", SMTimerReq, SMTimerReq)
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_LOST, "mnode-mq-consumer-lost", SMTimerReq, SMTimerReq)
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_DO_REBALANCE, "mnode-mq-do-rebalance", SMqDoRebalanceMsg, SMqDoRebalanceMsg)

include/util/tdef.h

@@ -125,12 +125,12 @@ extern const int32_t TYPE_BYTES[15];
 #define TSDB_INS_TABLE_QUERIES                "queries"
 #define TSDB_INS_TABLE_VNODES                 "vnodes"
 
-#define TSDB_PERFORMANCE_SCHEMA_DB   "performance_schema"
-#define TSDB_PERFS_TABLE_CONNECTIONS "connections"
-#define TSDB_PERFS_TABLE_QUERIES     "queries"
-#define TSDB_PERFS_TABLE_TOPICS      "topics"
-#define TSDB_PERFS_TABLE_CONSUMERS   "consumers"
-#define TSDB_PERFS_TABLE_SUBSCRIBES  "subscribes"
+#define TSDB_PERFORMANCE_SCHEMA_DB     "performance_schema"
+#define TSDB_PERFS_TABLE_CONNECTIONS   "connections"
+#define TSDB_PERFS_TABLE_QUERIES       "queries"
+#define TSDB_PERFS_TABLE_TOPICS        "topics"
+#define TSDB_PERFS_TABLE_CONSUMERS     "consumers"
+#define TSDB_PERFS_TABLE_SUBSCRIPTIONS "subscriptions"
 
 #define TSDB_INDEX_TYPE_SMA      "SMA"
 #define TSDB_INDEX_TYPE_FULLTEXT "FULLTEXT"
@@ -286,8 +286,9 @@ typedef enum ELogicConditionType {
 #define TSDB_IPv4ADDR_LEN       16
 #define TSDB_FILENAME_LEN       128
 #define TSDB_SHOW_SQL_LEN       512
-#define TSDB_SHOW_SUBQUERY_LEN  1000
 #define TSDB_SLOW_QUERY_SQL_LEN 512
+#define TSDB_SHOW_SUBQUERY_LEN  1000
+#define TSDB_SHOW_LIST_LEN      1000
 
 #define TSDB_TRANS_STAGE_LEN 12
 #define TSDB_TRANS_TYPE_LEN  16

source/client/src/tmq.c

@@ -25,7 +25,14 @@
 #include "tref.h"
 #include "ttimer.h"
 
-int32_t tmqAskEp(tmq_t* tmq, bool sync);
+int32_t tmqAskEp(tmq_t* tmq, bool async);
+
+typedef struct {
+  int8_t inited;
+  tmr_h  timer;
+} SMqMgmt;
+
+static SMqMgmt tmqMgmt = {0};
 
 typedef struct {
   int8_t  tmqRspType;
@@ -33,9 +40,9 @@ typedef struct {
 } SMqRspWrapper;
 
 typedef struct {
-  int8_t           tmqRspType;
-  int32_t          epoch;
-  SMqCMGetSubEpRsp msg;
+  int8_t      tmqRspType;
+  int32_t     epoch;
+  SMqAskEpRsp msg;
 } SMqAskEpRspWrapper;
 
 struct tmq_list_t {
@@ -64,13 +71,6 @@ struct tmq_conf_t {
   tmq_commit_cb* commit_cb;
 };
 
-typedef struct {
-  int8_t inited;
-  tmr_h  timer;
-} SMqMgmt;
-
-static SMqMgmt tmqMgmt = {0};
-
 struct tmq_t {
   // conf
   char           groupId[TSDB_CGROUP_LEN];
@@ -164,7 +164,7 @@ typedef struct {
 typedef struct {
   tmq_t*  tmq;
   int32_t code;
-  int32_t sync;
+  int32_t async;
   tsem_t  rspSem;
 } SMqAskEpCbParam;
 
@@ -188,6 +188,7 @@ typedef struct {
 tmq_conf_t* tmq_conf_new() {
   tmq_conf_t* conf = taosMemoryCalloc(1, sizeof(tmq_conf_t));
   conf->autoCommit = false;
+  conf->autoCommitInterval = 5000;
   conf->resetOffset = TMQ_CONF__RESET_OFFSET__EARLIEAST;
   return conf;
 }
@@ -324,7 +325,7 @@ int32_t tmqHandleAllDelayedTask(tmq_t* tmq) {
     if (pTaskType == NULL) break;
 
     if (*pTaskType == TMQ_DELAYED_TASK__HB) {
-      tmqAskEp(tmq, false);
+      tmqAskEp(tmq, true);
       taosTmrReset(tmqAssignDelayedHbTask, 1000, tmq, tmqMgmt.timer, &tmq->hbTimer);
     } else if (*pTaskType == TMQ_DELAYED_TASK__COMMIT) {
       tmq_commit(tmq, NULL, true);
@@ -472,8 +473,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
   // set conf
   strcpy(pTmq->clientId, conf->clientId);
   strcpy(pTmq->groupId, conf->groupId);
-  /*pTmq->autoCommit = conf->autoCommit;*/
-  pTmq->autoCommit = 0;
+  pTmq->autoCommit = conf->autoCommit;
   pTmq->autoCommitInterval = conf->autoCommitInterval;
   pTmq->commit_cb = conf->commit_cb;
   pTmq->resetOffsetCfg = conf->resetOffset;
@@ -662,8 +662,8 @@ tmq_resp_err_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
   if (code != 0) goto FAIL;
 
   // TODO: add max retry cnt
-  while (TSDB_CODE_MND_CONSUMER_NOT_READY == tmqAskEp(tmq, true)) {
-    tscDebug("not ready, retry\n");
+  while (TSDB_CODE_MND_CONSUMER_NOT_READY == tmqAskEp(tmq, false)) {
+    tscDebug("not ready, retry");
     taosMsleep(500);
   }
 
@@ -854,7 +854,7 @@ CREATE_MSG_FAIL:
   return -1;
 }
 
-bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqCMGetSubEpRsp* pRsp) {
+bool tmqUpdateEp(tmq_t* tmq, int32_t epoch, SMqAskEpRsp* pRsp) {
   /*printf("call update ep %d\n", epoch);*/
   bool    set = false;
   int32_t topicNumGet = taosArrayGetSize(pRsp->topics);
@@ -936,7 +936,7 @@ int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) {
   tmq_t*           tmq = pParam->tmq;
   pParam->code = code;
   if (code != 0) {
-    tscError("consumer %ld get topic endpoint error, not ready, wait:%d", tmq->consumerId, pParam->sync);
+    tscError("consumer %ld get topic endpoint error, not ready, wait:%d", tmq->consumerId, pParam->async);
     goto END;
   }
 
@@ -950,15 +950,15 @@ int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) {
     goto END;
   }
 
-  if (pParam->sync) {
-    SMqCMGetSubEpRsp rsp;
-    tDecodeSMqCMGetSubEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &rsp);
+  if (!pParam->async) {
+    SMqAskEpRsp rsp;
+    tDecodeSMqAskEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &rsp);
     /*printf("rsp epoch %ld sz %ld\n", rsp.epoch, rsp.topics->size);*/
     /*printf("tmq epoch %ld sz %ld\n", tmq->epoch, tmq->clientTopics->size);*/
     if (tmqUpdateEp(tmq, head->epoch, &rsp)) {
       atomic_store_8(&tmq->status, TMQ_CONSUMER_STATUS__READY);
     }
-    tDeleteSMqCMGetSubEpRsp(&rsp);
+    tDeleteSMqAskEpRsp(&rsp);
   } else {
     SMqAskEpRspWrapper* pWrapper = taosAllocateQitem(sizeof(SMqAskEpRspWrapper));
     if (pWrapper == NULL) {
@@ -969,7 +969,7 @@ int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) {
     pWrapper->tmqRspType = TMQ_MSG_TYPE__EP_RSP;
     pWrapper->epoch = head->epoch;
     memcpy(&pWrapper->msg, pMsg->pData, sizeof(SMqRspHead));
-    tDecodeSMqCMGetSubEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pWrapper->msg);
+    tDecodeSMqAskEpRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pWrapper->msg);
 
     taosWriteQitem(tmq->mqueue, pWrapper);
     /*tsem_post(&tmq->rspSem);*/
@@ -978,13 +978,13 @@ int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) {
 
 END:
   /*atomic_store_8(&tmq->epStatus, 0);*/
-  if (pParam->sync) {
+  if (!pParam->async) {
     tsem_post(&pParam->rspSem);
   }
   return code;
 }
 
-int32_t tmqAskEp(tmq_t* tmq, bool sync) {
+int32_t tmqAskEp(tmq_t* tmq, bool async) {
   int32_t code = 0;
 #if 0
   int8_t  epStatus = atomic_val_compare_exchange_8(&tmq->epStatus, 0, 1);
@@ -995,8 +995,8 @@ int32_t tmqAskEp(tmq_t* tmq, bool sync) {
   }
   atomic_store_32(&tmq->epSkipCnt, 0);
 #endif
-  int32_t           tlen = sizeof(SMqCMGetSubEpReq);
-  SMqCMGetSubEpReq* req = taosMemoryMalloc(tlen);
+  int32_t      tlen = sizeof(SMqAskEpReq);
+  SMqAskEpReq* req = taosMemoryMalloc(tlen);
   if (req == NULL) {
     tscError("failed to malloc get subscribe ep buf");
     /*atomic_store_8(&tmq->epStatus, 0);*/
@@ -1014,7 +1014,7 @@ int32_t tmqAskEp(tmq_t* tmq, bool sync) {
     return -1;
   }
   pParam->tmq = tmq;
-  pParam->sync = sync;
+  pParam->async = async;
   tsem_init(&pParam->rspSem, 0, 0);
 
   SMsgSendInfo* sendInfo = taosMemoryMalloc(sizeof(SMsgSendInfo));
@@ -1036,7 +1036,7 @@ int32_t tmqAskEp(tmq_t* tmq, bool sync) {
   sendInfo->requestObjRefId = 0;
   sendInfo->param = pParam;
   sendInfo->fp = tmqAskEpCb;
-  sendInfo->msgType = TDMT_MND_GET_SUB_EP;
+  sendInfo->msgType = TDMT_MND_MQ_ASK_EP;
 
   SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);
 
@@ -1045,7 +1045,7 @@ int32_t tmqAskEp(tmq_t* tmq, bool sync) {
   int64_t transporterId = 0;
   asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo);
 
-  if (sync) {
+  if (!async) {
     tsem_wait(&pParam->rspSem);
     code = pParam->code;
     taosMemoryFree(pParam);
@@ -1209,7 +1209,7 @@ int32_t tmqHandleNoPollRsp(tmq_t* tmq, SMqRspWrapper* rspWrapper, bool* pReset)
     /*printf("ep %d %d\n", rspMsg->head.epoch, tmq->epoch);*/
     if (rspWrapper->epoch > atomic_load_32(&tmq->epoch)) {
       SMqAskEpRspWrapper* pEpRspWrapper = (SMqAskEpRspWrapper*)rspWrapper;
-      SMqCMGetSubEpRsp*   rspMsg = &pEpRspWrapper->msg;
+      SMqAskEpRsp*        rspMsg = &pEpRspWrapper->msg;
       tmqUpdateEp(tmq, rspWrapper->epoch, rspMsg);
       /*tmqClearUnhandleMsg(tmq);*/
       *pReset = true;
@@ -1271,15 +1271,6 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t blocking_time) {
   SMqRspObj* rspObj;
   int64_t    startTime = taosGetTimestampMs();
 
-  // TODO: put into delayed queue
-#if 0
-  int8_t status = atomic_load_8(&tmq->status);
-  while (0 != tmqAskEp(tmq, status != TMQ_CONSUMER_STATUS__READY)) {
-    tscDebug("not ready, retry\n");
-    taosSsleep(1);
-  }
-#endif
-
   rspObj = tmqHandleAllRsp(tmq, blocking_time, false);
   if (rspObj) {
     return (TAOS_RES*)rspObj;

source/dnode/mgmt/mgmt_mnode/src/mmHandle.c

@@ -211,7 +211,7 @@ void mmInitMsgHandle(SMgmtWrapper *pWrapper) {
   dmSetMsgHandle(pWrapper, TDMT_MND_DROP_TOPIC, mmProcessWriteMsg, DEFAULT_HANDLE);
   dmSetMsgHandle(pWrapper, TDMT_MND_SUBSCRIBE, mmProcessWriteMsg, DEFAULT_HANDLE);
   dmSetMsgHandle(pWrapper, TDMT_MND_MQ_COMMIT_OFFSET, mmProcessWriteMsg, DEFAULT_HANDLE);
-  dmSetMsgHandle(pWrapper, TDMT_MND_GET_SUB_EP, mmProcessReadMsg, DEFAULT_HANDLE);
+  dmSetMsgHandle(pWrapper, TDMT_MND_MQ_ASK_EP, mmProcessReadMsg, DEFAULT_HANDLE);
   dmSetMsgHandle(pWrapper, TDMT_VND_MQ_VG_CHANGE_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);
   dmSetMsgHandle(pWrapper, TDMT_MND_CREATE_STREAM, mmProcessWriteMsg, DEFAULT_HANDLE);
   dmSetMsgHandle(pWrapper, TDMT_VND_TASK_DEPLOY_RSP, mmProcessWriteMsg, DEFAULT_HANDLE);

source/dnode/mnode/impl/inc/mndDef.h

@@ -440,14 +440,12 @@ static FORCE_INLINE void* tDecodeSMqOffsetObj(void* buf, SMqOffsetObj* pOffset)
 }
 
 typedef struct {
-  char    name[TSDB_TOPIC_FNAME_LEN];
-  char    db[TSDB_DB_FNAME_LEN];
-  int64_t createTime;
-  int64_t updateTime;
-  int64_t uid;
-  // TODO: use subDbUid
+  char           name[TSDB_TOPIC_FNAME_LEN];
+  char           db[TSDB_DB_FNAME_LEN];
+  int64_t        createTime;
+  int64_t        updateTime;
+  int64_t        uid;
   int64_t        dbUid;
-  int64_t        subDbUid;
   int32_t        version;
   int8_t         subType;  // db or table
   int8_t         withTbName;

source/dnode/mnode/impl/src/mndConsumer.c

@@ -59,7 +59,7 @@ int32_t mndInitConsumer(SMnode *pMnode) {
                      .deleteFp = (SdbDeleteFp)mndConsumerActionDelete};
 
   mndSetMsgHandle(pMnode, TDMT_MND_SUBSCRIBE, mndProcessSubscribeReq);
-  mndSetMsgHandle(pMnode, TDMT_MND_GET_SUB_EP, mndProcessAskEpReq);
+  mndSetMsgHandle(pMnode, TDMT_MND_MQ_ASK_EP, mndProcessAskEpReq);
   mndSetMsgHandle(pMnode, TDMT_MND_MQ_TIMER, mndProcessMqTimerMsg);
   mndSetMsgHandle(pMnode, TDMT_MND_MQ_CONSUMER_LOST, mndProcessConsumerLostMsg);
   return sdbSetTable(pMnode->pSdb, table);
@@ -86,7 +86,7 @@ static int32_t mndProcessConsumerLostMsg(SNodeMsg *pMsg) {
   mndTransDrop(pTrans);
   return 0;
 FAIL:
-  // TODO delete consumer
+  tDeleteSMqConsumerObj(pConsumerNew);
   mndTransDrop(pTrans);
   return -1;
 }
@@ -197,11 +197,11 @@ static int32_t mndProcessMqTimerMsg(SNodeMsg *pMsg) {
 }
 
 static int32_t mndProcessAskEpReq(SNodeMsg *pMsg) {
-  SMnode           *pMnode = pMsg->pNode;
-  SMqCMGetSubEpReq *pReq = (SMqCMGetSubEpReq *)pMsg->rpcMsg.pCont;
-  SMqCMGetSubEpRsp  rsp = {0};
-  int64_t           consumerId = be64toh(pReq->consumerId);
-  int32_t           epoch = ntohl(pReq->epoch);
+  SMnode      *pMnode = pMsg->pNode;
+  SMqAskEpReq *pReq = (SMqAskEpReq *)pMsg->rpcMsg.pCont;
+  SMqAskEpRsp  rsp = {0};
+  int64_t      consumerId = be64toh(pReq->consumerId);
+  int32_t      epoch = ntohl(pReq->epoch);
 
   SMqConsumerObj *pConsumer = mndAcquireConsumer(pMsg->pNode, consumerId);
   if (pConsumer == NULL) {
@@ -300,7 +300,7 @@ static int32_t mndProcessAskEpReq(SNodeMsg *pMsg) {
     taosRUnLockLatch(&pConsumer->lock);
   }
   // encode rsp
-  int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqCMGetSubEpRsp(NULL, &rsp);
+  int32_t tlen = sizeof(SMqRspHead) + tEncodeSMqAskEpRsp(NULL, &rsp);
   void   *buf = rpcMallocCont(tlen);
   if (buf == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -311,10 +311,10 @@ static int32_t mndProcessAskEpReq(SNodeMsg *pMsg) {
   ((SMqRspHead *)buf)->consumerId = pConsumer->consumerId;
 
   void *abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
-  tEncodeSMqCMGetSubEpRsp(&abuf, &rsp);
+  tEncodeSMqAskEpRsp(&abuf, &rsp);
 
   // release consumer and free memory
-  tDeleteSMqCMGetSubEpRsp(&rsp);
+  tDeleteSMqAskEpRsp(&rsp);
   mndReleaseConsumer(pMnode, pConsumer);
 
   // send rsp
@@ -322,7 +322,7 @@ static int32_t mndProcessAskEpReq(SNodeMsg *pMsg) {
   pMsg->rspLen = tlen;
   return 0;
 FAIL:
-  tDeleteSMqCMGetSubEpRsp(&rsp);
+  tDeleteSMqAskEpRsp(&rsp);
   mndReleaseConsumer(pMnode, pConsumer);
   return -1;
 }

source/dnode/mnode/impl/src/mndDef.c

@@ -215,7 +215,7 @@ SMqSubscribeObj *tNewSubscribeObj(const char key[TSDB_SUBSCRIBE_KEY_LEN]) {
   if (pSubNew == NULL) return NULL;
   memcpy(pSubNew->key, key, TSDB_SUBSCRIBE_KEY_LEN);
   taosInitRWLatch(&pSubNew->lock);
-  pSubNew->vgNum = -1;
+  pSubNew->vgNum = 0;
   pSubNew->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
   // TODO set free fp
   SMqConsumerEpInSub epInSub = {

source/dnode/mnode/impl/src/mndPerfSchema.c

@@ -41,29 +41,31 @@ static const SPerfsTableSchema queriesSchema[] = {
 
 static const SPerfsTableSchema topicSchema[] = {
     {.name = "topic_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
-    /*{.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},*/
+    {.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
     {.name = "sql", .bytes = TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
-    /*{.name = "row_len", .bytes = 4, .type = TSDB_DATA_TYPE_INT},*/
+    // TODO config
 };
 
 static const SPerfsTableSchema consumerSchema[] = {
-    {.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "consumer_id", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
     {.name = "app_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
-    {.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "status", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
-    // ep
-    // up time
-    // topics
+    {.name = "topics", .bytes = TSDB_SHOW_LIST_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
+    {.name = "end_point", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "up_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
 };
 
-static const SPerfsTableSchema subscribeSchema[] = {
+static const SPerfsTableSchema subscriptionSchema[] = {
     {.name = "topic_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
-    {.name = "offset", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
-    {.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "consumer_id", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
+    {.name = "committed_offset", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
+    {.name = "current_offset", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
+    {.name = "skip_log_cnt", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
 };
 
 static const SPerfsTableMeta perfsMeta[] = {
@@ -71,7 +73,7 @@ static const SPerfsTableMeta perfsMeta[] = {
     {TSDB_PERFS_TABLE_QUERIES, queriesSchema, tListLen(queriesSchema)},
     {TSDB_PERFS_TABLE_TOPICS, topicSchema, tListLen(topicSchema)},
     {TSDB_PERFS_TABLE_CONSUMERS, consumerSchema, tListLen(consumerSchema)},
-    {TSDB_PERFS_TABLE_SUBSCRIBES, subscribeSchema, tListLen(subscribeSchema)},
+    {TSDB_PERFS_TABLE_SUBSCRIPTIONS, subscriptionSchema, tListLen(subscriptionSchema)},
 };
 
 // connection/application/

source/dnode/mnode/impl/src/mndScheduler.c

@@ -478,6 +478,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
   SVgObj*     pVgroup = NULL;
   SQueryPlan* pPlan = NULL;
   SSubplan*   plan = NULL;
+
   if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {
     pPlan = qStringToQueryPlan(pTopic->physicalPlan);
     if (pPlan == NULL) {
@@ -485,10 +486,6 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
       return -1;
     }
 
-    ASSERT(pSub->vgNum == -1);
-
-    pSub->vgNum = 0;
-
     int32_t levelNum = LIST_LENGTH(pPlan->pSubplans);
     if (levelNum != 1) {
       qDestroyQueryPlan(pPlan);
@@ -529,7 +526,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
     pVgEp->vgId = pVgroup->vgId;
     taosArrayPush(pEpInSub->vgs, &pVgEp);
 
-    mDebug("init subscribption %s, assign vg: %d", pSub->key, pVgEp->vgId);
+    mDebug("init subscription %s, assign vg: %d", pSub->key, pVgEp->vgId);
 
     if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {
       int32_t msgLen;

source/dnode/mnode/impl/src/mndTopic.c

@@ -76,7 +76,6 @@ SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {
   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_INT64(pRaw, dataPos, pTopic->subDbUid, TOPIC_ENCODE_OVER);
   SDB_SET_INT32(pRaw, dataPos, pTopic->version, TOPIC_ENCODE_OVER);
   SDB_SET_INT8(pRaw, dataPos, pTopic->subType, TOPIC_ENCODE_OVER);
   SDB_SET_INT8(pRaw, dataPos, pTopic->withTbName, TOPIC_ENCODE_OVER);
@@ -139,7 +138,6 @@ SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {
   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_INT64(pRaw, dataPos, &pTopic->subDbUid, TOPIC_DECODE_OVER);
   SDB_GET_INT32(pRaw, dataPos, &pTopic->version, TOPIC_DECODE_OVER);
   SDB_GET_INT8(pRaw, dataPos, &pTopic->subType, TOPIC_DECODE_OVER);
   SDB_GET_INT8(pRaw, dataPos, &pTopic->withTbName, TOPIC_DECODE_OVER);
@@ -520,29 +518,33 @@ static int32_t mndRetrieveTopic(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pB
     pShow->pIter = sdbFetch(pSdb, SDB_TOPIC, pShow->pIter, (void **)&pTopic);
     if (pShow->pIter == NULL) break;
 
-    int32_t cols = 0;
+    SColumnInfoData *pColInfo;
+    SName            n;
+    int32_t          cols = 0;
 
     char topicName[TSDB_TOPIC_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
-
-    SName n;
-    tNameFromString(&n, pTopic->name, T_NAME_ACCT|T_NAME_DB);
+    tNameFromString(&n, pTopic->name, T_NAME_ACCT | T_NAME_DB);
     tNameGetDbName(&n, varDataVal(topicName));
     varDataSetLen(topicName, strlen(varDataVal(topicName)));
-
-    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, numOfRows, (const char *)topicName, false);
 
+    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)));
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-    colDataAppend(pColInfo, numOfRows, (const char *)&pTopic->createTime, false);
+    colDataAppend(pColInfo, numOfRows, (const char *)dbName, false);
 
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)&pTopic->createTime, false);
+
     char sql[TSDB_SHOW_SQL_LEN + VARSTR_HEADER_SIZE] = {0};
     tstrncpy(&sql[VARSTR_HEADER_SIZE], pTopic->sql, TSDB_SHOW_SQL_LEN);
     varDataSetLen(sql, strlen(&sql[VARSTR_HEADER_SIZE]));
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, numOfRows, (const char *)sql, false);
 
-//    taosMemoryFree(sql);
-
     numOfRows++;
     sdbRelease(pSdb, pTopic);
   }

source/libs/wal/src/walWrite.c

@@ -251,11 +251,14 @@ static int walWriteIndex(SWal *pWal, int64_t ver, int64_t offset) {
 
 int64_t walWriteWithSyncInfo(SWal *pWal, int64_t index, tmsg_t msgType, SSyncLogMeta syncMeta, const void *body,
                              int32_t bodyLen) {
-  if (pWal == NULL) return -1;
   int code = 0;
 
   // no wal
   if (pWal->cfg.level == TAOS_WAL_NOLOG) return 0;
+  if (bodyLen > TSDB_MAX_WAL_SIZE) {
+    terrno = TSDB_CODE_WAL_SIZE_LIMIT;
+    return -1;
+  }
 
   if (index == pWal->vers.lastVer + 1) {
     if (taosArrayGetSize(pWal->fileInfoSet) == 0) {

source/util/src/terror.c

@@ -270,8 +270,10 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_TRANS_NOT_EXIST,          "Transaction not exist
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_TRANS_INVALID_STAGE,      "Invalid stage to kill")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_TRANS_CANT_PARALLEL,      "Invalid stage to kill")
 
-// mnode-topic
+// mnode-mq
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_UNSUPPORTED_TOPIC,        "Topic with aggregation is unsupported")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_READY,       "Consumer waiting for rebalance")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_EXIST,       "Consumer not exist")
 
 // mnode-sma
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_SMA_ALREADY_EXIST,        "SMA already exists")