Merge branch '3.0' into feat/agg_client_api

include/common/systable.h

@@ -35,6 +35,7 @@ extern "C" {
 #define TSDB_INS_TABLE_USER_INDEXES           "user_indexes"
 #define TSDB_INS_TABLE_USER_STABLES           "user_stables"
 #define TSDB_INS_TABLE_USER_TABLES            "user_tables"
+#define TSDB_INS_TABLE_USER_TAGS              "user_tags"
 #define TSDB_INS_TABLE_USER_TABLE_DISTRIBUTED "user_table_distributed"
 #define TSDB_INS_TABLE_USER_USERS             "user_users"
 #define TSDB_INS_TABLE_LICENCES               "grants"

include/common/tcommon.h

@@ -268,6 +268,26 @@ typedef struct SSortExecInfo {
   int32_t readBytes;   // read io bytes
 } SSortExecInfo;
 
+//======================================================================================================================
+// for grant 
+typedef enum {
+  TSDB_GRANT_ALL,
+  TSDB_GRANT_TIME,
+  TSDB_GRANT_USER,
+  TSDB_GRANT_DB,
+  TSDB_GRANT_TIMESERIES,
+  TSDB_GRANT_DNODE,
+  TSDB_GRANT_ACCT,
+  TSDB_GRANT_STORAGE,
+  TSDB_GRANT_SPEED,
+  TSDB_GRANT_QUERY_TIME,
+  TSDB_GRANT_CONNS,
+  TSDB_GRANT_STREAMS,
+  TSDB_GRANT_CPU_CORES,
+} EGrantType;
+
+int32_t grantCheck(EGrantType grant);
+
 #ifdef __cplusplus
 }
 #endif

include/common/tglobal.h

@@ -139,6 +139,7 @@ extern int32_t tsTransPullupInterval;
 extern int32_t tsMqRebalanceInterval;
 extern int32_t tsTtlUnit;
 extern int32_t tsTtlPushInterval;
+extern int32_t tsGrantHBInterval;
 
 #define NEEDTO_COMPRESSS_MSG(size) (tsCompressMsgSize != -1 && (size) > tsCompressMsgSize)
 

include/common/tmsg.h

@@ -75,13 +75,18 @@ typedef uint16_t tmsg_t;
 #define TSDB_IE_TYPE_DNODE_EXT   6
 #define TSDB_IE_TYPE_DNODE_STATE 7
 
-enum { CONN_TYPE__QUERY = 1, CONN_TYPE__TMQ, CONN_TYPE__UDFD, CONN_TYPE__MAX };
+enum {
+  CONN_TYPE__QUERY = 1,
+  CONN_TYPE__TMQ,
+  CONN_TYPE__UDFD,
+  CONN_TYPE__MAX,
+};
 
 enum {
   HEARTBEAT_KEY_USER_AUTHINFO = 1,
   HEARTBEAT_KEY_DBINFO,
   HEARTBEAT_KEY_STBINFO,
-  HEARTBEAT_KEY_MQ_TMP,
+  HEARTBEAT_KEY_TMQ,
 };
 
 typedef enum _mgmt_table {
@@ -99,6 +104,7 @@ typedef enum _mgmt_table {
   TSDB_MGMT_TABLE_STB,
   TSDB_MGMT_TABLE_STREAMS,
   TSDB_MGMT_TABLE_TABLE,
+  TSDB_MGMT_TABLE_TAG,
   TSDB_MGMT_TABLE_USER,
   TSDB_MGMT_TABLE_GRANTS,
   TSDB_MGMT_TABLE_VGROUP,
@@ -2146,6 +2152,15 @@ typedef struct {
   char    cgroup[TSDB_CGROUP_LEN];
 } SMqAskEpReq;
 
+typedef struct {
+  int64_t consumerId;
+  int32_t epoch;
+} SMqHbReq;
+
+typedef struct {
+  int8_t reserved;
+} SMqHbRsp;
+
 typedef struct {
   int32_t key;
   int32_t valueLen;
@@ -2335,29 +2350,30 @@ static FORCE_INLINE int32_t tDecodeSClientHbKey(SDecoder* pDecoder, SClientHbKey
   return 0;
 }
 
-typedef struct SMqHbVgInfo {
+typedef struct {
   int32_t vgId;
-} SMqHbVgInfo;
+  // TODO stas
+} SMqReportVgInfo;
 
-static FORCE_INLINE int32_t taosEncodeSMqVgInfo(void** buf, const SMqHbVgInfo* pVgInfo) {
+static FORCE_INLINE int32_t taosEncodeSMqVgInfo(void** buf, const SMqReportVgInfo* pVgInfo) {
   int32_t tlen = 0;
   tlen += taosEncodeFixedI32(buf, pVgInfo->vgId);
   return tlen;
 }
 
-static FORCE_INLINE void* taosDecodeSMqVgInfo(void* buf, SMqHbVgInfo* pVgInfo) {
+static FORCE_INLINE void* taosDecodeSMqVgInfo(void* buf, SMqReportVgInfo* pVgInfo) {
   buf = taosDecodeFixedI32(buf, &pVgInfo->vgId);
   return buf;
 }
 
-typedef struct SMqHbTopicInfo {
+typedef struct {
   int32_t epoch;
   int64_t topicUid;
   char    name[TSDB_TOPIC_FNAME_LEN];
   SArray* pVgInfo;  // SArray<SMqHbVgInfo>
-} SMqHbTopicInfo;
+} SMqTopicInfo;
 
-static FORCE_INLINE int32_t taosEncodeSMqHbTopicInfoMsg(void** buf, const SMqHbTopicInfo* pTopicInfo) {
+static FORCE_INLINE int32_t taosEncodeSMqTopicInfoMsg(void** buf, const SMqTopicInfo* pTopicInfo) {
   int32_t tlen = 0;
   tlen += taosEncodeFixedI32(buf, pTopicInfo->epoch);
   tlen += taosEncodeFixedI64(buf, pTopicInfo->topicUid);
@@ -2365,35 +2381,35 @@ static FORCE_INLINE int32_t taosEncodeSMqHbTopicInfoMsg(void** buf, const SMqHbT
   int32_t sz = taosArrayGetSize(pTopicInfo->pVgInfo);
   tlen += taosEncodeFixedI32(buf, sz);
   for (int32_t i = 0; i < sz; i++) {
-    SMqHbVgInfo* pVgInfo = (SMqHbVgInfo*)taosArrayGet(pTopicInfo->pVgInfo, i);
+    SMqReportVgInfo* pVgInfo = (SMqReportVgInfo*)taosArrayGet(pTopicInfo->pVgInfo, i);
     tlen += taosEncodeSMqVgInfo(buf, pVgInfo);
   }
   return tlen;
 }
 
-static FORCE_INLINE void* taosDecodeSMqHbTopicInfoMsg(void* buf, SMqHbTopicInfo* pTopicInfo) {
+static FORCE_INLINE void* taosDecodeSMqTopicInfoMsg(void* buf, SMqTopicInfo* pTopicInfo) {
   buf = taosDecodeFixedI32(buf, &pTopicInfo->epoch);
   buf = taosDecodeFixedI64(buf, &pTopicInfo->topicUid);
   buf = taosDecodeStringTo(buf, pTopicInfo->name);
   int32_t sz;
   buf = taosDecodeFixedI32(buf, &sz);
-  pTopicInfo->pVgInfo = taosArrayInit(sz, sizeof(SMqHbVgInfo));
+  pTopicInfo->pVgInfo = taosArrayInit(sz, sizeof(SMqReportVgInfo));
   for (int32_t i = 0; i < sz; i++) {
-    SMqHbVgInfo vgInfo;
+    SMqReportVgInfo vgInfo;
     buf = taosDecodeSMqVgInfo(buf, &vgInfo);
     taosArrayPush(pTopicInfo->pVgInfo, &vgInfo);
   }
   return buf;
 }
 
-typedef struct SMqHbMsg {
+typedef struct {
   int32_t status;  // ask hb endpoint
   int32_t epoch;
   int64_t consumerId;
   SArray* pTopics;  // SArray<SMqHbTopicInfo>
-} SMqHbMsg;
+} SMqReportReq;
 
-static FORCE_INLINE int32_t taosEncodeSMqMsg(void** buf, const SMqHbMsg* pMsg) {
+static FORCE_INLINE int32_t taosEncodeSMqReportMsg(void** buf, const SMqReportReq* pMsg) {
   int32_t tlen = 0;
   tlen += taosEncodeFixedI32(buf, pMsg->status);
   tlen += taosEncodeFixedI32(buf, pMsg->epoch);
@@ -2401,22 +2417,22 @@ static FORCE_INLINE int32_t taosEncodeSMqMsg(void** buf, const SMqHbMsg* pMsg) {
   int32_t sz = taosArrayGetSize(pMsg->pTopics);
   tlen += taosEncodeFixedI32(buf, sz);
   for (int32_t i = 0; i < sz; i++) {
-    SMqHbTopicInfo* topicInfo = (SMqHbTopicInfo*)taosArrayGet(pMsg->pTopics, i);
-    tlen += taosEncodeSMqHbTopicInfoMsg(buf, topicInfo);
+    SMqTopicInfo* topicInfo = (SMqTopicInfo*)taosArrayGet(pMsg->pTopics, i);
+    tlen += taosEncodeSMqTopicInfoMsg(buf, topicInfo);
   }
   return tlen;
 }
 
-static FORCE_INLINE void* taosDecodeSMqMsg(void* buf, SMqHbMsg* pMsg) {
+static FORCE_INLINE void* taosDecodeSMqReportMsg(void* buf, SMqReportReq* pMsg) {
   buf = taosDecodeFixedI32(buf, &pMsg->status);
   buf = taosDecodeFixedI32(buf, &pMsg->epoch);
   buf = taosDecodeFixedI64(buf, &pMsg->consumerId);
   int32_t sz;
   buf = taosDecodeFixedI32(buf, &sz);
-  pMsg->pTopics = taosArrayInit(sz, sizeof(SMqHbTopicInfo));
+  pMsg->pTopics = taosArrayInit(sz, sizeof(SMqTopicInfo));
   for (int32_t i = 0; i < sz; i++) {
-    SMqHbTopicInfo topicInfo;
-    buf = taosDecodeSMqHbTopicInfoMsg(buf, &topicInfo);
+    SMqTopicInfo topicInfo;
+    buf = taosDecodeSMqTopicInfoMsg(buf, &topicInfo);
     taosArrayPush(pMsg->pTopics, &topicInfo);
   }
   return buf;
@@ -2921,89 +2937,6 @@ typedef struct {
 int32_t tEncodeSMqDataRsp(SEncoder* pEncoder, const SMqDataRsp* pRsp);
 int32_t tDecodeSMqDataRsp(SDecoder* pDecoder, SMqDataRsp* pRsp);
 
-#if 0
-typedef struct {
-  SMqRspHead head;
-  int64_t    reqOffset;
-  int64_t    rspOffset;
-  int32_t    skipLogNum;
-  int32_t    blockNum;
-  int8_t     withTbName;
-  int8_t     withSchema;
-  SArray*    blockDataLen;  // SArray<int32_t>
-  SArray*    blockData;     // SArray<SRetrieveTableRsp*>
-  SArray*    blockTbName;   // SArray<char*>
-  SArray*    blockSchema;   // SArray<SSchemaWrapper>
-} SMqDataBlkRsp;
-
-static FORCE_INLINE int32_t tEncodeSMqDataBlkRsp(void** buf, const SMqDataBlkRsp* pRsp) {
-  int32_t tlen = 0;
-  tlen += taosEncodeFixedI64(buf, pRsp->reqOffset);
-  tlen += taosEncodeFixedI64(buf, pRsp->rspOffset);
-  tlen += taosEncodeFixedI32(buf, pRsp->skipLogNum);
-  tlen += taosEncodeFixedI32(buf, pRsp->blockNum);
-  if (pRsp->blockNum != 0) {
-    tlen += taosEncodeFixedI8(buf, pRsp->withTbName);
-    tlen += taosEncodeFixedI8(buf, pRsp->withSchema);
-
-    for (int32_t i = 0; i < pRsp->blockNum; i++) {
-      int32_t bLen = *(int32_t*)taosArrayGet(pRsp->blockDataLen, i);
-      void*   data = taosArrayGetP(pRsp->blockData, i);
-      tlen += taosEncodeFixedI32(buf, bLen);
-      tlen += taosEncodeBinary(buf, data, bLen);
-      if (pRsp->withSchema) {
-        SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(pRsp->blockSchema, i);
-        tlen += taosEncodeSSchemaWrapper(buf, pSW);
-      }
-      if (pRsp->withTbName) {
-        char* tbName = (char*)taosArrayGetP(pRsp->blockTbName, i);
-        tlen += taosEncodeString(buf, tbName);
-      }
-    }
-  }
-  return tlen;
-}
-
-static FORCE_INLINE void* tDecodeSMqDataBlkRsp(const void* buf, SMqDataBlkRsp* pRsp) {
-  buf = taosDecodeFixedI64(buf, &pRsp->reqOffset);
-  buf = taosDecodeFixedI64(buf, &pRsp->rspOffset);
-  buf = taosDecodeFixedI32(buf, &pRsp->skipLogNum);
-  buf = taosDecodeFixedI32(buf, &pRsp->blockNum);
-  if (pRsp->blockNum != 0) {
-    pRsp->blockData = taosArrayInit(pRsp->blockNum, sizeof(void*));
-    pRsp->blockDataLen = taosArrayInit(pRsp->blockNum, sizeof(int32_t));
-    buf = taosDecodeFixedI8(buf, &pRsp->withTbName);
-    buf = taosDecodeFixedI8(buf, &pRsp->withSchema);
-    if (pRsp->withTbName) {
-      pRsp->blockTbName = taosArrayInit(pRsp->blockNum, sizeof(void*));
-    }
-    if (pRsp->withSchema) {
-      pRsp->blockSchema = taosArrayInit(pRsp->blockNum, sizeof(void*));
-    }
-
-    for (int32_t i = 0; i < pRsp->blockNum; i++) {
-      int32_t bLen = 0;
-      void*   data = NULL;
-      buf = taosDecodeFixedI32(buf, &bLen);
-      buf = taosDecodeBinary(buf, &data, bLen);
-      taosArrayPush(pRsp->blockDataLen, &bLen);
-      taosArrayPush(pRsp->blockData, &data);
-      if (pRsp->withSchema) {
-        SSchemaWrapper* pSW = (SSchemaWrapper*)taosMemoryMalloc(sizeof(SSchemaWrapper));
-        buf = taosDecodeSSchemaWrapper(buf, pSW);
-        taosArrayPush(pRsp->blockSchema, &pSW);
-      }
-      if (pRsp->withTbName) {
-        char* name = NULL;
-        buf = taosDecodeString(buf, &name);
-        taosArrayPush(pRsp->blockTbName, &name);
-      }
-    }
-  }
-  return (void*)buf;
-}
-#endif
-
 typedef struct {
   SMqRspHead head;
   char       cgroup[TSDB_CGROUP_LEN];

include/common/tmsgdef.h

@@ -144,6 +144,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_ASK_EP, "ask-ep", SMqAskEpReq, SMqAskEpRsp)
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_LOST, "consumer-lost", SMqConsumerLostMsg, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_RECOVER, "consumer-recover", SMqConsumerRecoverMsg, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_HB, "consumer-hb", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_DO_REBALANCE, "do-rebalance", SMqDoRebalanceMsg, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_DROP_CGROUP, "drop-cgroup", SMqDropCGroupReq, SMqDropCGroupRsp)
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_COMMIT_OFFSET, "mnode-commit-offset", SMqCMCommitOffsetReq, SMqCMCommitOffsetRsp)
@@ -151,6 +152,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_TELEM_TIMER, "telem-tmr", SMTimerReq, SMTimerReq)
   TD_DEF_MSG_TYPE(TDMT_MND_TRANS_TIMER, "trans-tmr", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_TTL_TIMER, "ttl-tmr", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_GRANT_HB_TIMER, "grant-hb-tmr", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_KILL_TRANS, "kill-trans", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_KILL_QUERY, "kill-query", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_KILL_CONN, "kill-conn", NULL, NULL)

include/libs/nodes/nodes.h

@@ -170,6 +170,7 @@ typedef enum ENodeType {
   QUERY_NODE_SHOW_STABLES_STMT,
   QUERY_NODE_SHOW_STREAMS_STMT,
   QUERY_NODE_SHOW_TABLES_STMT,
+  QUERY_NODE_SHOW_TAGS_STMT,
   QUERY_NODE_SHOW_USERS_STMT,
   QUERY_NODE_SHOW_LICENCE_STMT,
   QUERY_NODE_SHOW_VGROUPS_STMT,

include/libs/transport/trpc.h

@@ -135,7 +135,7 @@ void    rpcSendRequestWithCtx(void *thandle, const SEpSet *pEpSet, SRpcMsg *pMsg
 int32_t rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo);
 void    rpcSendRecv(void *shandle, SEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp);
 void    rpcSetDefaultAddr(void *thandle, const char *ip, const char *fqdn);
-int64_t rpcAllocHandle();
+void*   rpcAllocHandle();
 
 #ifdef __cplusplus
 }

include/util/taoserror.h

@@ -332,7 +332,6 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_VND_INVALID_TABLE_ACTION      TAOS_DEF_ERROR_CODE(0, 0x0519)
 #define TSDB_CODE_VND_COL_ALREADY_EXISTS        TAOS_DEF_ERROR_CODE(0, 0x051a)
 #define TSDB_CODE_VND_TABLE_COL_NOT_EXISTS      TAOS_DEF_ERROR_CODE(0, 0x051b)
-#define TSDB_CODE_VND_READ_END                  TAOS_DEF_ERROR_CODE(0, 0x051c)
 
 // tsdb
 #define TSDB_CODE_TDB_INVALID_TABLE_ID          TAOS_DEF_ERROR_CODE(0, 0x0600)

source/client/src/clientHb.c

@@ -171,6 +171,7 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
       pTscObj->pAppInfo->totalDnodes = pRsp->query->totalDnodes;
       pTscObj->pAppInfo->onlineDnodes = pRsp->query->onlineDnodes;
       pTscObj->connId = pRsp->query->connId;
+      tscTrace("conn %p hb rsp, dnodes %d/%d", pTscObj->connId, pTscObj->pAppInfo->onlineDnodes, pTscObj->pAppInfo->totalDnodes);
 
       if (pRsp->query->killRid) {
         tscDebug("request rid %" PRIx64 " need to be killed now", pRsp->query->killRid);
@@ -294,6 +295,7 @@ static int32_t hbAsyncCallBack(void *param, SDataBuf *pMsg, int32_t code) {
 
   if (code != 0) {
     (*pInst)->onlineDnodes = ((*pInst)->totalDnodes ? 0 : -1);
+    tscDebug("hb rsp error %s, update server status %d/%d", tstrerror(code), (*pInst)->onlineDnodes, (*pInst)->totalDnodes);
   }
 
   if (rspNum) {
@@ -571,7 +573,8 @@ int32_t hbQueryHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req
   return TSDB_CODE_SUCCESS;
 }
 
-void hbMgrInitMqHbHandle() {
+static FORCE_INLINE void hbMgrInitHandle() {
+  // init all handle
   clientHbMgr.reqHandle[CONN_TYPE__QUERY] = hbQueryHbReqHandle;
   clientHbMgr.reqHandle[CONN_TYPE__TMQ] = hbMqHbReqHandle;
 
@@ -579,11 +582,6 @@ void hbMgrInitMqHbHandle() {
   clientHbMgr.rspHandle[CONN_TYPE__TMQ] = hbMqHbRspHandle;
 }
 
-static FORCE_INLINE void hbMgrInitHandle() {
-  // init all handle
-  hbMgrInitMqHbHandle();
-}
-
 SClientHbBatchReq *hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
   SClientHbBatchReq *pBatchReq = taosMemoryCalloc(1, sizeof(SClientHbBatchReq));
   if (pBatchReq == NULL) {

source/client/src/clientImpl.c

@@ -1276,7 +1276,12 @@ int32_t doProcessMsgFromServer(void* param) {
   assert(pMsg->info.ahandle != NULL);
   STscObj* pTscObj = NULL;
 
-  tscDebug("processMsgFromServer message: %s, code: %s", TMSG_INFO(pMsg->msgType), tstrerror(pMsg->code));
+  STraceId* trace = &pMsg->info.traceId;
+  char      tbuf[40] = {0};
+  TRACE_TO_STR(trace, tbuf);
+
+  tscDebug("processMsgFromServer handle %p, message: %s, code: %s, gtid: %s", pMsg->info.handle, TMSG_INFO(pMsg->msgType), tstrerror(pMsg->code),
+           tbuf);
 
   if (pSendInfo->requestObjRefId != 0) {
     SRequestObj* pRequest = (SRequestObj*)taosAcquireRef(clientReqRefPool, pSendInfo->requestObjRefId);

source/common/src/systable.c

@@ -153,6 +153,15 @@ static const SSysDbTableSchema userTblsSchema[] = {
     {.name = "type", .bytes = 21 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
 };
 
+static const SSysDbTableSchema userTagsSchema[] = {
+    {.name = "table_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
+    {.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
+    {.name = "stable_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
+    {.name = "tag_name", .bytes = TSDB_COL_NAME_LEN - 1 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
+    {.name = "tag_type", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT},
+    {.name = "tag_value", .bytes = TSDB_MAX_TAGS_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
+};
+
 static const SSysDbTableSchema userTblDistSchema[] = {
     {.name = "db_name", .bytes = 32 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
     {.name = "table_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
@@ -255,6 +264,7 @@ static const SSysTableMeta infosMeta[] = {
     {TSDB_INS_TABLE_USER_STABLES, userStbsSchema, tListLen(userStbsSchema)},
     {TSDB_PERFS_TABLE_STREAMS, streamSchema, tListLen(streamSchema)},
     {TSDB_INS_TABLE_USER_TABLES, userTblsSchema, tListLen(userTblsSchema)},
+    {TSDB_INS_TABLE_USER_TAGS, userTagsSchema, tListLen(userTagsSchema)},
     // {TSDB_INS_TABLE_USER_TABLE_DISTRIBUTED, userTblDistSchema, tListLen(userTblDistSchema)},
     {TSDB_INS_TABLE_USER_USERS, userUsersSchema, tListLen(userUsersSchema)},
     {TSDB_INS_TABLE_LICENCES, grantsSchema, tListLen(grantsSchema)},

source/common/src/tdatablock.c

@@ -1752,7 +1752,7 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf)
   int32_t colNum = taosArrayGetSize(pDataBlock->pDataBlock);
   int32_t rows = pDataBlock->info.rows;
   int32_t len = 0;
-  len += snprintf(dumpBuf + len, size - len, "%s |block type %d |child id %d|group id:%" PRIu64 "| uid:%ld|======\n", "dumpBlockData",
+  len += snprintf(dumpBuf + len, size - len, "===stream===%s |block type %d |child id %d|group id:%" PRIu64 "| uid:%ld|\n", flag,
                   (int32_t)pDataBlock->info.type, pDataBlock->info.childId, pDataBlock->info.groupId,
                   pDataBlock->info.uid);
   if (len >= size - 1) return dumpBuf;

source/common/src/tglobal.c

@@ -184,6 +184,7 @@ int32_t tsTransPullupInterval = 2;
 int32_t tsMqRebalanceInterval = 2;
 int32_t tsTtlUnit = 86400;
 int32_t tsTtlPushInterval = 60;
+int32_t tsGrantHBInterval = 60;
 
 void taosAddDataDir(int32_t index, char *v1, int32_t level, int32_t primary) {
   tstrncpy(tsDiskCfg[index].dir, v1, TSDB_FILENAME_LEN);

source/dnode/mgmt/mgmt_dnode/inc/dmInt.h

@@ -46,6 +46,7 @@ int32_t dmProcessAuthRsp(SDnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t dmProcessGrantRsp(SDnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t dmProcessServerRunStatus(SDnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t dmProcessRetrieve(SDnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t dmProcessGrantReq(SRpcMsg *pMsg);
 
 // dmWorker.c
 int32_t dmPutNodeMsgToMgmtQueue(SDnodeMgmt *pMgmt, SRpcMsg *pMsg);

source/dnode/mgmt/mgmt_dnode/src/dmHandle.c

@@ -331,7 +331,8 @@ SArray *dmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_DND_SYSTABLE_RETRIEVE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
 
   // Requests handled by MNODE
-  if (dmSetMgmtHandle(pArray, TDMT_MND_GRANT_RSP, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_GRANT, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
+  // if (dmSetMgmtHandle(pArray, TDMT_MND_GRANT_RSP, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_AUTH_RSP, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
 
   code = 0;

source/dnode/mgmt/mgmt_dnode/src/dmWorker.c

@@ -144,6 +144,9 @@ static void dmProcessMgmtQueue(SQueueInfo *pInfo, SRpcMsg *pMsg) {
     case TDMT_DND_SYSTABLE_RETRIEVE:
       code = dmProcessRetrieve(pMgmt, pMsg);
       break;
+    case TDMT_MND_GRANT:
+      code = dmProcessGrantReq(pMsg);
+      break;
     default:
       terrno = TSDB_CODE_MSG_NOT_PROCESSED;
       break;

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

@@ -196,6 +196,7 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_TOPIC, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_SUBSCRIBE, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_MQ_ASK_EP, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_MQ_HB, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_MQ_DROP_CGROUP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_MQ_DROP_CGROUP_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_MQ_COMMIT_OFFSET, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
@@ -205,7 +206,7 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_MND_HEARTBEAT, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_STATUS, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_SYSTABLE_RETRIEVE, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_MND_GRANT, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  // if (dmSetMgmtHandle(pArray, TDMT_MND_GRANT, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_AUTH, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_SHOW_VARIABLES, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_SERVER_VERSION, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;

source/dnode/mgmt/mgmt_qnode/src/qmWorker.c

@@ -81,6 +81,7 @@ int32_t qmPutRpcMsgToQueue(SQnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) {
       taosWriteQitem(pMgmt->queryWorker.queue, pMsg);
       return 0;
     case READ_QUEUE:
+    case FETCH_QUEUE:
       dTrace("msg:%p, is created and will put into qnode-fetch queue", pMsg);
       taosWriteQitem(pMgmt->fetchWorker.queue, pMsg);
       return 0;

source/dnode/mgmt/mgmt_vnode/src/vmWorker.c

@@ -153,9 +153,15 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
 
   switch (qtype) {
     case QUERY_QUEUE:
-      vnodePreprocessQueryMsg(pVnode->pImpl, pMsg);
-      dGTrace("vgId:%d, msg:%p put into vnode-query queue", pVnode->vgId, pMsg);
-      taosWriteQitem(pVnode->pQueryQ, pMsg);
+      if ((pMsg->msgType == TDMT_SCH_QUERY) && (grantCheck(TSDB_GRANT_TIME) != TSDB_CODE_SUCCESS)) {
+        terrno = TSDB_CODE_GRANT_EXPIRED;
+        code = terrno;
+        dDebug("vgId:%d, msg:%p put into vnode-query queue failed since %s", pVnode->vgId, pMsg, terrstr());
+      } else {
+        vnodePreprocessQueryMsg(pVnode->pImpl, pMsg);
+        dGTrace("vgId:%d, msg:%p put into vnode-query queue", pVnode->vgId, pMsg);
+        taosWriteQitem(pVnode->pQueryQ, pMsg);
+      }
       break;
     case STREAM_QUEUE:
       dGTrace("vgId:%d, msg:%p put into vnode-stream queue", pVnode->vgId, pMsg);
@@ -166,8 +172,14 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
       taosWriteQitem(pVnode->pFetchQ, pMsg);
       break;
     case WRITE_QUEUE:
-      dGTrace("vgId:%d, msg:%p put into vnode-write queue", pVnode->vgId, pMsg);
-      taosWriteQitem(pVnode->pWriteQ, pMsg);
+      if ((pMsg->msgType == TDMT_VND_SUBMIT) && (grantCheck(TSDB_GRANT_STORAGE) != TSDB_CODE_SUCCESS)) {
+        terrno = TSDB_CODE_VND_NO_WRITE_AUTH;
+        code = terrno;
+        dDebug("vgId:%d, msg:%p put into vnode-write queue failed since %s", pVnode->vgId, pMsg, terrstr());
+      } else {
+        dGTrace("vgId:%d, msg:%p put into vnode-write queue", pVnode->vgId, pMsg);
+        taosWriteQitem(pVnode->pWriteQ, pMsg);
+      }
       break;
     case SYNC_QUEUE:
       dGTrace("vgId:%d, msg:%p put into vnode-sync queue", pVnode->vgId, pMsg);

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

@@ -26,6 +26,7 @@ int32_t mndInitCluster(SMnode *pMnode);
 void    mndCleanupCluster(SMnode *pMnode);
 int32_t mndGetClusterName(SMnode *pMnode, char *clusterName, int32_t len);
 int64_t mndGetClusterId(SMnode *pMnode);
+int64_t mndGetClusterCreateTime(SMnode *pMnode);
 
 #ifdef __cplusplus
 }

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

@@ -29,6 +29,7 @@ void       mndReleaseDnode(SMnode *pMnode, SDnodeObj *pDnode);
 SEpSet     mndGetDnodeEpset(SDnodeObj *pDnode);
 int32_t    mndGetDnodeSize(SMnode *pMnode);
 bool       mndIsDnodeOnline(SDnodeObj *pDnode, int64_t curMs);
+void       mndGetDnodeData(SMnode *pMnode, SArray *pDnodeEps);
 
 #ifdef __cplusplus
 }

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

@@ -22,27 +22,10 @@
 
 #include "mndInt.h"
 
-typedef enum {
-  TSDB_GRANT_ALL,
-  TSDB_GRANT_TIME,
-  TSDB_GRANT_USER,
-  TSDB_GRANT_DB,
-  TSDB_GRANT_TIMESERIES,
-  TSDB_GRANT_DNODE,
-  TSDB_GRANT_ACCT,
-  TSDB_GRANT_STORAGE,
-  TSDB_GRANT_SPEED,
-  TSDB_GRANT_QUERY_TIME,
-  TSDB_GRANT_CONNS,
-  TSDB_GRANT_STREAMS,
-  TSDB_GRANT_CPU_CORES,
-} EGrantType;
-
-int32_t mndInitGrant();
+int32_t mndInitGrant(SMnode *pMnode);
 void    mndCleanupGrant();
 void    grantParseParameter();
-int32_t grantCheck(EGrantType grant);
-void    grantReset(EGrantType grant, uint64_t value);
+void    grantReset(SMnode *pMnode, EGrantType grant, uint64_t value);
 void    grantAdd(EGrantType grant, uint64_t value);
 void    grantRestore(EGrantType grant, uint64_t value);
 

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

@@ -79,6 +79,23 @@ int64_t mndGetClusterId(SMnode *pMnode) {
   return clusterId;
 }
 
+int64_t mndGetClusterCreateTime(SMnode *pMnode) {
+  SSdb   *pSdb = pMnode->pSdb;
+  void   *pIter = NULL;
+  int64_t createTime = INT64_MAX;
+
+  while (1) {
+    SClusterObj *pCluster = NULL;
+    pIter = sdbFetch(pSdb, SDB_CLUSTER, pIter, (void **)&pCluster);
+    if (pIter == NULL) break;
+
+    createTime = pCluster->createdTime;
+    sdbRelease(pSdb, pCluster);
+  }
+
+  return createTime;
+}
+
 static SSdbRaw *mndClusterActionEncode(SClusterObj *pCluster) {
   terrno = TSDB_CODE_OUT_OF_MEMORY;
 

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

@@ -48,6 +48,7 @@ static void    mndCancelGetNextConsumer(SMnode *pMnode, void *pIter);
 
 static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg);
 static int32_t mndProcessAskEpReq(SRpcMsg *pMsg);
+static int32_t mndProcessMqHbReq(SRpcMsg *pMsg);
 static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg);
 static int32_t mndProcessConsumerLostMsg(SRpcMsg *pMsg);
 static int32_t mndProcessConsumerRecoverMsg(SRpcMsg *pMsg);
@@ -62,6 +63,7 @@ int32_t mndInitConsumer(SMnode *pMnode) {
                      .deleteFp = (SdbDeleteFp)mndConsumerActionDelete};
 
   mndSetMsgHandle(pMnode, TDMT_MND_SUBSCRIBE, mndProcessSubscribeReq);
+  mndSetMsgHandle(pMnode, TDMT_MND_MQ_HB, mndProcessMqHbReq);
   mndSetMsgHandle(pMnode, TDMT_MND_MQ_ASK_EP, mndProcessAskEpReq);
   mndSetMsgHandle(pMnode, TDMT_MND_MQ_TIMER, mndProcessMqTimerMsg);
   mndSetMsgHandle(pMnode, TDMT_MND_MQ_CONSUMER_LOST, mndProcessConsumerLostMsg);
@@ -255,6 +257,33 @@ static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg) {
   return 0;
 }
 
+static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) {
+  SMnode   *pMnode = pMsg->info.node;
+  SMqHbReq *pReq = (SMqHbReq *)pMsg->pCont;
+  int64_t   consumerId = be64toh(pReq->consumerId);
+
+  SMqConsumerObj *pConsumer = mndAcquireConsumer(pMnode, consumerId);
+
+  atomic_store_32(&pConsumer->hbStatus, 0);
+
+  int32_t status = atomic_load_32(&pConsumer->status);
+
+  if (status == MQ_CONSUMER_STATUS__LOST_REBD) {
+    SMqConsumerRecoverMsg *pRecoverMsg = rpcMallocCont(sizeof(SMqConsumerRecoverMsg));
+
+    pRecoverMsg->consumerId = consumerId;
+    SRpcMsg *pRpcMsg = taosMemoryCalloc(1, sizeof(SRpcMsg));
+    pRpcMsg->msgType = TDMT_MND_MQ_CONSUMER_RECOVER;
+    pRpcMsg->pCont = pRecoverMsg;
+    pRpcMsg->contLen = sizeof(SMqConsumerRecoverMsg);
+    tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, pRpcMsg);
+  }
+
+  mndReleaseConsumer(pMnode, pConsumer);
+
+  return 0;
+}
+
 static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {
   SMnode      *pMnode = pMsg->info.node;
   SMqAskEpReq *pReq = (SMqAskEpReq *)pMsg->pCont;
@@ -262,19 +291,22 @@ static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {
   int64_t      consumerId = be64toh(pReq->consumerId);
   int32_t      epoch = ntohl(pReq->epoch);
 
-  SMqConsumerObj *pConsumer = mndAcquireConsumer(pMsg->info.node, consumerId);
+  SMqConsumerObj *pConsumer = mndAcquireConsumer(pMnode, consumerId);
   if (pConsumer == NULL) {
     terrno = TSDB_CODE_MND_CONSUMER_NOT_EXIST;
     return -1;
   }
 
   ASSERT(strcmp(pReq->cgroup, pConsumer->cgroup) == 0);
-  /*int32_t hbStatus = atomic_load_32(&pConsumer->hbStatus);*/
+
+#if 1
   atomic_store_32(&pConsumer->hbStatus, 0);
+#endif
 
   // 1. check consumer status
   int32_t status = atomic_load_32(&pConsumer->status);
 
+#if 1
   if (status == MQ_CONSUMER_STATUS__LOST_REBD) {
     SMqConsumerRecoverMsg *pRecoverMsg = rpcMallocCont(sizeof(SMqConsumerRecoverMsg));
 
@@ -285,6 +317,7 @@ static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {
     pRpcMsg->contLen = sizeof(SMqConsumerRecoverMsg);
     tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, pRpcMsg);
   }
+#endif
 
   if (status != MQ_CONSUMER_STATUS__READY) {
     terrno = TSDB_CODE_MND_CONSUMER_NOT_READY;

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

@@ -509,6 +509,12 @@ static int32_t mndProcessCreateDbReq(SRpcMsg *pReq) {
   SUserObj    *pUser = NULL;
   SCreateDbReq createReq = {0};
 
+  code = grantCheck(TSDB_GRANT_DB);
+  if (code != 0) {
+    terrno = code;
+    goto _OVER;
+  }
+
   if (tDeserializeSCreateDbReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     goto _OVER;

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

@@ -262,7 +262,7 @@ bool mndIsDnodeOnline(SDnodeObj *pDnode, int64_t curMs) {
   return true;
 }
 
-static void mndGetDnodeData(SMnode *pMnode, SArray *pDnodeEps) {
+void mndGetDnodeData(SMnode *pMnode, SArray *pDnodeEps) {
   SSdb *pSdb = pMnode->pSdb;
 
   int32_t numOfEps = 0;
@@ -621,6 +621,12 @@ static int32_t mndProcessCreateDnodeReq(SRpcMsg *pReq) {
   SDnodeObj      *pDnode = NULL;
   SCreateDnodeReq createReq = {0};
 
+  code = grantCheck(TSDB_GRANT_DNODE);
+  if (code != TSDB_CODE_SUCCESS) {
+    terrno = code;
+    goto _OVER;
+  }
+
   if (tDeserializeSCreateDnodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     goto _OVER;

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

@@ -118,17 +118,21 @@ static int32_t mndRetrieveGrant(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBl
   return numOfRows;
 }
 
+static int32_t mndProcessGrantHB(SRpcMsg *pReq) { return TSDB_CODE_SUCCESS; }
+
 int32_t mndInitGrant(SMnode *pMnode) {
   mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_GRANTS, mndRetrieveGrant);
+  mndSetMsgHandle(pMnode, TDMT_MND_GRANT_HB_TIMER, mndProcessGrantHB);
   return 0;
 }
 
 void    mndCleanupGrant() {}
 void    grantParseParameter() { mError("can't parsed parameter k"); }
 int32_t grantCheck(EGrantType grant) { return TSDB_CODE_SUCCESS; }
-void    grantReset(EGrantType grant, uint64_t value) {}
+void    grantReset(SMnode *pMnode, EGrantType grant, uint64_t value) {}
 void    grantAdd(EGrantType grant, uint64_t value) {}
 void    grantRestore(EGrantType grant, uint64_t value) {}
+int32_t dmProcessGrantReq(SRpcMsg *pMsg) { return TSDB_CODE_SUCCESS; }
 
 #endif
 

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

@@ -90,6 +90,16 @@ static void mndPullupTelem(SMnode *pMnode) {
   }
 }
 
+static void mndGrantHeartBeat(SMnode *pMnode) {
+  int32_t contLen = 0;
+  void   *pReq = mndBuildTimerMsg(&contLen);
+  if (pReq != NULL) {
+    SRpcMsg rpcMsg = {
+        .msgType = TDMT_MND_GRANT_HB_TIMER, .pCont = pReq, .contLen = contLen, .info.ahandle = (void *)0x9527};
+    tmsgPutToQueue(&pMnode->msgCb, READ_QUEUE, &rpcMsg);
+  }
+}
+
 static void *mndThreadFp(void *param) {
   SMnode *pMnode = param;
   int64_t lastTime = 0;
@@ -115,6 +125,10 @@ static void *mndThreadFp(void *param) {
     if (lastTime % (tsTelemInterval * 10) == 0) {
       mndPullupTelem(pMnode);
     }
+
+    if (lastTime % (tsGrantHBInterval * 10) == 0) {
+      mndGrantHeartBeat(pMnode);
+    }
   }
 
   return NULL;
@@ -402,6 +416,9 @@ int32_t mndStart(SMnode *pMnode) {
     }
     mndSetRestore(pMnode, true);
   }
+  
+  grantReset(pMnode, TSDB_GRANT_ALL, 0);
+
   return mndInitTimer(pMnode);
 }
 

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

@@ -388,67 +388,7 @@ static void mndCancelGetNextApp(SMnode *pMnode, void *pIter) {
 }
 
 static SClientHbRsp *mndMqHbBuildRsp(SMnode *pMnode, SClientHbReq *pReq) {
-#if 0
-  SClientHbRsp* pRsp = taosMemoryMalloc(sizeof(SClientHbRsp));
-  if (pRsp == NULL) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return NULL;
-  }
-  pRsp->connKey = pReq->connKey;
-  SMqHbBatchRsp batchRsp;
-  batchRsp.batchRsps = taosArrayInit(0, sizeof(SMqHbRsp));
-  if (batchRsp.batchRsps == NULL) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return NULL;
-  }
-  SClientHbKey connKey = pReq->connKey;
-  SHashObj* pObj =  pReq->info;
-  SKv* pKv = taosHashGet(pObj, "mq-tmp", strlen("mq-tmp") + 1);
-  if (pKv == NULL) {
-    taosMemoryFree(pRsp);
-    return NULL;
-  }
-  SMqHbMsg mqHb;
-  taosDecodeSMqMsg(pKv->value, &mqHb);
-  /*int64_t clientUid = htonl(pKv->value);*/
-  /*if (mqHb.epoch )*/
-  int sz = taosArrayGetSize(mqHb.pTopics);
-  SMqConsumerObj* pConsumer = mndAcquireConsumer(pMnode, mqHb.consumerId); 
-  for (int i = 0; i < sz; i++) {
-    SMqHbOneTopicBatchRsp innerBatchRsp;
-    innerBatchRsp.rsps = taosArrayInit(sz, sizeof(SMqHbRsp));
-    if (innerBatchRsp.rsps == NULL) {
-      //TODO
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
-      return NULL;
-    }
-    SMqHbTopicInfo* topicInfo = taosArrayGet(mqHb.pTopics, i);
-    SMqConsumerTopic* pConsumerTopic = taosHashGet(pConsumer->topicHash, topicInfo->name, strlen(topicInfo->name)+1);
-    if (pConsumerTopic->epoch != topicInfo->epoch) {
-      //add new vgids into rsp
-      int vgSz = taosArrayGetSize(topicInfo->pVgInfo);
-      for (int j = 0; j < vgSz; j++) {
-        SMqHbRsp innerRsp;
-        SMqHbVgInfo* pVgInfo = taosArrayGet(topicInfo->pVgInfo, i);
-        SVgObj* pVgObj = mndAcquireVgroup(pMnode, pVgInfo->vgId);
-        innerRsp.epSet = mndGetVgroupEpset(pMnode, pVgObj);
-        taosArrayPush(innerBatchRsp.rsps, &innerRsp);
-      }
-    }
-    taosArrayPush(batchRsp.batchRsps, &innerBatchRsp);
-  }
-  int32_t tlen = taosEncodeSMqHbBatchRsp(NULL, &batchRsp);
-  void* buf = taosMemoryMalloc(tlen);
-  if (buf == NULL) {
-    //TODO
-    return NULL;
-  }
-  void* abuf = buf;
-  taosEncodeSMqHbBatchRsp(&abuf, &batchRsp);
-  pRsp->body = buf;
-  pRsp->bodyLen = tlen;
-  return pRsp;
-#endif
+  //
   return NULL;
 }
 

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

@@ -76,6 +76,8 @@ static int32_t convertToRetrieveType(char *name, int32_t len) {
     type = TSDB_MGMT_TABLE_STB;
   } else if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLES, len) == 0) {
     type = TSDB_MGMT_TABLE_TABLE;
+  } else if (strncasecmp(name, TSDB_INS_TABLE_USER_TAGS, len) == 0) {
+    type = TSDB_MGMT_TABLE_TAG;
   } else if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLE_DISTRIBUTED, len) == 0) {
     //    type = TSDB_MGMT_TABLE_DIST;
   } else if (strncasecmp(name, TSDB_INS_TABLE_USER_USERS, len) == 0) {

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

@@ -709,7 +709,7 @@ static int32_t mndProcessCreateSmaReq(SRpcMsg *pReq) {
 
 _OVER:
   if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
-    mError("sma:%s, failed to create since %s", createReq.name, terrstr(terrno));
+    mError("sma:%s, failed to create since %s", createReq.name, terrstr());
   }
 
   mndReleaseStb(pMnode, pStb);

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

@@ -1689,6 +1689,9 @@ static int32_t mndAlterStb(SMnode *pMnode, SRpcMsg *pReq, const SMAlterStbReq *p
 _OVER:
   taosMemoryFreeClear(stbObj.pTags);
   taosMemoryFreeClear(stbObj.pColumns);
+  if (pAlter->commentLen > 0) {
+    taosMemoryFreeClear(stbObj.comment);
+  }
   return code;
 }
 
@@ -1733,7 +1736,7 @@ _OVER:
 
   mndReleaseStb(pMnode, pStb);
   mndReleaseDb(pMnode, pDb);
-  taosArrayDestroy(alterReq.pFields);
+  tFreeSMAltertbReq(&alterReq);
 
   return code;
 }

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

@@ -363,6 +363,12 @@ static int32_t mndProcessCreateUserReq(SRpcMsg *pReq) {
     goto _OVER;
   }
 
+  code = grantCheck(TSDB_GRANT_USER);
+  if (code != TSDB_CODE_SUCCESS) {
+    terrno = code;
+    goto _OVER;
+  }
+
   code = mndCreateUser(pMnode, pOperUser->acct, &createReq, pReq);
   if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
 

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

@@ -639,6 +639,7 @@ static int32_t mndRetrieveVgroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *p
     if (pShow->pIter == NULL) break;
 
     if (pDb != NULL && pVgroup->dbUid != pDb->uid) {
+      sdbRelease(pSdb, pVgroup);
       continue;
     }
 

source/dnode/vnode/inc/vnode.h

@@ -138,7 +138,7 @@ void   *tsdbGetIdx(SMeta *pMeta);
 void   *tsdbGetIvtIdx(SMeta *pMeta);
 
 int32_t tsdbLastRowReaderOpen(void *pVnode, int32_t type, SArray *pTableIdList, int32_t numOfCols, void **pReader);
-int32_t tsdbRetrieveLastRow(void *pReader, SSDataBlock *pResBlock, const int32_t *slotIds, SArray* pTableUids);
+int32_t tsdbRetrieveLastRow(void *pReader, SSDataBlock *pResBlock, const int32_t *slotIds, SArray *pTableUids);
 int32_t tsdbLastrowReaderClose(void *pReader);
 int32_t tsdbGetTableSchema(SVnode *pVnode, int64_t uid, STSchema **pSchema, int64_t *suid);
 

source/dnode/vnode/src/inc/sma.h

@@ -32,6 +32,8 @@ extern "C" {
 #define smaTrace(...) do { if (smaDebugFlag & DEBUG_TRACE) { taosPrintLog("SMA ", DEBUG_TRACE, tsdbDebugFlag, __VA_ARGS__); }} while(0)
 // clang-format on
 
+#define RSMA_TASK_INFO_HASH_SLOT 8
+
 typedef struct SSmaEnv       SSmaEnv;
 typedef struct SSmaStat      SSmaStat;
 typedef struct STSmaStat     STSmaStat;
@@ -41,7 +43,7 @@ typedef struct SRSmaInfo     SRSmaInfo;
 typedef struct SRSmaInfoItem SRSmaInfoItem;
 
 struct SSmaEnv {
-  TdThreadRwlock lock;
+  SRWLatch       lock;
   int8_t         type;
   SSmaStat      *pStat;
 };
@@ -52,7 +54,7 @@ typedef struct {
   void   *tmrHandle;  // shared by all fetch tasks
 } SSmaMgmt;
 
-#define SMA_ENV_LOCK(env) ((env)->lock)
+#define SMA_ENV_LOCK(env) (&(env)->lock)
 #define SMA_ENV_TYPE(env) ((env)->type)
 #define SMA_ENV_STAT(env) ((env)->pStat)
 
@@ -64,10 +66,14 @@ struct STSmaStat {
 
 struct SRSmaStat {
   SSma     *pSma;
-  int64_t   submitVer;
-  int64_t   refId;         // shared by fetch tasks
-  int8_t    triggerStat;   // shared by fetch tasks
-  SHashObj *rsmaInfoHash;  // key: stbUid, value: SRSmaInfo;
+  int64_t   commitAppliedVer;  // vnode applied version for async commit
+  int64_t   commitSubmitVer;   // rsma submit version for async commit
+  int64_t   submitVer;         // latest submit version
+  int64_t   refId;             // shared by fetch tasks
+  int8_t    triggerStat;       // shared by fetch tasks
+  int8_t    commitStat;        // 0 not in committing, 1 in committing
+  SHashObj *rsmaInfoHash;      // key: stbUid, value: SRSmaInfo;
+  SHashObj *iRsmaInfoHash;     // key: stbUid, value: SRSmaInfo; immutable rsmaInfoHash
 };
 
 struct SSmaStat {
@@ -78,12 +84,29 @@ struct SSmaStat {
   T_REF_DECLARE()
 };
 
-#define SMA_TSMA_STAT(s)     (&(s)->tsmaStat)
-#define SMA_RSMA_STAT(s)     (&(s)->rsmaStat)
-#define RSMA_INFO_HASH(r)    ((r)->rsmaInfoHash)
-#define RSMA_TRIGGER_STAT(r) (&(r)->triggerStat)
-#define RSMA_REF_ID(r)       ((r)->refId)
-#define RSMA_SUBMIT_VER(r)   ((r)->submitVer)
+#define SMA_TSMA_STAT(s)      (&(s)->tsmaStat)
+#define SMA_RSMA_STAT(s)      (&(s)->rsmaStat)
+#define RSMA_INFO_HASH(r)     ((r)->rsmaInfoHash)
+#define RSMA_IMU_INFO_HASH(r) ((r)->iRsmaInfoHash)
+#define RSMA_TRIGGER_STAT(r)  (&(r)->triggerStat)
+#define RSMA_COMMIT_STAT(r)   (&(r)->commitStat)
+#define RSMA_REF_ID(r)        ((r)->refId)
+#define RSMA_SUBMIT_VER(r)    ((r)->submitVer)
+
+struct SRSmaInfoItem {
+  void   *taskInfo;  // qTaskInfo_t
+  int64_t refId;
+  tmr_h   tmrId;
+  int32_t maxDelay;
+  int8_t  level;
+  int8_t  triggerStat;
+};
+
+struct SRSmaInfo {
+  STSchema     *pTSchema;
+  int64_t       suid;
+  SRSmaInfoItem items[TSDB_RETENTION_L2];
+};
 
 enum {
   TASK_TRIGGER_STAT_INIT = 0,
@@ -94,6 +117,14 @@ enum {
   TASK_TRIGGER_STAT_DROPPED = 5,
 };
 
+enum {
+  RSMA_ROLE_CREATE = 0,
+  RSMA_ROLE_DROP = 1,
+  RSMA_ROLE_FETCH = 2,
+  RSMA_ROLE_SUBMIT = 3,
+  RSMA_ROLE_ITERATE = 4,
+};
+
 void  tdDestroySmaEnv(SSmaEnv *pSmaEnv);
 void *tdFreeSmaEnv(SSmaEnv *pSmaEnv);
 
@@ -112,33 +143,6 @@ int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType);
 int32_t tdLockSma(SSma *pSma);
 int32_t tdUnLockSma(SSma *pSma);
 
-static FORCE_INLINE int32_t tdRLockSmaEnv(SSmaEnv *pEnv) {
-  int code = taosThreadRwlockRdlock(&(pEnv->lock));
-  if (code != 0) {
-    terrno = TAOS_SYSTEM_ERROR(code);
-    return -1;
-  }
-  return 0;
-}
-
-static FORCE_INLINE int32_t tdWLockSmaEnv(SSmaEnv *pEnv) {
-  int code = taosThreadRwlockWrlock(&(pEnv->lock));
-  if (code != 0) {
-    terrno = TAOS_SYSTEM_ERROR(code);
-    return -1;
-  }
-  return 0;
-}
-
-static FORCE_INLINE int32_t tdUnLockSmaEnv(SSmaEnv *pEnv) {
-  int code = taosThreadRwlockUnlock(&(pEnv->lock));
-  if (code != 0) {
-    terrno = TAOS_SYSTEM_ERROR(code);
-    return -1;
-  }
-  return 0;
-}
-
 static FORCE_INLINE int8_t tdSmaStat(STSmaStat *pTStat) {
   if (pTStat) {
     return atomic_load_8(&pTStat->state);
@@ -184,10 +188,12 @@ static FORCE_INLINE void tdSmaStatSetDropped(STSmaStat *pTStat) {
   }
 }
 
+int32_t        tdCloneRSmaInfo(SSma *pSma, SRSmaInfo *pDest, SRSmaInfo *pSrc);
+void           tdFreeQTaskInfo(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level);
 static int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType);
 void          *tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType);
-void          *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo);
-int32_t        tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat);
+void          *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree);
+int32_t        tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash);
 
 int32_t tdProcessRSmaCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName);
 int32_t tdProcessRSmaRestoreImpl(SSma *pSma);

source/dnode/vnode/src/inc/tsdb.h

@@ -24,12 +24,12 @@ extern "C" {
 
 // tsdbDebug ================
 // clang-format off
-#define tsdbFatal(...) do { if (tsdbDebugFlag & DEBUG_FATAL) { taosPrintLog("TSDB FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}     while(0)
-#define tsdbError(...) do { if (tsdbDebugFlag & DEBUG_ERROR) { taosPrintLog("TSDB ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }}     while(0)
-#define tsdbWarn(...)  do { if (tsdbDebugFlag & DEBUG_WARN)  { taosPrintLog("TSDB WARN ", DEBUG_WARN, 255, __VA_ARGS__); }}       while(0)
-#define tsdbInfo(...)  do { if (tsdbDebugFlag & DEBUG_INFO)  { taosPrintLog("TSDB ", DEBUG_INFO, 255, __VA_ARGS__); }}            while(0)
-#define tsdbDebug(...) do { if (tsdbDebugFlag & DEBUG_DEBUG) { taosPrintLog("TSDB ", DEBUG_DEBUG, tsdbDebugFlag, __VA_ARGS__); }} while(0)
-#define tsdbTrace(...) do { if (tsdbDebugFlag & DEBUG_TRACE) { taosPrintLog("TSDB ", DEBUG_TRACE, tsdbDebugFlag, __VA_ARGS__); }} while(0)
+#define tsdbFatal(...) do { if (tsdbDebugFlag & DEBUG_FATAL) { taosPrintLog("TSD FATAL ", DEBUG_FATAL, 255, __VA_ARGS__); }}     while(0)
+#define tsdbError(...) do { if (tsdbDebugFlag & DEBUG_ERROR) { taosPrintLog("TSD ERROR ", DEBUG_ERROR, 255, __VA_ARGS__); }}     while(0)
+#define tsdbWarn(...)  do { if (tsdbDebugFlag & DEBUG_WARN)  { taosPrintLog("TSD WARN ", DEBUG_WARN, 255, __VA_ARGS__); }}       while(0)
+#define tsdbInfo(...)  do { if (tsdbDebugFlag & DEBUG_INFO)  { taosPrintLog("TSD ", DEBUG_INFO, 255, __VA_ARGS__); }}            while(0)
+#define tsdbDebug(...) do { if (tsdbDebugFlag & DEBUG_DEBUG) { taosPrintLog("TSD ", DEBUG_DEBUG, tsdbDebugFlag, __VA_ARGS__); }} while(0)
+#define tsdbTrace(...) do { if (tsdbDebugFlag & DEBUG_TRACE) { taosPrintLog("TSD ", DEBUG_TRACE, tsdbDebugFlag, __VA_ARGS__); }} while(0)
 // clang-format on
 
 typedef struct TSDBROW       TSDBROW;
@@ -115,7 +115,6 @@ int32_t tGetBlock(uint8_t *p, void *ph);
 int32_t tBlockCmprFn(const void *p1, const void *p2);
 bool    tBlockHasSma(SBlock *pBlock);
 // SBlockIdx
-void    tBlockIdxReset(SBlockIdx *pBlockIdx);
 int32_t tPutBlockIdx(uint8_t *p, void *ph);
 int32_t tGetBlockIdx(uint8_t *p, void *ph);
 int32_t tCmprBlockIdx(void const *lhs, void const *rhs);
@@ -126,6 +125,8 @@ void    tColDataClear(void *ph);
 int32_t tColDataAppendValue(SColData *pColData, SColVal *pColVal);
 int32_t tColDataGetValue(SColData *pColData, int32_t iRow, SColVal *pColVal);
 int32_t tColDataCopy(SColData *pColDataSrc, SColData *pColDataDest);
+int32_t tPutColData(uint8_t *p, SColData *pColData);
+int32_t tGetColData(uint8_t *p, SColData *pColData);
 // SBlockData
 #define tBlockDataFirstRow(PBLOCKDATA) tsdbRowFromBlockData(PBLOCKDATA, 0)
 #define tBlockDataLastRow(PBLOCKDATA)  tsdbRowFromBlockData(PBLOCKDATA, (PBLOCKDATA)->nRow - 1)
@@ -134,14 +135,17 @@ int32_t tColDataCopy(SColData *pColDataSrc, SColData *pColDataDest);
 int32_t   tBlockDataInit(SBlockData *pBlockData);
 void      tBlockDataReset(SBlockData *pBlockData);
 int32_t   tBlockDataSetSchema(SBlockData *pBlockData, STSchema *pTSchema);
+int32_t   tBlockDataCorrectSchema(SBlockData *pBlockData, SBlockData *pBlockDataFrom);
 void      tBlockDataClearData(SBlockData *pBlockData);
-void      tBlockDataClear(SBlockData *pBlockData);
+void      tBlockDataClear(SBlockData *pBlockData, int8_t deepClear);
 int32_t   tBlockDataAddColData(SBlockData *pBlockData, int32_t iColData, SColData **ppColData);
 int32_t   tBlockDataAppendRow(SBlockData *pBlockData, TSDBROW *pRow, STSchema *pTSchema);
 int32_t   tBlockDataMerge(SBlockData *pBlockData1, SBlockData *pBlockData2, SBlockData *pBlockData);
 int32_t   tBlockDataCopy(SBlockData *pBlockDataSrc, SBlockData *pBlockDataDest);
 SColData *tBlockDataGetColDataByIdx(SBlockData *pBlockData, int32_t idx);
 void      tBlockDataGetColData(SBlockData *pBlockData, int16_t cid, SColData **ppColData);
+int32_t   tPutBlockData(uint8_t *p, SBlockData *pBlockData);
+int32_t   tGetBlockData(uint8_t *p, SBlockData *pBlockData);
 // SDelIdx
 int32_t tPutDelIdx(uint8_t *p, void *ph);
 int32_t tGetDelIdx(uint8_t *p, void *ph);
@@ -202,7 +206,7 @@ int32_t    tsdbFSStateUpsertDelFile(STsdbFSState *pState, SDelFile *pDelFile);
 int32_t    tsdbFSStateUpsertDFileSet(STsdbFSState *pState, SDFileSet *pSet);
 void       tsdbFSStateDeleteDFileSet(STsdbFSState *pState, int32_t fid);
 SDelFile  *tsdbFSStateGetDelFile(STsdbFSState *pState);
-SDFileSet *tsdbFSStateGetDFileSet(STsdbFSState *pState, int32_t fid);
+SDFileSet *tsdbFSStateGetDFileSet(STsdbFSState *pState, int32_t fid, int32_t flag);
 // tsdbReaderWriter.c ==============================================================================================
 // SDataFWriter
 int32_t tsdbDataFWriterOpen(SDataFWriter **ppWriter, STsdb *pTsdb, SDFileSet *pSet);
@@ -357,10 +361,6 @@ struct TSDBROW {
 struct SBlockIdx {
   int64_t suid;
   int64_t uid;
-  TSKEY   minKey;
-  TSKEY   maxKey;
-  int64_t minVersion;
-  int64_t maxVersion;
   int64_t offset;
   int64_t size;
 };

source/dnode/vnode/src/inc/vnodeInt.h

@@ -99,7 +99,8 @@ STSchema*       metaGetTbTSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver);
 int32_t         metaGetTbTSchemaEx(SMeta* pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sver, STSchema** ppTSchema);
 int             metaGetTableEntryByName(SMetaReader* pReader, const char* name);
 tb_uid_t        metaGetTableEntryUidByName(SMeta* pMeta, const char* name);
-int             metaGetTbNum(SMeta* pMeta);
+int64_t         metaGetTbNum(SMeta* pMeta);
+int64_t         metaGetTimeSeriesNum(SMeta* pMeta);
 SMCtbCursor*    metaOpenCtbCursor(SMeta* pMeta, tb_uid_t uid);
 void            metaCloseCtbCursor(SMCtbCursor* pCtbCur);
 tb_uid_t        metaCtbCursorNext(SMCtbCursor* pCtbCur);
@@ -164,9 +165,12 @@ void    smaCleanUp();
 int32_t smaOpen(SVnode* pVnode);
 int32_t smaClose(SSma* pSma);
 int32_t smaBegin(SSma* pSma);
-int32_t smaPreCommit(SSma* pSma);
-int32_t smaCommit(SSma* pSma);
-int32_t smaPostCommit(SSma* pSma);
+int32_t smaSyncPreCommit(SSma* pSma);
+int32_t smaSyncCommit(SSma* pSma);
+int32_t smaSyncPostCommit(SSma* pSma);
+int32_t smaAsyncPreCommit(SSma* pSma);
+int32_t smaAsyncCommit(SSma* pSma);
+int32_t smaAsyncPostCommit(SSma* pSma);
 
 int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg);
 int32_t tdProcessTSmaInsert(SSma* pSma, int64_t indexUid, const char* msg);
@@ -309,6 +313,7 @@ void smaHandleRes(void* pVnode, int64_t smaId, const SArray* data);
 
 struct SSnapDataHdr {
   int8_t  type;
+  int64_t index;
   int64_t size;
   uint8_t data[];
 };

source/dnode/vnode/src/meta/metaQuery.c

@@ -463,10 +463,16 @@ _err:
   return code;
 }
 
-int metaGetTbNum(SMeta *pMeta) {
+// N.B. Called by statusReq per second
+int64_t metaGetTbNum(SMeta *pMeta) {
   // TODO
-  // ASSERT(0);
-  return 0;
+  return 100;
+}
+
+// N.B. Called by statusReq per second
+int64_t metaGetTimeSeriesNum(SMeta *pMeta) { 
+  // TODO
+  return 400; 
 }
 
 typedef struct {

source/dnode/vnode/src/meta/metaSnapshot.c

@@ -26,60 +26,72 @@ struct SMetaSnapReader {
 int32_t metaSnapReaderOpen(SMeta* pMeta, int64_t sver, int64_t ever, SMetaSnapReader** ppReader) {
   int32_t          code = 0;
   int32_t          c = 0;
-  SMetaSnapReader* pMetaSnapReader = NULL;
+  SMetaSnapReader* pReader = NULL;
 
   // alloc
-  pMetaSnapReader = (SMetaSnapReader*)taosMemoryCalloc(1, sizeof(*pMetaSnapReader));
-  if (pMetaSnapReader == NULL) {
+  pReader = (SMetaSnapReader*)taosMemoryCalloc(1, sizeof(*pReader));
+  if (pReader == NULL) {
     code = TSDB_CODE_OUT_OF_MEMORY;
     goto _err;
   }
-  pMetaSnapReader->pMeta = pMeta;
-  pMetaSnapReader->sver = sver;
-  pMetaSnapReader->ever = ever;
+  pReader->pMeta = pMeta;
+  pReader->sver = sver;
+  pReader->ever = ever;
 
   // impl
-  code = tdbTbcOpen(pMeta->pTbDb, &pMetaSnapReader->pTbc, NULL);
+  code = tdbTbcOpen(pMeta->pTbDb, &pReader->pTbc, NULL);
   if (code) {
+    taosMemoryFree(pReader);
     goto _err;
   }
 
-  code = tdbTbcMoveTo(pMetaSnapReader->pTbc, &(STbDbKey){.version = sver, .uid = INT64_MIN}, sizeof(STbDbKey), &c);
+  code = tdbTbcMoveTo(pReader->pTbc, &(STbDbKey){.version = sver, .uid = INT64_MIN}, sizeof(STbDbKey), &c);
   if (code) {
+    taosMemoryFree(pReader);
     goto _err;
   }
 
-  *ppReader = pMetaSnapReader;
+  metaInfo("vgId:%d vnode snapshot meta reader opened", TD_VID(pMeta->pVnode));
+
+  *ppReader = pReader;
   return code;
 
 _err:
-  metaError("vgId:%d meta snap reader open failed since %s", TD_VID(pMeta->pVnode), tstrerror(code));
+  metaError("vgId:%d vnode snapshot meta reader open failed since %s", TD_VID(pMeta->pVnode), tstrerror(code));
   *ppReader = NULL;
   return code;
 }
 
 int32_t metaSnapReaderClose(SMetaSnapReader** ppReader) {
+  int32_t code = 0;
+
   tdbTbcClose((*ppReader)->pTbc);
   taosMemoryFree(*ppReader);
   *ppReader = NULL;
-  return 0;
+
+  return code;
 }
 
 int32_t metaSnapRead(SMetaSnapReader* pReader, uint8_t** ppData) {
+  int32_t     code = 0;
   const void* pKey = NULL;
   const void* pData = NULL;
   int32_t     nKey = 0;
   int32_t     nData = 0;
-  int32_t     code = 0;
+  STbDbKey    key;
 
+  *ppData = NULL;
   for (;;) {
-    code = tdbTbcGet(pReader->pTbc, &pKey, &nKey, &pData, &nData);
-    if (code || ((STbDbKey*)pData)->version > pReader->ever) {
-      code = TSDB_CODE_VND_READ_END;
+    if (tdbTbcGet(pReader->pTbc, &pKey, &nKey, &pData, &nData)) {
       goto _exit;
     }
 
-    if (((STbDbKey*)pData)->version < pReader->sver) {
+    key = ((STbDbKey*)pKey)[0];
+    if (key.version > pReader->ever) {
+      goto _exit;
+    }
+
+    if (key.version < pReader->sver) {
       tdbTbcMoveToNext(pReader->pTbc);
       continue;
     }
@@ -88,17 +100,28 @@ int32_t metaSnapRead(SMetaSnapReader* pReader, uint8_t** ppData) {
     break;
   }
 
-  // copy the data
-  if (tRealloc(ppData, sizeof(SSnapDataHdr) + nData) < 0) {
+  ASSERT(pData && nData);
+
+  *ppData = taosMemoryMalloc(sizeof(SSnapDataHdr) + nData);
+  if (*ppData == NULL) {
     code = TSDB_CODE_OUT_OF_MEMORY;
-    return code;
+    goto _err;
   }
-  ((SSnapDataHdr*)(*ppData))->type = 0;  // TODO: use macro
-  ((SSnapDataHdr*)(*ppData))->size = nData;
-  memcpy(((SSnapDataHdr*)(*ppData))->data, pData, nData);
+
+  SSnapDataHdr* pHdr = (SSnapDataHdr*)(*ppData);
+  pHdr->type = 0;  // TODO: use macro
+  pHdr->size = nData;
+  memcpy(pHdr->data, pData, nData);
+
+  metaInfo("vgId:%d vnode snapshot meta read data, version:%" PRId64 " uid:%" PRId64 " nData:%d",
+           TD_VID(pReader->pMeta->pVnode), key.version, key.uid, nData);
 
 _exit:
   return code;
+
+_err:
+  metaError("vgId:%d vnode snapshot meta read data failed since %s", TD_VID(pReader->pMeta->pVnode), tstrerror(code));
+  return code;
 }
 
 // SMetaSnapWriter ========================================
@@ -108,18 +131,6 @@ struct SMetaSnapWriter {
   int64_t ever;
 };
 
-static int32_t metaSnapRollback(SMetaSnapWriter* pWriter) {
-  int32_t code = 0;
-  // TODO
-  return code;
-}
-
-static int32_t metaSnapCommit(SMetaSnapWriter* pWriter) {
-  int32_t code = 0;
-  // TODO
-  return code;
-}
-
 int32_t metaSnapWriterOpen(SMeta* pMeta, int64_t sver, int64_t ever, SMetaSnapWriter** ppWriter) {
   int32_t          code = 0;
   SMetaSnapWriter* pWriter;
@@ -148,10 +159,9 @@ int32_t metaSnapWriterClose(SMetaSnapWriter** ppWriter, int8_t rollback) {
   SMetaSnapWriter* pWriter = *ppWriter;
 
   if (rollback) {
-    code = metaSnapRollback(pWriter);
-    if (code) goto _err;
+    ASSERT(0);
   } else {
-    code = metaSnapCommit(pWriter);
+    code = metaCommit(pWriter->pMeta);
     if (code) goto _err;
   }
   taosMemoryFree(pWriter);
@@ -170,15 +180,16 @@ int32_t metaSnapWrite(SMetaSnapWriter* pWriter, uint8_t* pData, uint32_t nData)
   SMetaEntry metaEntry = {0};
   SDecoder*  pDecoder = &(SDecoder){0};
 
-  tDecoderInit(pDecoder, pData, nData);
+  tDecoderInit(pDecoder, pData + sizeof(SSnapDataHdr), nData - sizeof(SSnapDataHdr));
   metaDecodeEntry(pDecoder, &metaEntry);
 
   code = metaHandleEntry(pMeta, &metaEntry);
   if (code) goto _err;
 
+  tDecoderClear(pDecoder);
   return code;
 
 _err:
-  metaError("vgId:%d meta snapshot write failed since %s", TD_VID(pMeta->pVnode), tstrerror(code));
+  metaError("vgId:%d vnode snapshot meta write failed since %s", TD_VID(pMeta->pVnode), tstrerror(code));
   return code;
 }
\ No newline at end of file

source/dnode/vnode/src/sma/smaCommit.c

@@ -15,9 +15,13 @@
 
 #include "sma.h"
 
-static int32_t tdProcessRSmaPreCommitImpl(SSma *pSma);
-static int32_t tdProcessRSmaCommitImpl(SSma *pSma);
-static int32_t tdProcessRSmaPostCommitImpl(SSma *pSma);
+static int32_t tdProcessRSmaSyncPreCommitImpl(SSma *pSma);
+static int32_t tdProcessRSmaSyncCommitImpl(SSma *pSma);
+static int32_t tdProcessRSmaSyncPostCommitImpl(SSma *pSma);
+static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma);
+static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma);
+static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma);
+static int32_t tdCleanupQTaskInfoFiles(SSma *pSma, SRSmaStat *pRSmaStat);
 
 /**
  * @brief Only applicable to Rollup SMA
@@ -25,7 +29,7 @@ static int32_t tdProcessRSmaPostCommitImpl(SSma *pSma);
  * @param pSma
  * @return int32_t
  */
-int32_t smaPreCommit(SSma *pSma) { return tdProcessRSmaPreCommitImpl(pSma); }
+int32_t smaSyncPreCommit(SSma *pSma) { return tdProcessRSmaSyncPreCommitImpl(pSma); }
 
 /**
  * @brief Only applicable to Rollup SMA
@@ -33,7 +37,7 @@ int32_t smaPreCommit(SSma *pSma) { return tdProcessRSmaPreCommitImpl(pSma); }
  * @param pSma
  * @return int32_t
  */
-int32_t smaCommit(SSma *pSma) { return tdProcessRSmaCommitImpl(pSma); }
+int32_t smaSyncCommit(SSma *pSma) { return tdProcessRSmaSyncCommitImpl(pSma); }
 
 /**
  * @brief Only applicable to Rollup SMA
@@ -41,7 +45,31 @@ int32_t smaCommit(SSma *pSma) { return tdProcessRSmaCommitImpl(pSma); }
  * @param pSma
  * @return int32_t
  */
-int32_t smaPostCommit(SSma *pSma) { return tdProcessRSmaPostCommitImpl(pSma); }
+int32_t smaSyncPostCommit(SSma *pSma) { return tdProcessRSmaSyncPostCommitImpl(pSma); }
+
+/**
+ * @brief Only applicable to Rollup SMA
+ *
+ * @param pSma
+ * @return int32_t
+ */
+int32_t smaAsyncPreCommit(SSma *pSma) { return tdProcessRSmaAsyncPreCommitImpl(pSma); }
+
+/**
+ * @brief Only applicable to Rollup SMA
+ *
+ * @param pSma
+ * @return int32_t
+ */
+int32_t smaAsyncCommit(SSma *pSma) { return tdProcessRSmaAsyncCommitImpl(pSma); }
+
+/**
+ * @brief Only applicable to Rollup SMA
+ *
+ * @param pSma
+ * @return int32_t
+ */
+int32_t smaAsyncPostCommit(SSma *pSma) { return tdProcessRSmaAsyncPostCommitImpl(pSma); }
 
 /**
  * @brief set rsma trigger stat active
@@ -62,18 +90,17 @@ int32_t smaBegin(SSma *pSma) {
       atomic_val_compare_exchange_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_PAUSED, TASK_TRIGGER_STAT_ACTIVE);
   switch (rsmaTriggerStat) {
     case TASK_TRIGGER_STAT_PAUSED: {
-      smaDebug("vgId:%d rsma trigger stat from paused to active", SMA_VID(pSma));
+      smaDebug("vgId:%d, rsma trigger stat from paused to active", SMA_VID(pSma));
       break;
     }
     case TASK_TRIGGER_STAT_INIT: {
       atomic_store_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_ACTIVE);
-      smaDebug("vgId:%d rsma trigger stat from init to active", SMA_VID(pSma));
+      smaDebug("vgId:%d, rsma trigger stat from init to active", SMA_VID(pSma));
       break;
     }
     default: {
       atomic_store_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_ACTIVE);
-      smaWarn("vgId:%d rsma trigger stat %" PRIi8 " is unexpected", SMA_VID(pSma), rsmaTriggerStat);
-      ASSERT(0);
+      smaError("vgId:%d, rsma trigger stat %" PRIi8 " is unexpected", SMA_VID(pSma), rsmaTriggerStat);
       break;
     }
   }
@@ -81,7 +108,7 @@ int32_t smaBegin(SSma *pSma) {
 }
 
 /**
- * @brief pre-commit for rollup sma.
+ * @brief pre-commit for rollup sma(sync commit).
  *  1) set trigger stat of rsma timer TASK_TRIGGER_STAT_PAUSED.
  *  2) wait all triggered fetch tasks finished
  *  3) perform persist task for qTaskInfo
@@ -89,7 +116,7 @@ int32_t smaBegin(SSma *pSma) {
  * @param pSma
  * @return int32_t
  */
-static int32_t tdProcessRSmaPreCommitImpl(SSma *pSma) {
+static int32_t tdProcessRSmaSyncPreCommitImpl(SSma *pSma) {
   SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
   if (!pSmaEnv) {
     return TSDB_CODE_SUCCESS;
@@ -98,8 +125,7 @@ static int32_t tdProcessRSmaPreCommitImpl(SSma *pSma) {
   SSmaStat  *pStat = SMA_ENV_STAT(pSmaEnv);
   SRSmaStat *pRSmaStat = SMA_RSMA_STAT(pStat);
 
-
-  // step 1: set persistence task paused
+  // step 1: set rsma stat paused
   atomic_store_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_PAUSED);
 
   // step 2: wait all triggered fetch tasks finished
@@ -119,7 +145,9 @@ static int32_t tdProcessRSmaPreCommitImpl(SSma *pSma) {
   }
 
   // step 3: perform persist task for qTaskInfo
-  tdRSmaPersistExecImpl(pRSmaStat);
+  pRSmaStat->commitAppliedVer = pSma->pVnode->state.applied;
+  pRSmaStat->commitSubmitVer = pRSmaStat->submitVer;
+  tdRSmaPersistExecImpl(pRSmaStat, RSMA_INFO_HASH(pRSmaStat));
 
   smaDebug("vgId:%d, rsma pre commit success", SMA_VID(pSma));
 
@@ -132,7 +160,7 @@ static int32_t tdProcessRSmaPreCommitImpl(SSma *pSma) {
  * @param pSma
  * @return int32_t
  */
-static int32_t tdProcessRSmaCommitImpl(SSma *pSma) {
+static int32_t tdProcessRSmaSyncCommitImpl(SSma *pSma) {
   SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
   if (!pSmaEnv) {
     return TSDB_CODE_SUCCESS;
@@ -140,21 +168,9 @@ static int32_t tdProcessRSmaCommitImpl(SSma *pSma) {
   return TSDB_CODE_SUCCESS;
 }
 
-/**
- * @brief post-commit for rollup sma
- *  1) clean up the outdated qtaskinfo files
- *
- * @param pSma
- * @return int32_t
- */
-static int32_t tdProcessRSmaPostCommitImpl(SSma *pSma) {
-  SVnode *pVnode = pSma->pVnode;
-
-  if (!VND_IS_RSMA(pVnode)) {
-    return TSDB_CODE_SUCCESS;
-  }
-
-  int64_t       committed = pVnode->state.committed;
+static int32_t tdCleanupQTaskInfoFiles(SSma *pSma, SRSmaStat *pRSmaStat) {
+  SVnode       *pVnode = pSma->pVnode;
+  int64_t       committed = pRSmaStat->commitAppliedVer;
   TdDirPtr      pDir = NULL;
   TdDirEntryPtr pDirEntry = NULL;
   char          dir[TSDB_FILENAME_LEN];
@@ -222,5 +238,159 @@ static int32_t tdProcessRSmaPostCommitImpl(SSma *pSma) {
 
   taosCloseDir(&pDir);
   regfree(&regex);
+
+  return TSDB_CODE_SUCCESS;
+}
+
+/**
+ * @brief post-commit for rollup sma
+ *  1) clean up the outdated qtaskinfo files
+ *
+ * @param pSma
+ * @return int32_t
+ */
+static int32_t tdProcessRSmaSyncPostCommitImpl(SSma *pSma) {
+  SVnode *pVnode = pSma->pVnode;
+  if (!VND_IS_RSMA(pVnode)) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SSmaEnv   *pSmaEnv = SMA_RSMA_ENV(pSma);
+  SRSmaStat *pRSmaStat = SMA_RSMA_STAT(SMA_ENV_STAT(pSmaEnv));
+
+  // cleanup outdated qtaskinfo files
+  tdCleanupQTaskInfoFiles(pSma, pRSmaStat);
+
+  return TSDB_CODE_SUCCESS;
+}
+
+/**
+ * @brief Rsma async commit implementation
+ *  1) set rsma stat TASK_TRIGGER_STAT_PAUSED
+ *  2) Wait all running fetch task finish to fetch and put submitMsg into level 2/3 wQueue(blocking level 1 write)
+ *  3)
+ *
+ * @param pSma
+ * @return int32_t
+ */
+static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma) {
+  SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
+  if (!pSmaEnv) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SSmaStat  *pStat = SMA_ENV_STAT(pSmaEnv);
+  SRSmaStat *pRSmaStat = SMA_RSMA_STAT(pStat);
+
+  // step 1: set rsma stat
+  atomic_store_8(RSMA_TRIGGER_STAT(pRSmaStat), TASK_TRIGGER_STAT_PAUSED);
+  atomic_store_8(RSMA_COMMIT_STAT(pRSmaStat), 1);
+
+  // step 2: wait all triggered fetch tasks finished
+  int32_t nLoops = 0;
+  while (1) {
+    if (T_REF_VAL_GET(pStat) == 0) {
+      smaDebug("vgId:%d, rsma fetch tasks all finished", SMA_VID(pSma));
+      break;
+    } else {
+      smaDebug("vgId:%d, rsma fetch tasks not all finished yet", SMA_VID(pSma));
+    }
+    ++nLoops;
+    if (nLoops > 1000) {
+      sched_yield();
+      nLoops = 0;
+    }
+  }
+
+  // step 3:  swap rsmaInfoHash and iRsmaInfoHash
+  ASSERT(!RSMA_IMU_INFO_HASH(pRSmaStat));
+  ASSERT(RSMA_INFO_HASH(pRSmaStat));
+
+  RSMA_IMU_INFO_HASH(pRSmaStat) = RSMA_INFO_HASH(pRSmaStat);
+  RSMA_INFO_HASH(pRSmaStat) =
+      taosHashInit(RSMA_TASK_INFO_HASH_SLOT, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK);
+
+  if (!RSMA_INFO_HASH(pRSmaStat)) {
+    smaError("vgId:%d, rsma async commit failed since %s", SMA_VID(pSma), terrstr());
+    return TSDB_CODE_FAILED;
+  }
+
+  // step 4: others
+  pRSmaStat->commitAppliedVer = pSma->pVnode->state.applied;
+  pRSmaStat->commitSubmitVer = pRSmaStat->submitVer;
+
+  return TSDB_CODE_SUCCESS;
+}
+
+/**
+ * @brief commit for rollup sma
+ *
+ * @param pSma
+ * @return int32_t
+ */
+static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma) {
+  SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
+  if (!pSmaEnv) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SSmaStat  *pStat = SMA_ENV_STAT(pSmaEnv);
+  SRSmaStat *pRSmaStat = SMA_RSMA_STAT(pStat);
+
+  // perform persist task for qTaskInfo
+  tdRSmaPersistExecImpl(pRSmaStat, RSMA_IMU_INFO_HASH(pRSmaStat));
+
+  return TSDB_CODE_SUCCESS;
+}
+
+/**
+ * @brief Migrate rsmaInfo from iRsmaInfo to rsmaInfo if rsmaInfoHash not empty.
+ *
+ * @param pSma
+ * @return int32_t
+ */
+static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma) {
+  SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
+  if (!pSmaEnv) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SSmaStat  *pStat = SMA_ENV_STAT(pSmaEnv);
+  SRSmaStat *pRSmaStat = SMA_RSMA_STAT(pStat);
+
+  // step 1: merge rsmaInfoHash and iRsmaInfoHash
+  taosWLockLatch(SMA_ENV_LOCK(pSmaEnv));
+
+  if (taosHashGetSize(RSMA_INFO_HASH(pRSmaStat)) <= 0) {
+    // TODO: optimization - just switch the hash pointer if rsmaInfoHash is empty
+  }
+
+  void *pIter = taosHashIterate(RSMA_IMU_INFO_HASH(pRSmaStat), NULL);
+  while (pIter) {
+    tb_uid_t *pSuid = (tb_uid_t *)taosHashGetKey(pIter, NULL);
+
+    if (!taosHashGet(RSMA_INFO_HASH(pRSmaStat), pSuid, sizeof(tb_uid_t))) {
+      taosHashPut(RSMA_INFO_HASH(pRSmaStat), pSuid, sizeof(tb_uid_t), pIter, sizeof(pIter));
+      smaDebug("vgId:%d, rsma async post commit, migrated from iRsmaInfoHash for table:%" PRIi64, SMA_VID(pSma),
+               *pSuid);
+    } else {
+      // free the resources
+      SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)pIter;
+      tdFreeRSmaInfo(pSma, pRSmaInfo, false);
+      smaDebug("vgId:%d, rsma async post commit, free rsma info since already COW for table:%" PRIi64, SMA_VID(pSma),
+               *pSuid);
+    }
+
+    pIter = taosHashIterate(RSMA_IMU_INFO_HASH(pRSmaStat), pIter);
+  }
+
+  taosHashCleanup(RSMA_IMU_INFO_HASH(pRSmaStat));
+  RSMA_IMU_INFO_HASH(pRSmaStat) = NULL;
+
+  taosWUnLockLatch(SMA_ENV_LOCK(pSmaEnv));
+
+  // step 2: cleanup outdated qtaskinfo files
+  tdCleanupQTaskInfoFiles(pSma, pRSmaStat);
+
   return TSDB_CODE_SUCCESS;
 }

source/dnode/vnode/src/sma/smaEnv.c

@@ -17,7 +17,6 @@
 
 typedef struct SSmaStat SSmaStat;
 
-#define RSMA_TASK_INFO_HASH_SLOT 8
 #define SMA_MGMT_REF_NUM         10240
 
 extern SSmaMgmt smaMgmt;
@@ -109,12 +108,7 @@ static SSmaEnv *tdNewSmaEnv(const SSma *pSma, int8_t smaType, const char *path)
 
   SMA_ENV_TYPE(pEnv) = smaType;
 
-  int code = taosThreadRwlockInit(&(pEnv->lock), NULL);
-  if (code) {
-    terrno = TAOS_SYSTEM_ERROR(code);
-    taosMemoryFree(pEnv);
-    return NULL;
-  }
+  taosInitRWLatch(&(pEnv->lock));
 
   if (tdInitSmaStat(&SMA_ENV_STAT(pEnv), smaType, pSma) != TSDB_CODE_SUCCESS) {
     tdFreeSmaEnv(pEnv);
@@ -148,7 +142,6 @@ static int32_t tdInitSmaEnv(SSma *pSma, int8_t smaType, const char *path, SSmaEn
 void tdDestroySmaEnv(SSmaEnv *pSmaEnv) {
   if (pSmaEnv) {
     pSmaEnv->pStat = tdFreeSmaState(pSmaEnv->pStat, SMA_ENV_TYPE(pSmaEnv));
-    taosThreadRwlockDestroy(&(pSmaEnv->lock));
   }
 }
 
@@ -260,7 +253,7 @@ static void tdDestroyRSmaStat(void *pRSmaStat) {
       void *infoHash = taosHashIterate(RSMA_INFO_HASH(pStat), NULL);
       while (infoHash) {
         SRSmaInfo *pSmaInfo = *(SRSmaInfo **)infoHash;
-        tdFreeRSmaInfo(pSma, pSmaInfo);
+        tdFreeRSmaInfo(pSma, pSmaInfo, true);
         infoHash = taosHashIterate(RSMA_INFO_HASH(pStat), infoHash);
       }
     }
@@ -311,7 +304,6 @@ int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType) {
       if (taosRemoveRef(smaMgmt.rsetId, RSMA_REF_ID(pRSmaStat)) < 0) {
         smaError("vgId:%d, remove refId:%" PRIi64 " from rsmaRef:%" PRIi32 " failed since %s", SMA_VID(pRSmaStat->pSma),
                  RSMA_REF_ID(pRSmaStat), smaMgmt.rsetId, terrstr());
-        ASSERT(0);
       } else {
         smaDebug("vgId:%d, remove refId:%" PRIi64 " from rsmaRef:%" PRIi32 " succeed", SMA_VID(pRSmaStat->pSma),
                  RSMA_REF_ID(pRSmaStat), smaMgmt.rsetId);

source/dnode/vnode/src/sma/smaRollup.c

@@ -48,20 +48,7 @@ static int32_t tdRSmaRestoreQTaskInfoInit(SSma *pSma, int64_t *nTables);
 static int32_t tdRSmaRestoreQTaskInfoReload(SSma *pSma, int64_t *committed);
 static int32_t tdRSmaRestoreTSDataReload(SSma *pSma, int64_t committed);
 
-struct SRSmaInfoItem {
-  void   *taskInfo;  // qTaskInfo_t
-  int64_t refId;
-  tmr_h   tmrId;
-  int32_t maxDelay;
-  int8_t  level;
-  int8_t  triggerStat;
-};
 
-struct SRSmaInfo {
-  STSchema     *pTSchema;
-  int64_t       suid;
-  SRSmaInfoItem items[TSDB_RETENTION_L2];
-};
 
 static SRSmaInfo *tdGetRSmaInfoByItem(SRSmaInfoItem *pItem) {
   // adapt accordingly if definition of SRSmaInfo update
@@ -102,8 +89,9 @@ static FORCE_INLINE int32_t tdRSmaQTaskInfoContLen(int32_t lenWithHead) {
 
 static FORCE_INLINE void tdRSmaQTaskInfoIterDestroy(SRSmaQTaskInfoIter *pIter) { taosMemoryFreeClear(pIter->pBuf); }
 
-static FORCE_INLINE void tdFreeTaskHandle(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level) {
+void tdFreeQTaskInfo(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level) {
   // Note: free/kill may in RC
+  if (!taskHandle) return;
   qTaskInfo_t otaskHandle = atomic_load_ptr(taskHandle);
   if (otaskHandle && atomic_val_compare_exchange_ptr(taskHandle, otaskHandle, NULL)) {
     smaDebug("vgId:%d, free qTaskInfo_t %p of level %d", vgId, otaskHandle, level);
@@ -111,25 +99,36 @@ static FORCE_INLINE void tdFreeTaskHandle(qTaskInfo_t *taskHandle, int32_t vgId,
   } else {
     smaDebug("vgId:%d, not free qTaskInfo_t %p of level %d", vgId, otaskHandle, level);
   }
+  // TODO: clear files related to qTaskInfo?
 }
 
-void *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo) {
+/**
+ * @brief general function to free rsmaInfo
+ *
+ * @param pSma
+ * @param pInfo
+ * @param isDeepFree Only stop tmrId and free pTSchema for deep free
+ * @return void*
+ */
+void *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree) {
   if (pInfo) {
     for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
       SRSmaInfoItem *pItem = &pInfo->items[i];
       if (pItem->taskInfo) {
-        if (pItem->tmrId) {
+        if (isDeepFree && pItem->tmrId) {
           smaDebug("vgId:%d, table %" PRIi64 " stop fetch timer %p level %d", SMA_VID(pSma), pInfo->suid, pItem->tmrId,
                    i + 1);
           taosTmrStopA(&pItem->tmrId);
         }
-        tdFreeTaskHandle(&pItem->taskInfo, SMA_VID(pSma), i + 1);
+        tdFreeQTaskInfo(&pItem->taskInfo, SMA_VID(pSma), i + 1);
       } else {
         smaDebug("vgId:%d, table %" PRIi64 " no need to destroy rsma info level %d since empty taskInfo", SMA_VID(pSma),
                  pInfo->suid, i + 1);
       }
     }
-    taosMemoryFree(pInfo->pTSchema);
+    if (isDeepFree) {
+      taosMemoryFree(pInfo->pTSchema);
+    }
     taosMemoryFree(pInfo);
   }
 
@@ -151,7 +150,7 @@ static int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids)
 
   if (!suid || !tbUids) {
     terrno = TSDB_CODE_INVALID_PTR;
-    smaError("vgId:%d, failed to get rsma info for uid:%" PRIi64 " since %s", SMA_VID(pSma), *suid, terrstr(terrno));
+    smaError("vgId:%d, failed to get rsma info for uid:%" PRIi64 " since %s", SMA_VID(pSma), *suid, terrstr());
     return TSDB_CODE_FAILED;
   }
 
@@ -165,7 +164,7 @@ static int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids)
 
   if (pRSmaInfo->items[0].taskInfo) {
     if ((qUpdateQualifiedTableId(pRSmaInfo->items[0].taskInfo, tbUids, true) < 0)) {
-      smaError("vgId:%d, update tbUidList failed for uid:%" PRIi64 " since %s", SMA_VID(pSma), *suid, terrstr(terrno));
+      smaError("vgId:%d, update tbUidList failed for uid:%" PRIi64 " since %s", SMA_VID(pSma), *suid, terrstr());
       return TSDB_CODE_FAILED;
     } else {
       smaDebug("vgId:%d, update tbUidList succeed for qTaskInfo:%p with suid:%" PRIi64 ", uid:%" PRIi64, SMA_VID(pSma),
@@ -175,7 +174,7 @@ static int32_t tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids)
 
   if (pRSmaInfo->items[1].taskInfo) {
     if ((qUpdateQualifiedTableId(pRSmaInfo->items[1].taskInfo, tbUids, true) < 0)) {
-      smaError("vgId:%d, update tbUidList failed for uid:%" PRIi64 " since %s", SMA_VID(pSma), *suid, terrstr(terrno));
+      smaError("vgId:%d, update tbUidList failed for uid:%" PRIi64 " since %s", SMA_VID(pSma), *suid, terrstr());
       return TSDB_CODE_FAILED;
     } else {
       smaDebug("vgId:%d, update tbUidList succeed for qTaskInfo:%p with suid:%" PRIi64 ", uid:%" PRIi64, SMA_VID(pSma),
@@ -257,22 +256,22 @@ int32_t tdFetchTbUidList(SSma *pSma, STbUidStore **ppStore, tb_uid_t suid, tb_ui
 
 static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat *pStat, SRSmaInfo *pRSmaInfo,
                                        int8_t idx) {
-  SRetention *pRetention = SMA_RETENTION(pSma);
-  STsdbCfg   *pTsdbCfg = SMA_TSDB_CFG(pSma);
-
-  SReadHandle handle = {
-      .meta = pSma->pVnode->pMeta,
-      .vnode = pSma->pVnode,
-      .initTqReader = 1,
-  };
+  if ((param->qmsgLen > 0) && param->qmsg[idx]) {
+    SRetention *pRetention = SMA_RETENTION(pSma);
+    STsdbCfg   *pTsdbCfg = SMA_TSDB_CFG(pSma);
+    SVnode     *pVnode = pSma->pVnode;
+    SReadHandle handle = {
+        .meta = pVnode->pMeta,
+        .vnode = pVnode,
+        .initTqReader = 1,
+    };
 
-  if (param->qmsg[idx]) {
     SRSmaInfoItem *pItem = &(pRSmaInfo->items[idx]);
     pItem->refId = RSMA_REF_ID(pStat);
     pItem->taskInfo = qCreateStreamExecTaskInfo(param->qmsg[idx], &handle);
     if (!pItem->taskInfo) {
       terrno = TSDB_CODE_RSMA_QTASKINFO_CREATE;
-      goto _err;
+      return TSDB_CODE_FAILED;
     }
     pItem->triggerStat = TASK_TRIGGER_STAT_INACTIVE;
     if (param->maxdelay[idx] < TSDB_MIN_ROLLUP_MAX_DELAY) {
@@ -286,13 +285,11 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
       pItem->maxDelay = TSDB_MAX_ROLLUP_MAX_DELAY;
     }
     pItem->level = idx == 0 ? TSDB_RETENTION_L1 : TSDB_RETENTION_L2;
-    smaInfo("vgId:%d table:%" PRIi64 " level:%" PRIi8 " maxdelay:%" PRIi64 " watermark:%" PRIi64
+    smaInfo("vgId:%d, table:%" PRIi64 " level:%" PRIi8 " maxdelay:%" PRIi64 " watermark:%" PRIi64
             ", finally maxdelay:%" PRIi32,
-            SMA_VID(pSma), pRSmaInfo->suid, idx + 1, param->maxdelay[idx], param->watermark[idx], pItem->maxDelay);
+            TD_VID(pVnode), pRSmaInfo->suid, idx + 1, param->maxdelay[idx], param->watermark[idx], pItem->maxDelay);
   }
   return TSDB_CODE_SUCCESS;
-_err:
-  return TSDB_CODE_FAILED;
 }
 
 /**
@@ -357,7 +354,7 @@ int32_t tdProcessRSmaCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, con
 
   return TSDB_CODE_SUCCESS;
 _err:
-  tdFreeRSmaInfo(pSma, pRSmaInfo);
+  tdFreeRSmaInfo(pSma, pRSmaInfo, true);
   return TSDB_CODE_FAILED;
 }
 
@@ -562,7 +559,9 @@ static int32_t tdRSmaFetchAndSubmitResult(SRSmaInfoItem *pItem, STSchema *pTSche
     SSDataBlock *output = NULL;
     uint64_t     ts;
     if (qExecTask(pItem->taskInfo, &output, &ts) < 0) {
-      ASSERT(false);
+      smaError("vgId:%d, qExecTask for rsma table %" PRIi64 "l evel %" PRIi8 " failed since %s", SMA_VID(pSma), suid,
+               pItem->level, terrstr());
+      goto _err;
     }
     if (!output) {
       break;
@@ -572,7 +571,7 @@ static int32_t tdRSmaFetchAndSubmitResult(SRSmaInfoItem *pItem, STSchema *pTSche
       pResult = taosArrayInit(1, sizeof(SSDataBlock));
       if (!pResult) {
         terrno = TSDB_CODE_OUT_OF_MEMORY;
-        return TSDB_CODE_FAILED;
+        goto _err;
       }
     }
 
@@ -649,9 +648,18 @@ static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t inputType
   return TSDB_CODE_SUCCESS;
 }
 
+/**
+ * @brief During async commit, the SRSmaInfo object would be COW from iRSmaInfoHash and write lock should be applied.
+ * 
+ * @param pSma 
+ * @param suid 
+ * @return SRSmaInfo* 
+ */
 static SRSmaInfo *tdGetRSmaInfoBySuid(SSma *pSma, int64_t suid) {
   SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
   SRSmaStat *pStat = NULL;
+  SRSmaInfo *pRSmaInfo = NULL;
+
   if (!pEnv) {
     // only applicable when rsma env exists
     return NULL;
@@ -662,11 +670,37 @@ static SRSmaInfo *tdGetRSmaInfoBySuid(SSma *pSma, int64_t suid) {
     return NULL;
   }
 
-  SRSmaInfo *pRSmaInfo = taosHashGet(RSMA_INFO_HASH(pStat), &suid, sizeof(tb_uid_t));
-  if (!pRSmaInfo || !(pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) {
+  pRSmaInfo = taosHashGet(RSMA_INFO_HASH(pStat), &suid, sizeof(tb_uid_t));
+  if (pRSmaInfo && (pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) {
+    return pRSmaInfo;
+  }
+
+  if (RSMA_COMMIT_STAT(pStat) == 0) {
     return NULL;
   }
-  return pRSmaInfo;
+
+  // clone the SRSmaInfo from iRsmaInfoHash to rsmaInfoHash if in committing stat
+  SRSmaInfo *pCowRSmaInfo = NULL;
+  // lock
+  taosWLockLatch(SMA_ENV_LOCK(pEnv));
+  void *iRSmaInfo = taosHashGet(RSMA_IMU_INFO_HASH(pStat), &suid, sizeof(tb_uid_t));
+  if (iRSmaInfo) {
+    SRSmaInfo *pIRSmaInfo = *(SRSmaInfo **)iRSmaInfo;
+    if (pIRSmaInfo) {
+      if (tdCloneRSmaInfo(pSma, pCowRSmaInfo, pIRSmaInfo) < 0) {
+        taosWUnLockLatch(SMA_ENV_LOCK(pEnv));
+        smaError("vgId:%d, clone rsma info failed for suid:%" PRIu64 " since %s", SMA_VID(pSma), suid, terrstr());
+        return NULL;
+      }
+      if (taosHashPut(RSMA_INFO_HASH(pStat), &suid, sizeof(tb_uid_t), &pCowRSmaInfo, sizeof(pCowRSmaInfo)) < 0) {
+        taosWUnLockLatch(SMA_ENV_LOCK(pEnv));
+        return NULL;
+      }
+    }
+  }
+  // unlock
+  taosWUnLockLatch(SMA_ENV_LOCK(pEnv));
+  return pCowRSmaInfo;
 }
 
 static int32_t tdExecuteRSma(SSma *pSma, const void *pMsg, int32_t inputType, tb_uid_t suid) {
@@ -891,10 +925,17 @@ int32_t tdProcessRSmaRestoreImpl(SSma *pSma) {
 
   return TSDB_CODE_SUCCESS;
 _err:
-  smaError("vgId:%d failed to restore rsma task since %s", SMA_VID(pSma), terrstr());
+  smaError("vgId:%d, failed to restore rsma task since %s", SMA_VID(pSma), terrstr());
   return TSDB_CODE_FAILED;
 }
 
+/**
+ * @brief Restore from SRSmaQTaskInfoItem
+ * 
+ * @param pSma 
+ * @param pItem 
+ * @return int32_t 
+ */
 static int32_t tdRSmaQTaskInfoItemRestore(SSma *pSma, const SRSmaQTaskInfoItem *pItem) {
   SRSmaInfo *pRSmaInfo = NULL;
   void      *qTaskInfo = NULL;
@@ -920,7 +961,7 @@ static int32_t tdRSmaQTaskInfoItemRestore(SSma *pSma, const SRSmaQTaskInfoItem *
 
   if (qDeserializeTaskStatus(qTaskInfo, pItem->qTaskInfo, pItem->len) < 0) {
     smaError("vgId:%d, restore rsma task failed for table:%" PRIi64 " level %d since %s", SMA_VID(pSma), pItem->suid,
-             pItem->type, terrstr(terrno));
+             pItem->type, terrstr());
     return TSDB_CODE_FAILED;
   }
   smaDebug("vgId:%d, restore rsma task success for table:%" PRIi64 " level %d", SMA_VID(pSma), pItem->suid,
@@ -1067,26 +1108,27 @@ static int32_t tdRSmaQTaskInfoRestore(SSma *pSma, SRSmaQTaskInfoIter *pIter) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat) {
+int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash) {
   SSma   *pSma = pRSmaStat->pSma;
   SVnode *pVnode = pSma->pVnode;
   int32_t vid = SMA_VID(pSma);
   int64_t toffset = 0;
   bool    isFileCreated = false;
 
-  if (taosHashGetSize(RSMA_INFO_HASH(pRSmaStat)) <= 0) {
+  if (taosHashGetSize(pInfoHash) <= 0) {
     return TSDB_CODE_SUCCESS;
   }
 
-  void *infoHash = taosHashIterate(RSMA_INFO_HASH(pRSmaStat), NULL);
+  void *infoHash = taosHashIterate(pInfoHash, NULL);
   if (!infoHash) {
     return TSDB_CODE_SUCCESS;
   }
 
   STFile tFile = {0};
-  if (RSMA_SUBMIT_VER(pRSmaStat) > 0) {
+#if 0
+  if (pRSmaStat->commitAppliedVer > 0) {
     char qTaskInfoFName[TSDB_FILENAME_LEN];
-    tdRSmaQTaskInfoGetFName(vid, pSma->pVnode->state.applied, qTaskInfoFName);
+    tdRSmaQTaskInfoGetFName(vid, pRSmaStat->commitAppliedVer, qTaskInfoFName);
     if (tdInitTFile(&tFile, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFName) < 0) {
       smaError("vgId:%d, rsma persit, init %s failed since %s", vid, qTaskInfoFName, terrstr());
       goto _err;
@@ -1099,6 +1141,7 @@ int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat) {
 
     isFileCreated = true;
   }
+#endif
 
   while (infoHash) {
     SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)infoHash;
@@ -1114,7 +1157,7 @@ int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat) {
       int8_t  type = (int8_t)(i + 1);
       if (qSerializeTaskStatus(taskInfo, &pOutput, &len) < 0) {
         smaError("vgId:%d, rsma, table %" PRIi64 " level %d serialize qTaskInfo failed since %s", vid, pRSmaInfo->suid,
-                 i + 1, terrstr(terrno));
+                 i + 1, terrstr());
         goto _err;
       }
       if (!pOutput || len <= 0) {
@@ -1130,7 +1173,7 @@ int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat) {
 
       if (!isFileCreated) {
         char qTaskInfoFName[TSDB_FILENAME_LEN];
-        tdRSmaQTaskInfoGetFName(vid, pSma->pVnode->state.applied, qTaskInfoFName);
+        tdRSmaQTaskInfoGetFName(vid, pRSmaStat->commitAppliedVer, qTaskInfoFName);
         if (tdInitTFile(&tFile, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFName) < 0) {
           smaError("vgId:%d, rsma persit, init %s failed since %s", vid, qTaskInfoFName, terrstr());
           goto _err;
@@ -1163,11 +1206,11 @@ int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat) {
       taosMemoryFree(pOutput);
     }
 
-    infoHash = taosHashIterate(RSMA_INFO_HASH(pRSmaStat), infoHash);
+    infoHash = taosHashIterate(pInfoHash, infoHash);
   }
 
   if (isFileCreated) {
-    tFile.info.qTaskInfo.submitVer = atomic_load_64(&pRSmaStat->submitVer);
+    tFile.info.qTaskInfo.submitVer = atomic_load_64(&pRSmaStat->commitSubmitVer);
     if (tdUpdateTFileHeader(&tFile) < 0) {
       smaError("vgId:%d, rsma, failed to update tfile %s header since %s", vid, TD_TFILE_FULL_NAME(&tFile),
                tstrerror(terrno));
@@ -1217,6 +1260,10 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
       smaDebug("vgId:%d, not fetch rsma level %" PRIi8 " data since stat is %" PRIi8 ", rsetId rsetId:%" PRIi64
                " refId:%d",
                SMA_VID(pSma), pItem->level, rsmaTriggerStat, smaMgmt.rsetId, pItem->refId);
+      if (rsmaTriggerStat == TASK_TRIGGER_STAT_PAUSED) {
+        taosTmrReset(tdRSmaFetchTrigger, pItem->maxDelay > 5000 ? 5000 : pItem->maxDelay, pItem, smaMgmt.tmrHandle,
+                     &pItem->tmrId);
+      }
       return;
     }
     default:

source/dnode/vnode/src/sma/smaUtil.c

@@ -313,4 +313,99 @@ int32_t tdReleaseSmaRef(int32_t rsetId, int64_t refId, const char *tags, int32_t
 
   return TSDB_CODE_SUCCESS;
 }
+
+static int32_t tdCloneQTaskInfo(SSma *pSma, qTaskInfo_t dstTaskInfo, qTaskInfo_t srcTaskInfo, SRSmaParam *param,
+                                tb_uid_t suid, int8_t idx) {
+  SVnode *pVnode = pSma->pVnode;
+  char   *pOutput = NULL;
+  int32_t len = 0;
+
+  if (qSerializeTaskStatus(srcTaskInfo, &pOutput, &len) < 0) {
+    smaError("vgId:%d, rsma clone, table %" PRIi64 " serialize qTaskInfo failed since %s", TD_VID(pVnode), suid,
+             terrstr());
+    goto _err;
+  }
+
+  SReadHandle handle = {
+      .meta = pVnode->pMeta,
+      .vnode = pVnode,
+      .initTqReader = 1,
+  };
+  ASSERT(!dstTaskInfo);
+  dstTaskInfo = qCreateStreamExecTaskInfo(param->qmsg[idx], &handle);
+  if (!dstTaskInfo) {
+    terrno = TSDB_CODE_RSMA_QTASKINFO_CREATE;
+    goto _err;
+  }
+
+  if (qDeserializeTaskStatus(dstTaskInfo, pOutput, len) < 0) {
+    smaError("vgId:%d, rsma clone, restore rsma task for table:%" PRIi64 " failed since %s", TD_VID(pVnode), suid,
+             terrstr());
+    goto _err;
+  }
+
+  smaError("vgId:%d, rsma clone, restore rsma task for table:%" PRIi64 " succeed", TD_VID(pVnode), suid);
+
+  taosMemoryFreeClear(pOutput);
+  return TSDB_CODE_SUCCESS;
+_err:
+  taosMemoryFreeClear(pOutput);
+  tdFreeQTaskInfo(dstTaskInfo, TD_VID(pVnode), idx + 1);
+  return TSDB_CODE_FAILED;
+}
+
+/**
+ * @brief pTSchema is shared
+ * 
+ * @param pSma 
+ * @param pDest 
+ * @param pSrc 
+ * @return int32_t 
+ */
+int32_t tdCloneRSmaInfo(SSma *pSma, SRSmaInfo *pDest, SRSmaInfo *pSrc) {
+  SVnode     *pVnode = pSma->pVnode;
+  SRSmaParam *param = NULL;
+  if (!pSrc) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  if (!pDest) {
+    pDest = taosMemoryCalloc(1, sizeof(SRSmaInfo));
+    if (!pDest) {
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      return TSDB_CODE_FAILED;
+    }
+  }
+
+  memcpy(pDest, pSrc, sizeof(SRSmaInfo));
+
+  SMetaReader mr = {0};
+  metaReaderInit(&mr, SMA_META(pSma), 0);
+  smaDebug("vgId:%d, rsma clone, suid is %" PRIi64, TD_VID(pVnode), pSrc->suid);
+  if (metaGetTableEntryByUid(&mr, pSrc->suid) < 0) {
+    smaError("vgId:%d, rsma clone, failed to get table meta for %" PRIi64 " since %s", TD_VID(pVnode), pSrc->suid,
+             terrstr());
+    goto _err;
+  }
+  ASSERT(mr.me.type == TSDB_SUPER_TABLE);
+  ASSERT(mr.me.uid == pSrc->suid);
+  if (TABLE_IS_ROLLUP(mr.me.flags)) {
+    param = &mr.me.stbEntry.rsmaParam;
+    for (int i = 0; i < TSDB_RETENTION_L2; ++i) {
+      SRSmaInfoItem *pItem = &pSrc->items[i];
+      if (pItem->taskInfo) {
+        tdCloneQTaskInfo(pSma, pDest->items[i].taskInfo, pItem->taskInfo, param, pSrc->suid, i);
+      }
+    }
+    smaDebug("vgId:%d, rsma clone env success for %" PRIi64, TD_VID(pVnode), pSrc->suid);
+  }
+
+  metaReaderClear(&mr);
+  return TSDB_CODE_SUCCESS;
+_err:
+  metaReaderClear(&mr);
+  tdFreeRSmaInfo(pSma, pDest, false);
+  return TSDB_CODE_FAILED;
+}
+
 // ...
\ No newline at end of file

source/dnode/vnode/src/tsdb/tsdbCache.c

@@ -476,9 +476,9 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow) {
       if (--state->iFileSet >= 0) {
         pFileSet = (SDFileSet *)taosArrayGet(state->aDFileSet, state->iFileSet);
       } else {
-        // tBlockDataClear(&state->blockData);
+        // tBlockDataClear(&state->blockData, 1);
         if (state->pBlockData) {
-          tBlockDataClear(state->pBlockData);
+          tBlockDataClear(state->pBlockData, 1);
           state->pBlockData = NULL;
         }
 
@@ -500,9 +500,7 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow) {
       if (code) goto _err;
 
       /* if (state->pBlockIdx) { */
-      /*   tBlockIdxReset(state->blockIdx); */
       /* } */
-      /* tBlockIdxReset(state->blockIdx); */
       /* code = tMapDataSearch(&state->blockIdxMap, state->pBlockIdxExp, tGetBlockIdx, tCmprBlockIdx,
        * &state->blockIdx);
        */
@@ -582,8 +580,8 @@ _err:
     state->aBlockIdx = NULL;
   }
   if (state->pBlockData) {
-    // tBlockDataClear(&state->blockData);
-    tBlockDataClear(state->pBlockData);
+    // tBlockDataClear(&state->blockData, 1);
+    tBlockDataClear(state->pBlockData, 1);
     state->pBlockData = NULL;
   }
 
@@ -609,8 +607,8 @@ int32_t clearNextRowFromFS(void *iter) {
     state->aBlockIdx = NULL;
   }
   if (state->pBlockData) {
-    // tBlockDataClear(&state->blockData);
-    tBlockDataClear(state->pBlockData);
+    // tBlockDataClear(&state->blockData, 1);
+    tBlockDataClear(state->pBlockData, 1);
     state->pBlockData = NULL;
   }
 

source/dnode/vnode/src/tsdb/tsdbCommit.c

@@ -263,7 +263,7 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) {
   taosArrayClear(pCommitter->aBlockIdx);
   tMapDataReset(&pCommitter->oBlockMap);
   tBlockDataReset(&pCommitter->oBlockData);
-  pRSet = tsdbFSStateGetDFileSet(pTsdb->fs->nState, pCommitter->commitFid);
+  pRSet = tsdbFSStateGetDFileSet(pTsdb->fs->nState, pCommitter->commitFid, TD_EQ);
   if (pRSet) {
     code = tsdbDataFReaderOpen(&pCommitter->pReader, pTsdb, pRSet);
     if (code) goto _err;
@@ -284,16 +284,7 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) {
                        .fLast = {.commitID = pCommitter->commitID, .size = 0},
                        .fSma = pRSet->fSma};
   } else {
-    STfs   *pTfs = pTsdb->pVnode->pTfs;
-    SDiskID did = {.level = 0, .id = 0};
-
-    // TODO: alloc a new disk
-    // tfsAllocDisk(pTfs, 0, &did);
-
-    // create the directory
-    tfsMkdirRecurAt(pTfs, pTsdb->path, did);
-
-    wSet = (SDFileSet){.diskId = did,
+    wSet = (SDFileSet){.diskId = (SDiskID){.level = 0, .id = 0},
                        .fid = pCommitter->commitFid,
                        .fHead = {.commitID = pCommitter->commitID, .offset = 0, .size = 0},
                        .fData = {.commitID = pCommitter->commitID, .size = 0},
@@ -1001,10 +992,10 @@ _exit:
 static void tsdbCommitDataEnd(SCommitter *pCommitter) {
   taosArrayDestroy(pCommitter->aBlockIdx);
   tMapDataClear(&pCommitter->oBlockMap);
-  tBlockDataClear(&pCommitter->oBlockData);
+  tBlockDataClear(&pCommitter->oBlockData, 1);
   taosArrayDestroy(pCommitter->aBlockIdxN);
   tMapDataClear(&pCommitter->nBlockMap);
-  tBlockDataClear(&pCommitter->nBlockData);
+  tBlockDataClear(&pCommitter->nBlockData, 1);
   tTSchemaDestroy(pCommitter->skmTable.pTSchema);
   tTSchemaDestroy(pCommitter->skmRow.pTSchema);
 }

source/dnode/vnode/src/tsdb/tsdbFS.c

@@ -698,6 +698,6 @@ void tsdbFSStateDeleteDFileSet(STsdbFSState *pState, int32_t fid) {
 
 SDelFile *tsdbFSStateGetDelFile(STsdbFSState *pState) { return pState->pDelFile; }
 
-SDFileSet *tsdbFSStateGetDFileSet(STsdbFSState *pState, int32_t fid) {
-  return (SDFileSet *)taosArraySearch(pState->aDFileSet, &(SDFileSet){.fid = fid}, tDFileSetCmprFn, TD_EQ);
+SDFileSet *tsdbFSStateGetDFileSet(STsdbFSState *pState, int32_t fid, int32_t flag) {
+  return (SDFileSet *)taosArraySearch(pState->aDFileSet, &(SDFileSet){.fid = fid}, tDFileSetCmprFn, flag);
 }

source/dnode/vnode/src/tsdb/tsdbRead.c

@@ -63,10 +63,10 @@ typedef struct SBlockLoadSuppInfo {
 } SBlockLoadSuppInfo;
 
 typedef struct SFilesetIter {
-  int32_t          numOfFiles;  // number of total files
-  int32_t          index;       // current accessed index in the list
-  SArray*          pFileList;   // data file list
-  int32_t          order;
+  int32_t numOfFiles;  // number of total files
+  int32_t index;       // current accessed index in the list
+  SArray* pFileList;   // data file list
+  int32_t order;
 } SFilesetIter;
 
 typedef struct SFileDataBlockInfo {
@@ -830,8 +830,8 @@ static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockI
   SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
   SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
 
-  int32_t  code = tsdbReadColData(pReader->pFileReader, &pBlockScanInfo->blockIdx, pBlock, pSupInfo->colIds, numOfCols,
-                                  pBlockData, NULL, NULL);
+  int32_t code = tsdbReadColData(pReader->pFileReader, &pBlockScanInfo->blockIdx, pBlock, pSupInfo->colIds, numOfCols,
+                                 pBlockData, NULL, NULL);
   if (code != TSDB_CODE_SUCCESS) {
     goto _error;
   }
@@ -1991,7 +1991,7 @@ static TSDBKEY getCurrentKeyInBuf(SDataBlockIter* pBlockIter, STsdbReader* pRead
 
 static int32_t moveToNextFile(STsdbReader* pReader, int32_t* numOfBlocks) {
   SReaderStatus* pStatus = &pReader->status;
-  SArray* pIndexList = taosArrayInit(4, sizeof(SBlockIdx));
+  SArray*        pIndexList = taosArrayInit(4, sizeof(SBlockIdx));
 
   while (1) {
     bool hasNext = filesetIteratorNext(&pStatus->fileIter, pReader);
@@ -3006,14 +3006,14 @@ SArray* tsdbRetrieveDataBlock(STsdbReader* pReader, SArray* pIdList) {
 
   code = doLoadFileBlockData(pReader, &pStatus->blockIter, pBlockScanInfo, &pStatus->fileBlockData);
   if (code != TSDB_CODE_SUCCESS) {
-    tBlockDataClear(&pStatus->fileBlockData);
+    tBlockDataClear(&pStatus->fileBlockData, 1);
 
     terrno = code;
     return NULL;
   }
 
   copyBlockDataToSDataBlock(pReader, pBlockScanInfo);
-  tBlockDataClear(&pStatus->fileBlockData);
+  tBlockDataClear(&pStatus->fileBlockData, 1);
   return pReader->pResBlock->pDataBlock;
 }
 
@@ -3132,8 +3132,8 @@ int32_t tsdbGetFileBlocksDistInfo(STsdbReader* pReader, STableBlockDistInfo* pTa
       hasNext = (pBlockIter->numOfBlocks > 0);
     }
 
-//    tsdbDebug("%p %d blocks found in file for %d table(s), fid:%d, %s", pReader, numOfBlocks, numOfTables,
-//              pReader->pFileGroup->fid, pReader->idStr);
+    //    tsdbDebug("%p %d blocks found in file for %d table(s), fid:%d, %s", pReader, numOfBlocks, numOfTables,
+    //              pReader->pFileGroup->fid, pReader->idStr);
   }
 
   return code;
@@ -3204,4 +3204,3 @@ int32_t tsdbGetTableSchema(SVnode* pVnode, int64_t uid, STSchema** pSchema, int6
 
   return TSDB_CODE_SUCCESS;
 }
-

source/dnode/vnode/src/tsdb/tsdbReaderWriter.c

@@ -979,21 +979,21 @@ int32_t tsdbReadColData(SDataFReader *pReader, SBlockIdx *pBlockIdx, SBlock *pBl
 
       code = tBlockDataCopy(pBlockData, pBlockData2);
       if (code) {
-        tBlockDataClear(pBlockData1);
-        tBlockDataClear(pBlockData2);
+        tBlockDataClear(pBlockData1, 1);
+        tBlockDataClear(pBlockData2, 1);
         goto _err;
       }
 
       code = tBlockDataMerge(pBlockData1, pBlockData2, pBlockData);
       if (code) {
-        tBlockDataClear(pBlockData1);
-        tBlockDataClear(pBlockData2);
+        tBlockDataClear(pBlockData1, 1);
+        tBlockDataClear(pBlockData2, 1);
         goto _err;
       }
     }
 
-    tBlockDataClear(pBlockData1);
-    tBlockDataClear(pBlockData2);
+    tBlockDataClear(pBlockData1, 1);
+    tBlockDataClear(pBlockData2, 1);
   }
 
   tFree(pBuf1);
@@ -1115,29 +1115,29 @@ int32_t tsdbReadBlockData(SDataFReader *pReader, SBlockIdx *pBlockIdx, SBlock *p
     for (iSubBlock = 1; iSubBlock < pBlock->nSubBlock; iSubBlock++) {
       code = tsdbReadSubBlockData(pReader, pBlockIdx, pBlock, iSubBlock, pBlockData1, ppBuf1, ppBuf2);
       if (code) {
-        tBlockDataClear(pBlockData1);
-        tBlockDataClear(pBlockData2);
+        tBlockDataClear(pBlockData1, 1);
+        tBlockDataClear(pBlockData2, 1);
         goto _err;
       }
 
       code = tBlockDataCopy(pBlockData, pBlockData2);
       if (code) {
-        tBlockDataClear(pBlockData1);
-        tBlockDataClear(pBlockData2);
+        tBlockDataClear(pBlockData1, 1);
+        tBlockDataClear(pBlockData2, 1);
         goto _err;
       }
 
       // merge two block data
       code = tBlockDataMerge(pBlockData1, pBlockData2, pBlockData);
       if (code) {
-        tBlockDataClear(pBlockData1);
-        tBlockDataClear(pBlockData2);
+        tBlockDataClear(pBlockData1, 1);
+        tBlockDataClear(pBlockData2, 1);
         goto _err;
       }
     }
 
-    tBlockDataClear(pBlockData1);
-    tBlockDataClear(pBlockData2);
+    tBlockDataClear(pBlockData1, 1);
+    tBlockDataClear(pBlockData2, 1);
   }
 
   ASSERT(pBlock->nRow == pBlockData->nRow);

source/dnode/vnode/src/tsdb/tsdbUtil.c

@@ -36,15 +36,15 @@ int32_t tMapDataPutItem(SMapData *pMapData, void *pItem, int32_t (*tPutItemFn)(u
 
   // alloc
   code = tRealloc((uint8_t **)&pMapData->aOffset, sizeof(int32_t) * pMapData->nItem);
-  if (code) goto _err;
+  if (code) goto _exit;
   code = tRealloc(&pMapData->pData, pMapData->nData);
-  if (code) goto _err;
+  if (code) goto _exit;
 
   // put
   pMapData->aOffset[nItem] = offset;
   tPutItemFn(pMapData->pData + offset, pItem);
 
-_err:
+_exit:
   return code;
 }
 
@@ -189,25 +189,12 @@ static FORCE_INLINE int32_t tGetTSDBKEY(uint8_t *p, TSDBKEY *pKey) {
 }
 
 // SBlockIdx ======================================================
-void tBlockIdxReset(SBlockIdx *pBlockIdx) {
-  pBlockIdx->minKey = TSKEY_MAX;
-  pBlockIdx->maxKey = TSKEY_MIN;
-  pBlockIdx->minVersion = VERSION_MAX;
-  pBlockIdx->maxVersion = VERSION_MIN;
-  pBlockIdx->offset = -1;
-  pBlockIdx->size = -1;
-}
-
 int32_t tPutBlockIdx(uint8_t *p, void *ph) {
   int32_t    n = 0;
   SBlockIdx *pBlockIdx = (SBlockIdx *)ph;
 
   n += tPutI64(p ? p + n : p, pBlockIdx->suid);
   n += tPutI64(p ? p + n : p, pBlockIdx->uid);
-  n += tPutI64(p ? p + n : p, pBlockIdx->minKey);
-  n += tPutI64(p ? p + n : p, pBlockIdx->maxKey);
-  n += tPutI64v(p ? p + n : p, pBlockIdx->minVersion);
-  n += tPutI64v(p ? p + n : p, pBlockIdx->maxVersion);
   n += tPutI64v(p ? p + n : p, pBlockIdx->offset);
   n += tPutI64v(p ? p + n : p, pBlockIdx->size);
 
@@ -220,10 +207,6 @@ int32_t tGetBlockIdx(uint8_t *p, void *ph) {
 
   n += tGetI64(p + n, &pBlockIdx->suid);
   n += tGetI64(p + n, &pBlockIdx->uid);
-  n += tGetI64(p + n, &pBlockIdx->minKey);
-  n += tGetI64(p + n, &pBlockIdx->maxKey);
-  n += tGetI64v(p + n, &pBlockIdx->minVersion);
-  n += tGetI64v(p + n, &pBlockIdx->maxVersion);
   n += tGetI64v(p + n, &pBlockIdx->offset);
   n += tGetI64v(p + n, &pBlockIdx->size);
 
@@ -921,6 +904,76 @@ _exit:
   return code;
 }
 
+int32_t tPutColData(uint8_t *p, SColData *pColData) {
+  int32_t n = 0;
+
+  n += tPutI16v(p ? p + n : p, pColData->cid);
+  n += tPutI8(p ? p + n : p, pColData->type);
+  n += tPutI8(p ? p + n : p, pColData->smaOn);
+  n += tPutI32v(p ? p + n : p, pColData->nVal);
+  n += tPutU8(p ? p + n : p, pColData->flag);
+
+  if (pColData->flag == HAS_NONE || pColData->flag == HAS_NULL) goto _exit;
+  if (pColData->flag != HAS_VALUE) {
+    // bitmap
+
+    int32_t size = BIT2_SIZE(pColData->nVal);
+    if (p) {
+      memcpy(p + n, pColData->pBitMap, size);
+    }
+    n += size;
+  }
+  if (IS_VAR_DATA_TYPE(pColData->type)) {
+    // offset
+
+    int32_t size = sizeof(int32_t) * pColData->nVal;
+    if (p) {
+      memcpy(p + n, pColData->aOffset, size);
+    }
+    n += size;
+  }
+  n += tPutI32v(p ? p + n : p, pColData->nData);
+  if (p) {
+    memcpy(p + n, pColData->pData, pColData->nData);
+  }
+  n += pColData->nData;
+
+_exit:
+  return n;
+}
+
+int32_t tGetColData(uint8_t *p, SColData *pColData) {
+  int32_t n = 0;
+
+  n += tGetI16v(p + n, &pColData->cid);
+  n += tGetI8(p + n, &pColData->type);
+  n += tGetI8(p + n, &pColData->smaOn);
+  n += tGetI32v(p + n, &pColData->nVal);
+  n += tGetU8(p + n, &pColData->flag);
+
+  if (pColData->flag == HAS_NONE || pColData->flag == HAS_NULL) goto _exit;
+  if (pColData->flag != HAS_VALUE) {
+    // bitmap
+
+    int32_t size = BIT2_SIZE(pColData->nVal);
+    pColData->pBitMap = p + n;
+    n += size;
+  }
+  if (IS_VAR_DATA_TYPE(pColData->type)) {
+    // offset
+
+    int32_t size = sizeof(int32_t) * pColData->nVal;
+    pColData->aOffset = (int32_t *)(p + n);
+    n += size;
+  }
+  n += tGetI32v(p + n, &pColData->nData);
+  pColData->pData = p + n;
+  n += pColData->nData;
+
+_exit:
+  return n;
+}
+
 static FORCE_INLINE int32_t tColDataCmprFn(const void *p1, const void *p2) {
   SColData *pColData1 = (SColData *)p1;
   SColData *pColData2 = (SColData *)p2;
@@ -962,11 +1015,11 @@ void tBlockDataReset(SBlockData *pBlockData) {
   taosArrayClear(pBlockData->aIdx);
 }
 
-void tBlockDataClear(SBlockData *pBlockData) {
+void tBlockDataClear(SBlockData *pBlockData, int8_t deepClear) {
   tFree((uint8_t *)pBlockData->aVersion);
   tFree((uint8_t *)pBlockData->aTSKEY);
   taosArrayDestroy(pBlockData->aIdx);
-  taosArrayDestroyEx(pBlockData->aColData, tColDataClear);
+  taosArrayDestroyEx(pBlockData->aColData, deepClear ? tColDataClear : NULL);
 }
 
 int32_t tBlockDataSetSchema(SBlockData *pBlockData, STSchema *pTSchema) {
@@ -1079,6 +1132,46 @@ _err:
   return code;
 }
 
+int32_t tBlockDataCorrectSchema(SBlockData *pBlockData, SBlockData *pBlockDataFrom) {
+  int32_t code = 0;
+
+  int32_t iColData = 0;
+  for (int32_t iColDataFrom = 0; iColDataFrom < taosArrayGetSize(pBlockDataFrom->aIdx); iColDataFrom++) {
+    SColData *pColDataFrom = tBlockDataGetColDataByIdx(pBlockDataFrom, iColDataFrom);
+
+    while (true) {
+      SColData *pColData;
+      if (iColData < taosArrayGetSize(pBlockData->aIdx)) {
+        pColData = tBlockDataGetColDataByIdx(pBlockData, iColData);
+      } else {
+        pColData = NULL;
+      }
+
+      if (pColData == NULL || pColData->cid > pColDataFrom->cid) {
+        code = tBlockDataAddColData(pBlockData, iColData, &pColData);
+        if (code) goto _exit;
+
+        tColDataInit(pColData, pColDataFrom->cid, pColDataFrom->type, pColDataFrom->smaOn);
+        for (int32_t iRow = 0; iRow < pBlockData->nRow; iRow++) {
+          code = tColDataAppendValue(pColData, &COL_VAL_NONE(pColData->cid, pColData->type));
+          if (code) goto _exit;
+        }
+
+        iColData++;
+        break;
+      } else if (pColData->cid == pColDataFrom->cid) {
+        iColData++;
+        break;
+      } else {
+        iColData++;
+      }
+    }
+  }
+
+_exit:
+  return code;
+}
+
 int32_t tBlockDataMerge(SBlockData *pBlockData1, SBlockData *pBlockData2, SBlockData *pBlockData) {
   int32_t code = 0;
 
@@ -1239,6 +1332,52 @@ void tBlockDataGetColData(SBlockData *pBlockData, int16_t cid, SColData **ppColD
   *ppColData = NULL;
 }
 
+int32_t tPutBlockData(uint8_t *p, SBlockData *pBlockData) {
+  int32_t n = 0;
+
+  n += tPutI32v(p ? p + n : p, pBlockData->nRow);
+  if (p) {
+    memcpy(p + n, pBlockData->aVersion, sizeof(int64_t) * pBlockData->nRow);
+  }
+  n = n + sizeof(int64_t) * pBlockData->nRow;
+  if (p) {
+    memcpy(p + n, pBlockData->aTSKEY, sizeof(TSKEY) * pBlockData->nRow);
+  }
+  n = n + sizeof(TSKEY) * pBlockData->nRow;
+
+  int32_t nCol = taosArrayGetSize(pBlockData->aIdx);
+  n += tPutI32v(p ? p + n : p, nCol);
+  for (int32_t iCol = 0; iCol < nCol; iCol++) {
+    SColData *pColData = tBlockDataGetColDataByIdx(pBlockData, iCol);
+    n += tPutColData(p ? p + n : p, pColData);
+  }
+
+  return n;
+}
+
+int32_t tGetBlockData(uint8_t *p, SBlockData *pBlockData) {
+  int32_t n = 0;
+
+  tBlockDataReset(pBlockData);
+
+  n += tGetI32v(p + n, &pBlockData->nRow);
+  pBlockData->aVersion = (int64_t *)(p + n);
+  n = n + sizeof(int64_t) * pBlockData->nRow;
+  pBlockData->aTSKEY = (TSKEY *)(p + n);
+  n = n + sizeof(TSKEY) * pBlockData->nRow;
+
+  int32_t nCol;
+  n += tGetI32v(p + n, &nCol);
+  for (int32_t iCol = 0; iCol < nCol; iCol++) {
+    SColData *pColData;
+
+    if (tBlockDataAddColData(pBlockData, iCol, &pColData)) return -1;
+    n += tGetColData(p + n, pColData);
+  }
+
+  return n;
+}
+
 // ALGORITHM ==============================
 void tsdbCalcColDataSMA(SColData *pColData, SColumnDataAgg *pColAgg) {
   SColVal  colVal;

source/dnode/vnode/src/vnd/vnodeCommit.c

@@ -233,7 +233,8 @@ int vnodeCommit(SVnode *pVnode) {
   walBeginSnapshot(pVnode->pWal, pVnode->state.applied);
 
   // preCommit
-  smaPreCommit(pVnode->pSma);
+  // smaSyncPreCommit(pVnode->pSma);
+  smaAsyncPreCommit(pVnode->pSma);
 
   // commit each sub-system
   if (metaCommit(pVnode->pMeta) < 0) {
@@ -242,6 +243,8 @@ int vnodeCommit(SVnode *pVnode) {
   }
 
   if (VND_IS_RSMA(pVnode)) {
+    smaAsyncCommit(pVnode->pSma);
+
     if (tsdbCommit(VND_RSMA0(pVnode)) < 0) {
       ASSERT(0);
       return -1;
@@ -276,7 +279,8 @@ int vnodeCommit(SVnode *pVnode) {
   pVnode->state.committed = info.state.committed;
 
   // postCommit
-  smaPostCommit(pVnode->pSma);
+  // smaSyncPostCommit(pVnode->pSma);
+  smaAsyncPostCommit(pVnode->pSma);
 
   // apply the commit (TODO)
   walEndSnapshot(pVnode->pWal);

source/dnode/vnode/src/vnd/vnodeQuery.c

@@ -239,9 +239,9 @@ int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad) {
   pLoad->vgId = TD_VID(pVnode);
   pLoad->syncState = syncGetMyRole(pVnode->sync);
   pLoad->numOfTables = metaGetTbNum(pVnode->pMeta);
-  pLoad->numOfTimeSeries = 400;
-  pLoad->totalStorage = 300;
-  pLoad->compStorage = 200;
+  pLoad->numOfTimeSeries = metaGetTimeSeriesNum(pVnode->pMeta);
+  pLoad->totalStorage = (int64_t)3 * 1073741824;
+  pLoad->compStorage = (int64_t)2 * 1073741824;
   pLoad->pointsWritten = 100;
   pLoad->numOfSelectReqs = 1;
   pLoad->numOfInsertReqs = 3;

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -689,6 +689,7 @@ _exit:
   tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
   tEncodeSVDropTbBatchRsp(&encoder, &rsp);
   tEncoderClear(&encoder);
+  taosArrayDestroy(rsp.pArray);
   return 0;
 }
 
@@ -900,7 +901,7 @@ static int32_t vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *
 _err:
   tDecoderClear(&coder);
   vError("vgId:%d, failed to create tsma %s:%" PRIi64 " version %" PRIi64 "for table %" PRIi64 " since %s",
-         TD_VID(pVnode), req.indexName, req.indexUid, version, req.tableUid, terrstr(terrno));
+         TD_VID(pVnode), req.indexName, req.indexUid, version, req.tableUid, terrstr());
   return -1;
 }
 

source/libs/catalog/src/ctgAsync.c

@@ -163,7 +163,7 @@ int32_t ctgInitGetQnodeTask(SCtgJob *pJob, int32_t taskIdx, void* param) {
 
   taosArrayPush(pJob->pTasks, &task);
 
-  qDebug("QID:0x%" PRIx64 " the %d task type %s initialized", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type));
+  qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type));
 
   return TSDB_CODE_SUCCESS;
 }
@@ -178,7 +178,7 @@ int32_t ctgInitGetDnodeTask(SCtgJob *pJob, int32_t taskIdx, void* param) {
 
   taosArrayPush(pJob->pTasks, &task);
 
-  qDebug("QID:0x%" PRIx64 " the %d task type %s initialized", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type));
+  qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type));
 
   return TSDB_CODE_SUCCESS;
 }
@@ -264,7 +264,7 @@ int32_t ctgInitGetSvrVerTask(SCtgJob *pJob, int32_t taskIdx, void* param) {
 
   taosArrayPush(pJob->pTasks, &task);
 
-  qDebug("QID:0x%" PRIx64 " [%dth] task type %s initialized", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type));
+  qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type));
 
   return TSDB_CODE_SUCCESS;
 }

source/libs/catalog/src/ctgUtil.c

@@ -272,7 +272,7 @@ void ctgFreeHandle(SCatalog* pCtg) {
 
   taosMemoryFree(pCtg);
 
-  ctgInfo("handle freed, culsterId:0x%" PRIx64, clusterId);
+  ctgInfo("handle freed, clusterId:0x%" PRIx64, clusterId);
 }
 
 void ctgClearHandle(SCatalog* pCtg) {
@@ -303,7 +303,7 @@ void ctgClearHandle(SCatalog* pCtg) {
 
   CTG_CACHE_STAT_INC(numOfClear, 1);
 
-  ctgInfo("handle cleared, culsterId:0x%" PRIx64, clusterId);
+  ctgInfo("handle cleared, clusterId:0x%" PRIx64, clusterId);
 }
 
 void ctgFreeSUseDbOutput(SUseDbOutput* pOutput) {

source/libs/executor/src/executil.c

@@ -115,9 +115,6 @@ void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int
     p->groupId = *(uint64_t*)key;
     p->pos = *(SResultRowPosition*)pData;
     memcpy(p->key, (char*)key + sizeof(uint64_t), keyLen - sizeof(uint64_t));
-#ifdef BUF_PAGE_DEBUG
-    qDebug("page_groupRes, groupId:%" PRIu64 ",pageId:%d,offset:%d\n", p->groupId, p->pos.pageId, p->pos.offset);
-#endif
     taosArrayPush(pGroupResInfo->pRows, &p);
   }