Merge remote-tracking branch 'origin/3.0' into feature/3.0_wxy

include/common/tmsg.h

@@ -1533,9 +1533,7 @@ typedef struct SMqSetCVgReq {
   char*    sql;
   char*    logicalPlan;
   char*    physicalPlan;
-  uint32_t qmsgLen;
-  void*    qmsg;
-  //SSubQueryMsg msg;
+  char*    qmsg;
 } SMqSetCVgReq;
 
 static FORCE_INLINE int32_t tEncodeSSubQueryMsg(void** buf, const SSubQueryMsg* pMsg) {
@@ -1567,7 +1565,6 @@ static FORCE_INLINE int32_t tEncodeSMqSetCVgReq(void** buf, const SMqSetCVgReq*
   tlen += taosEncodeString(buf, pReq->sql);
   tlen += taosEncodeString(buf, pReq->logicalPlan);
   tlen += taosEncodeString(buf, pReq->physicalPlan);
-  tlen += taosEncodeFixedU32(buf, pReq->qmsgLen);
   tlen += taosEncodeString(buf, (char*)pReq->qmsg);
   //tlen += tEncodeSSubQueryMsg(buf, &pReq->msg);
   return tlen;
@@ -1582,7 +1579,6 @@ static FORCE_INLINE void* tDecodeSMqSetCVgReq(void* buf, SMqSetCVgReq* pReq) {
   buf = taosDecodeString(buf, &pReq->sql);
   buf = taosDecodeString(buf, &pReq->logicalPlan);
   buf = taosDecodeString(buf, &pReq->physicalPlan);
-  buf = taosDecodeFixedU32(buf, &pReq->qmsgLen);
   buf = taosDecodeString(buf, (char**)&pReq->qmsg);
   //buf = tDecodeSSubQueryMsg(buf, &pReq->msg);
   return buf;
@@ -1600,37 +1596,41 @@ typedef struct SMqColData {
   int16_t colId;
   int16_t type;
   int16_t bytes;
-  char    data[];
-} SMqColData;
+} SMqColMeta;
 
 typedef struct SMqTbData {
   int64_t    uid;
-  int32_t    numOfCols;
   int32_t    numOfRows;
-  SMqColData colData[];
+  char       colData[];
 } SMqTbData;
 
 typedef struct SMqTopicBlk {
-  char      topicName[TSDB_TOPIC_FNAME_LEN];
-  int64_t   committedOffset;
-  int64_t   reqOffset;
-  int64_t   rspOffset;
-  int32_t   skipLogNum;
-  int32_t   bodyLen;
-  int32_t   numOfTb;
-  SMqTbData tbData[];
+  char       topicName[TSDB_TOPIC_FNAME_LEN];
+  int64_t    committedOffset;
+  int64_t    reqOffset;
+  int64_t    rspOffset;
+  int32_t    skipLogNum;
+  int32_t    bodyLen;
+  int32_t    numOfTb;
+  SMqTbData* tbData;
 } SMqTopicData;
 
 typedef struct SMqConsumeRsp {
-  int64_t      reqId;
-  int64_t      consumerId;
-  int32_t      bodyLen;
-  int32_t      numOfTopics;
-  SMqTopicData data[];
+  int64_t       consumerId;
+  int32_t       numOfCols;
+  SMqColMeta*   meta;
+  int32_t       numOfTopics;
+  SMqTopicData* data;
 } SMqConsumeRsp;
 
+static FORCE_INLINE int32_t tEncodeSMqConsumeRsp(void** buf, const SMqConsumeRsp* pRsp) {
+  int32_t tlen = 0;
+  return tlen;
+}
+
 // one req for one vg+topic
 typedef struct SMqConsumeReq {
+  SMsgHead       head;
   //0: commit only, current offset
   //1: consume only, poll next offset
   //2: commit current and consume next offset
@@ -1663,7 +1663,7 @@ typedef struct SMqCMGetSubEpRsp {
 
 static FORCE_INLINE int32_t tEncodeSMqSubVgEp(void** buf, const SMqSubVgEp* pVgEp) {
   int32_t tlen = 0;
-  tlen += taosEncodeFixedI16(buf, pVgEp->vgId);
+  tlen += taosEncodeFixedI32(buf, pVgEp->vgId);
   tlen += taosEncodeSEpSet(buf, &pVgEp->epSet);
   return tlen;
 }

include/libs/executor/executor.h

@@ -24,6 +24,7 @@ extern "C" {
 
 typedef void* qTaskInfo_t;
 typedef void* DataSinkHandle;
+struct SRpcMsg;
 struct SSubplan;
 
  /**
@@ -208,6 +209,8 @@ void** qReleaseTask(void* pMgmt, void* pQInfo, bool freeHandle);
  */
 void** qDeregisterQInfo(void* pMgmt, void* pQInfo);
 
+void qProcessFetchRsp(void* parent, struct SRpcMsg* pMsg, struct SEpSet* pEpSet);
+
 #ifdef __cplusplus
 }
 #endif

include/libs/qworker/qworker.h

@@ -49,9 +49,10 @@ typedef struct {
 } SQWorkerStat;
 
 typedef int32_t (*putReqToQueryQFp)(void *, struct SRpcMsg *);
+typedef int32_t (*sendReqToDnodeFp)(void *, struct SEpSet *, struct SRpcMsg *);
 
-
-int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, void *nodeObj, putReqToQueryQFp fp);
+int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, void *nodeObj,
+                    putReqToQueryQFp fp1, sendReqToDnodeFp fp2);
 
 int32_t qWorkerProcessQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
 
@@ -65,6 +66,8 @@ int32_t qWorkerProcessStatusMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
 
 int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
 
+int32_t qWorkerProcessFetchRsp(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
+
 int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);
 
 int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg);

include/libs/wal/wal.h

@@ -19,6 +19,7 @@
 #include "tarray.h"
 #include "tdef.h"
 #include "tlog.h"
+#include "tmsg.h"
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -159,7 +160,7 @@ int32_t walAlter(SWal *, SWalCfg *pCfg);
 void    walClose(SWal *);
 
 // write
-int64_t walWrite(SWal *, int64_t index, uint8_t msgType, const void *body, int32_t bodyLen);
+int64_t walWrite(SWal *, int64_t index, tmsg_t msgType, const void *body, int32_t bodyLen);
 void    walFsync(SWal *, bool force);
 
 // apis for lifecycle management

include/util/taoserror.h

@@ -253,6 +253,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_MND_INVALID_TOPIC_OPTION      TAOS_DEF_ERROR_CODE(0, 0x03E4)
 #define TSDB_CODE_MND_TOPIC_OPTION_UNCHNAGED    TAOS_DEF_ERROR_CODE(0, 0x03E5)
 #define TSDB_CODE_MND_NAME_CONFLICT_WITH_STB    TAOS_DEF_ERROR_CODE(0, 0x03E6)
+#define TSDB_CODE_MND_CONSUMER_NOT_EXIST        TAOS_DEF_ERROR_CODE(0, 0x03E7)
 
 // dnode
 #define TSDB_CODE_DND_ACTION_IN_PROGRESS        TAOS_DEF_ERROR_CODE(0, 0x0400)

source/client/src/clientImpl.c

@@ -328,6 +328,7 @@ struct tmq_t {
   char           clientId[256];
   int64_t        consumerId;
   int64_t        status;
+  tsem_t         rspSem;
   STscObj*       pTscObj;
   tmq_commit_cb* commit_cb;
   int32_t        nextTopicIdx;
@@ -344,6 +345,7 @@ tmq_t* taos_consumer_new(void* conn, tmq_conf_t* conf, char* errstr, int32_t err
   strcpy(pTmq->clientId, conf->clientId);
   strcpy(pTmq->groupId, conf->groupId);
   pTmq->commit_cb = conf->commit_cb;
+  tsem_init(&pTmq->rspSem, 0, 0);
   pTmq->consumerId = generateRequestId() & ((uint64_t)-1 >> 1);
   pTmq->clientTopics = taosArrayInit(0, sizeof(SMqClientTopic));
   return pTmq;
@@ -372,11 +374,27 @@ int32_t tmq_list_append(tmq_list_t* ptr, char* src) {
 }
 
 
+int32_t tmq_null_cb(void* param, const SDataBuf* pMsg, int32_t code)  {
+  if (code == 0) {
+    //
+  }
+  //
+  return 0;
+}
+
 TAOS_RES* tmq_subscribe(tmq_t* tmq, tmq_list_t* topic_list) {
   SRequestObj *pRequest = NULL;
-  tmq->status = 1;
   int32_t sz = topic_list->cnt;
-  tmq->clientTopics = taosArrayInit(sz, sizeof(void*));
+  //destroy ex
+  taosArrayDestroy(tmq->clientTopics);
+  tmq->clientTopics = taosArrayInit(sz, sizeof(SMqClientTopic));
+
+  SCMSubscribeReq req;
+  req.topicNum = sz;
+  req.consumerId = tmq->consumerId;
+  req.consumerGroup = strdup(tmq->groupId);
+  req.topicNames = taosArrayInit(sz, sizeof(void*));
+
   for (int i = 0; i < sz; i++) {
     char* topicName = topic_list->elems[i];
 
@@ -391,16 +409,21 @@ TAOS_RES* tmq_subscribe(tmq_t* tmq, tmq_list_t* topic_list) {
     }
     tNameExtractFullName(&name, topicFname);
     tscDebug("subscribe topic: %s", topicFname);
-    taosArrayPush(tmq->clientTopics, &topicFname); 
+    SMqClientTopic topic = {
+      .nextVgIdx = 0,
+      .sql = NULL,
+      .sqlLen = 0,
+      .topicId = 0,
+      .topicName = topicFname,
+      .vgs = NULL
+    };
+    topic.vgs = taosArrayInit(0, sizeof(SMqClientVg));
+    taosArrayPush(tmq->clientTopics, &topic); 
     /*SMqClientTopic topic = {*/
       /*.*/
     /*};*/
+    taosArrayPush(req.topicNames, &topicFname);
   }
-  SCMSubscribeReq req;
-  req.topicNum = taosArrayGetSize(tmq->clientTopics);
-  req.consumerId = tmq->consumerId;
-  req.consumerGroup = strdup(tmq->groupId);
-  req.topicNames = tmq->clientTopics;
 
   int tlen = tSerializeSCMSubscribeReq(NULL, &req);
   void* buf = malloc(tlen);
@@ -419,18 +442,19 @@ TAOS_RES* tmq_subscribe(tmq_t* tmq, tmq_list_t* topic_list) {
 
   pRequest->body.requestMsg = (SDataBuf){ .pData = buf, .len = tlen };
 
-  SMsgSendInfo* body = buildMsgInfoImpl(pRequest);
+  SMsgSendInfo* sendInfo = buildMsgInfoImpl(pRequest);
+  /*sendInfo->fp*/
   SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);
 
   int64_t transporterId = 0;
-  asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, body);
+  asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo);
 
   tsem_wait(&pRequest->body.rspSem);
 
 _return:
-  if (body != NULL) {
-    destroySendMsgInfo(body);
-  }
+  /*if (sendInfo != NULL) {*/
+    /*destroySendMsgInfo(sendInfo);*/
+  /*}*/
 
   if (pRequest != NULL && terrno != TSDB_CODE_SUCCESS) {
     pRequest->code = terrno;
@@ -569,19 +593,19 @@ TAOS_RES *taos_create_topic(TAOS* taos, const char* topicName, const char* sql,
   pRequest->body.requestMsg = (SDataBuf){ .pData = buf, .len = tlen };
   pRequest->type = TDMT_MND_CREATE_TOPIC;
 
-  SMsgSendInfo* body = buildMsgInfoImpl(pRequest);
+  SMsgSendInfo* sendInfo = buildMsgInfoImpl(pRequest);
   SEpSet epSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
 
   int64_t transporterId = 0;
-  asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, body);
+  asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo);
 
   tsem_wait(&pRequest->body.rspSem);
 
 _return:
   qDestroyQuery(pQueryNode);
-  if (body != NULL) {
-    destroySendMsgInfo(body);
-  }
+  /*if (sendInfo != NULL) {*/
+    /*destroySendMsgInfo(sendInfo);*/
+  /*}*/
 
   if (pRequest != NULL && terrno != TSDB_CODE_SUCCESS) {
     pRequest->code = terrno;
@@ -602,6 +626,12 @@ int32_t tmq_poll_cb_inner(void* param, const SDataBuf* pMsg, int32_t code) {
 
 int32_t tmq_ask_ep_cb(void* param, const SDataBuf* pMsg, int32_t code) {
   tmq_t* tmq = (tmq_t*)param;
+  if (code != 0) {
+    tsem_post(&tmq->rspSem);
+    return 0;
+  }
+  tscDebug("tmq ask ep cb called");
+  bool set = false;
   SMqCMGetSubEpRsp rsp;
   tDecodeSMqCMGetSubEpRsp(pMsg->pData, &rsp);
   int32_t sz = taosArrayGetSize(rsp.topics);
@@ -616,38 +646,35 @@ int32_t tmq_ask_ep_cb(void* param, const SDataBuf* pMsg, int32_t code) {
     for (int32_t j = 0; j < vgSz; j++) {
       SMqSubVgEp* pVgEp = taosArrayGet(pTopicEp->vgs, j);
       SMqClientVg clientVg = {
+        .pollCnt = 0,
+        .committedOffset = -1,
+        .currentOffset = -1,
         .vgId = pVgEp->vgId,
         .epSet = pVgEp->epSet
       };
       taosArrayPush(topic.vgs, &clientVg);
+      set = true;
     }
     taosArrayPush(tmq->clientTopics, &topic);
   }
+  if(set) tmq->status = 1;
   // unlock
+  tsem_post(&tmq->rspSem);
   return 0;
 }
 
 tmq_message_t* tmq_consume_poll(tmq_t* tmq, int64_t blocking_time) {
-  if (tmq->clientTopics == NULL || taosArrayGetSize(tmq->clientTopics) == 0) {
-    return NULL;
-  }
-  SRequestObj *pRequest = NULL;
-  SMqConsumeReq req = {0};
-  req.reqType = 1;
-  req.blockingTime = blocking_time;
-  req.consumerId = tmq->consumerId;
-  tmq_message_t* tmq_message = NULL;
-  strcpy(req.cgroup, tmq->groupId);
 
-  if (taosArrayGetSize(tmq->clientTopics) == 0) {
+  if (taosArrayGetSize(tmq->clientTopics) == 0 || tmq->status == 0) {
     int32_t tlen = sizeof(SMqCMGetSubEpReq);
     SMqCMGetSubEpReq* buf = malloc(tlen);
     if (buf == NULL) {
       tscError("failed to malloc get subscribe ep buf");
     }
-    buf->consumerId = htobe64(buf->consumerId);
+    buf->consumerId = htobe64(tmq->consumerId);
+    strcpy(buf->cgroup, tmq->groupId);
     
-    pRequest = createRequest(tmq->pTscObj, NULL, NULL, TDMT_MND_GET_SUB_EP);
+    SRequestObj *pRequest = createRequest(tmq->pTscObj, NULL, NULL, TDMT_MND_GET_SUB_EP);
     if (pRequest == NULL) {
       tscError("failed to malloc subscribe ep request");
     }
@@ -664,24 +691,39 @@ tmq_message_t* tmq_consume_poll(tmq_t* tmq, int64_t blocking_time) {
     int64_t transporterId = 0;
     asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo);
 
-    tsem_wait(&pRequest->body.rspSem);
+    tsem_wait(&tmq->rspSem);
   }
 
-  SMqClientTopic* pTopic = taosArrayGetP(tmq->clientTopics, tmq->nextTopicIdx);
+  if (taosArrayGetSize(tmq->clientTopics) == 0) {
+    tscDebug("consumer:%ld poll but not assigned", tmq->consumerId);
+    return NULL;
+  }
+
+  SMqConsumeReq* pReq = malloc(sizeof(SMqConsumeReq));
+  pReq->reqType = 1;
+  pReq->blockingTime = blocking_time;
+  pReq->consumerId = tmq->consumerId;
+  tmq_message_t* tmq_message = NULL;
+  strcpy(pReq->cgroup, tmq->groupId);
+
+  SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, tmq->nextTopicIdx);
   tmq->nextTopicIdx = (tmq->nextTopicIdx + 1) % taosArrayGetSize(tmq->clientTopics);
-  strcpy(req.topic, pTopic->topicName);
+  strcpy(pReq->topic, pTopic->topicName);
   int32_t nextVgIdx = pTopic->nextVgIdx;
   pTopic->nextVgIdx = (nextVgIdx + 1) % taosArrayGetSize(pTopic->vgs);
   SMqClientVg* pVg = taosArrayGet(pTopic->vgs, nextVgIdx);
-  req.offset = pVg->currentOffset;
+  pReq->offset = pVg->currentOffset;
+
+  pReq->head.vgId = htonl(pVg->vgId);
+  pReq->head.contLen = htonl(sizeof(SMqConsumeReq));
 
-  pRequest->body.requestMsg = (SDataBuf){ .pData = &req, .len = sizeof(SMqConsumeReq) };
-  pRequest->type = TDMT_VND_CONSUME;
+  SRequestObj* pRequest = createRequest(tmq->pTscObj, NULL, NULL, TDMT_VND_CONSUME);
+  pRequest->body.requestMsg = (SDataBuf){ .pData = pReq, .len = sizeof(SMqConsumeReq) };
 
   SMsgSendInfo* sendInfo = buildMsgInfoImpl(pRequest);
-  sendInfo->requestObjRefId = 0;
-  sendInfo->param = &tmq_message;
-  sendInfo->fp = tmq_poll_cb_inner;
+  /*sendInfo->requestObjRefId = 0;*/
+  /*sendInfo->param = &tmq_message;*/
+  /*sendInfo->fp = tmq_poll_cb_inner;*/
 
   int64_t transporterId = 0;
   asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &pVg->epSet, &transporterId, sendInfo);

source/client/test/clientTests.cpp

@@ -536,7 +536,7 @@ TEST(testCase, create_topic_Test) {
   if (taos_errno(pRes) != 0) {
     printf("error in use db, reason:%s\n", taos_errstr(pRes));
   }
-  taos_free_result(pRes);
+  //taos_free_result(pRes);
 
   TAOS_FIELD* pFields = taos_fetch_fields(pRes);
   ASSERT_TRUE(pFields == nullptr);
@@ -570,30 +570,32 @@ TEST(testCase, create_topic_Test) {
   //taos_close(pConn);
 //}
 
-//TEST(testCase, tmq_subscribe_Test) {
-  //TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-  //assert(pConn != NULL);
+#if 0
+TEST(testCase, tmq_subscribe_Test) {
+  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  assert(pConn != NULL);
 
-  //TAOS_RES* pRes = taos_query(pConn, "use abc1");
-  //if (taos_errno(pRes) != 0) {
-    //printf("error in use db, reason:%s\n", taos_errstr(pRes));
-  //}
-  //taos_free_result(pRes);
+  TAOS_RES* pRes = taos_query(pConn, "use abc1");
+  if (taos_errno(pRes) != 0) {
+    printf("error in use db, reason:%s\n", taos_errstr(pRes));
+  }
+  taos_free_result(pRes);
 
-  //tmq_conf_t* conf = tmq_conf_new();
-  //tmq_conf_set(conf, "group.id", "tg1");
-  //tmq_t* tmq = taos_consumer_new(pConn, conf, NULL, 0);
+  tmq_conf_t* conf = tmq_conf_new();
+  tmq_conf_set(conf, "group.id", "tg1");
+  tmq_t* tmq = taos_consumer_new(pConn, conf, NULL, 0);
   
-  //tmq_list_t* topic_list = tmq_list_new();
-  //tmq_list_append(topic_list, "test_topic_1");
-  //tmq_subscribe(tmq, topic_list);
+  tmq_list_t* topic_list = tmq_list_new();
+  tmq_list_append(topic_list, "test_topic_1");
+  tmq_subscribe(tmq, topic_list);
 
-  //while (1) {
-    //tmq_message_t* msg = tmq_consume_poll(tmq, 0);
-    //printf("get msg\n");
+  while (1) {
+    tmq_message_t* msg = tmq_consume_poll(tmq, 0);
+    printf("get msg\n");
     //if (msg == NULL) break;
-  //}
-//}
+  }
+}
+#endif
 
 TEST(testCase, tmq_consume_Test) {
 }

source/dnode/mgmt/impl/src/dndEnv.c

@@ -289,6 +289,7 @@ int32_t dndInit(const SDnodeEnvCfg *pCfg) {
       .charset = pCfg->charset,
       .nthreads = pCfg->numOfCommitThreads,
       .putReqToVQueryQFp = dndPutReqToVQueryQ,
+      .sendReqToDnodeFp = dndSendReqToDnode
   };
 
   if (vnodeInit(&vnodeOpt) != 0) {

source/dnode/mgmt/impl/src/dndMgmt.c

@@ -529,7 +529,7 @@ int32_t dndInitMgmt(SDnode *pDnode) {
   }
 
   if (pMgmt->dropped) {
-    dError("dnode will not start for its already dropped");
+    dError("dnode not start since its already dropped");
     return -1;
   }
 

source/dnode/mgmt/impl/src/dndTransport.c

@@ -122,6 +122,7 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_QUERY)] = dndProcessVnodeQueryMsg;
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_QUERY_CONTINUE)] = dndProcessVnodeQueryMsg;
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_FETCH)] = dndProcessVnodeFetchMsg;
+  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_FETCH_RSP)] = dndProcessVnodeFetchMsg;
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_ALTER_TABLE)] = dndProcessVnodeWriteMsg;
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_UPDATE_TAG_VAL)] = dndProcessVnodeWriteMsg;
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_TABLE_META)] = dndProcessVnodeFetchMsg;
@@ -148,6 +149,7 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES_FETCH)] = dndProcessVnodeFetchMsg;
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_SET_CONN)] = dndProcessVnodeWriteMsg;
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_SET_CUR)] = dndProcessVnodeFetchMsg;
+  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_CONSUME)] = dndProcessVnodeFetchMsg;
 }
 
 static void dndProcessResponse(void *parent, SRpcMsg *pRsp, SEpSet *pEpSet) {

source/dnode/mgmt/impl/src/dndVnodes.c

@@ -527,7 +527,6 @@ static void dndGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
   pCfg->vgId = pCreate->vgId;
   pCfg->wsize = pCreate->cacheBlockSize;
   pCfg->ssize = pCreate->cacheBlockSize;
-  pCfg->wsize = pCreate->cacheBlockSize;
   pCfg->lsize = pCreate->cacheBlockSize;
   pCfg->isHeapAllocator = true;
   pCfg->ttl = 4;

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

@@ -363,9 +363,7 @@ typedef struct SMqConsumerEp {
   int64_t      consumerId;  // -1 for unassigned
   int64_t      lastConsumerHbTs;
   int64_t      lastVgHbTs;
-  uint32_t     qmsgLen;
   char*        qmsg;
-  //SSubQueryMsg qExec;
 } SMqConsumerEp;
 
 static FORCE_INLINE int32_t tEncodeSMqConsumerEp(void** buf, SMqConsumerEp* pConsumerEp) {
@@ -374,9 +372,10 @@ static FORCE_INLINE int32_t tEncodeSMqConsumerEp(void** buf, SMqConsumerEp* pCon
   tlen += taosEncodeFixedI32(buf, pConsumerEp->status);
   tlen += taosEncodeSEpSet(buf, &pConsumerEp->epSet);
   tlen += taosEncodeFixedI64(buf, pConsumerEp->consumerId);
+  tlen += taosEncodeFixedI64(buf, pConsumerEp->lastConsumerHbTs);
+  tlen += taosEncodeFixedI64(buf, pConsumerEp->lastVgHbTs);
   //tlen += tEncodeSSubQueryMsg(buf, &pConsumerEp->qExec);
-  tlen += taosEncodeFixedU32(buf, pConsumerEp->qmsgLen);
-  tlen += taosEncodeBinary(buf, pConsumerEp->qmsg, pConsumerEp->qmsgLen);
+  tlen += taosEncodeString(buf, pConsumerEp->qmsg);
   return tlen;
 }
 
@@ -385,9 +384,10 @@ static FORCE_INLINE void* tDecodeSMqConsumerEp(void** buf, SMqConsumerEp* pConsu
   buf = taosDecodeFixedI32(buf, &pConsumerEp->status);
   buf = taosDecodeSEpSet(buf, &pConsumerEp->epSet);
   buf = taosDecodeFixedI64(buf, &pConsumerEp->consumerId);
+  buf = taosDecodeFixedI64(buf, &pConsumerEp->lastConsumerHbTs);
+  buf = taosDecodeFixedI64(buf, &pConsumerEp->lastVgHbTs);
   //buf = tDecodeSSubQueryMsg(buf, &pConsumerEp->qExec);
-  buf = taosDecodeFixedU32(buf, &pConsumerEp->qmsgLen);
-  buf = taosDecodeBinary(buf, (void**)&pConsumerEp->qmsg, pConsumerEp->qmsgLen);
+  buf = taosDecodeString(buf, &pConsumerEp->qmsg);
   return buf;
 }
 
@@ -423,18 +423,27 @@ static FORCE_INLINE SMqSubscribeObj* tNewSubscribeObj() {
     free(pSub);
     return NULL;
   }
+  pSub->lostConsumer = taosArrayInit(0, sizeof(SMqConsumerEp));
+  if (pSub->lostConsumer == NULL) {
+    taosArrayDestroy(pSub->availConsumer);
+    taosArrayDestroy(pSub->assigned);
+    free(pSub);
+    return NULL;
+  }
   pSub->idleConsumer = taosArrayInit(0, sizeof(SMqConsumerEp));
-  if (pSub->assigned == NULL) {
+  if (pSub->idleConsumer == NULL) {
     taosArrayDestroy(pSub->availConsumer);
-    taosArrayDestroy(pSub->idleConsumer);
+    taosArrayDestroy(pSub->assigned);
+    taosArrayDestroy(pSub->lostConsumer);
     free(pSub);
     return NULL;
   }
   pSub->unassignedVg = taosArrayInit(0, sizeof(SMqConsumerEp));
-  if (pSub->assigned == NULL) {
+  if (pSub->unassignedVg == NULL) {
     taosArrayDestroy(pSub->availConsumer);
+    taosArrayDestroy(pSub->assigned);
+    taosArrayDestroy(pSub->lostConsumer);
     taosArrayDestroy(pSub->idleConsumer);
-    taosArrayDestroy(pSub->unassignedVg);
     free(pSub);
     return NULL;
   }
@@ -461,6 +470,13 @@ static FORCE_INLINE int32_t tEncodeSubscribeObj(void** buf, const SMqSubscribeOb
     tlen += tEncodeSMqConsumerEp(buf, pCEp);
   }
 
+  sz = taosArrayGetSize(pSub->lostConsumer);
+  tlen += taosEncodeFixedI32(buf, sz);
+  for (int32_t i = 0; i < sz; i++) {
+    SMqConsumerEp* pCEp = taosArrayGet(pSub->lostConsumer, i);
+    tlen += tEncodeSMqConsumerEp(buf, pCEp);
+  }
+
   sz = taosArrayGetSize(pSub->idleConsumer);
   tlen += taosEncodeFixedI32(buf, sz);
   for (int32_t i = 0; i < sz; i++) {
@@ -485,20 +501,47 @@ static FORCE_INLINE void* tDecodeSubscribeObj(void* buf, SMqSubscribeObj* pSub)
   int32_t sz;
 
   buf = taosDecodeFixedI32(buf, &sz);
-  pSub->assigned = taosArrayInit(sz, sizeof(int64_t));
-  if (pSub->assigned == NULL) {
+  pSub->availConsumer = taosArrayInit(sz, sizeof(int64_t));
+  if (pSub->availConsumer == NULL) {
     return NULL;
   }
   for (int32_t i = 0; i < sz; i++) {
     int64_t consumerId;
     buf = taosDecodeFixedI64(buf, &consumerId);
-    taosArrayPush(pSub->assigned, &consumerId);
+    taosArrayPush(pSub->availConsumer, &consumerId);
+  }
+
+  buf = taosDecodeFixedI32(buf, &sz);
+  pSub->assigned = taosArrayInit(sz, sizeof(SMqConsumerEp));
+  if (pSub->assigned == NULL) {
+    taosArrayDestroy(pSub->availConsumer);
+    return NULL;
+  }
+  for (int32_t i = 0; i < sz; i++) {
+    SMqConsumerEp cEp;
+    buf = tDecodeSMqConsumerEp(buf, &cEp);
+    taosArrayPush(pSub->assigned, &cEp);
+  }
+
+  buf = taosDecodeFixedI32(buf, &sz);
+  pSub->lostConsumer = taosArrayInit(sz, sizeof(SMqConsumerEp));
+  if (pSub->lostConsumer == NULL) {
+    taosArrayDestroy(pSub->availConsumer);
+    taosArrayDestroy(pSub->assigned);
+    return NULL;
+  }
+  for (int32_t i = 0; i < sz; i++) {
+    SMqConsumerEp cEp;
+    buf = tDecodeSMqConsumerEp(buf, &cEp);
+    taosArrayPush(pSub->lostConsumer, &cEp);
   }
 
   buf = taosDecodeFixedI32(buf, &sz);
   pSub->idleConsumer = taosArrayInit(sz, sizeof(SMqConsumerEp));
   if (pSub->idleConsumer == NULL) {
+    taosArrayDestroy(pSub->availConsumer);
     taosArrayDestroy(pSub->assigned);
+    taosArrayDestroy(pSub->lostConsumer);
     return NULL;
   }
   for (int32_t i = 0; i < sz; i++) {
@@ -507,10 +550,13 @@ static FORCE_INLINE void* tDecodeSubscribeObj(void* buf, SMqSubscribeObj* pSub)
     taosArrayPush(pSub->idleConsumer, &cEp);
   }
 
+
   buf = taosDecodeFixedI32(buf, &sz);
   pSub->unassignedVg = taosArrayInit(sz, sizeof(SMqConsumerEp));
   if (pSub->unassignedVg == NULL) {
+    taosArrayDestroy(pSub->availConsumer);
     taosArrayDestroy(pSub->assigned);
+    taosArrayDestroy(pSub->lostConsumer);
     taosArrayDestroy(pSub->idleConsumer);
     return NULL;
   }
@@ -580,7 +626,10 @@ static FORCE_INLINE int32_t tEncodeSMqConsumerTopic(void** buf, SMqConsumerTopic
   int32_t tlen = 0;
   tlen += taosEncodeString(buf, pConsumerTopic->name);
   tlen += taosEncodeFixedI32(buf, pConsumerTopic->epoch);
-  int32_t sz = taosArrayGetSize(pConsumerTopic->pVgInfo);
+  int32_t sz = 0;
+  if (pConsumerTopic->pVgInfo != NULL) {
+    sz = taosArrayGetSize(pConsumerTopic->pVgInfo);
+  }
   tlen += taosEncodeFixedI32(buf, sz);
   for (int32_t i = 0; i < sz; i++) {
     int32_t* pVgInfo = taosArrayGet(pConsumerTopic->pVgInfo, i);

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

@@ -226,10 +226,10 @@ static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
   if (pCfg->cacheBlockSize < TSDB_MIN_CACHE_BLOCK_SIZE || pCfg->cacheBlockSize > TSDB_MAX_CACHE_BLOCK_SIZE) return -1;
   if (pCfg->totalBlocks < TSDB_MIN_TOTAL_BLOCKS || pCfg->totalBlocks > TSDB_MAX_TOTAL_BLOCKS) return -1;
   if (pCfg->daysPerFile < TSDB_MIN_DAYS_PER_FILE || pCfg->daysPerFile > TSDB_MAX_DAYS_PER_FILE) return -1;
-  if (pCfg->daysToKeep0 < pCfg->daysPerFile) return -1;
   if (pCfg->daysToKeep0 < TSDB_MIN_KEEP || pCfg->daysToKeep0 > TSDB_MAX_KEEP) return -1;
   if (pCfg->daysToKeep1 < TSDB_MIN_KEEP || pCfg->daysToKeep1 > TSDB_MAX_KEEP) return -1;
   if (pCfg->daysToKeep2 < TSDB_MIN_KEEP || pCfg->daysToKeep2 > TSDB_MAX_KEEP) return -1;
+  if (pCfg->daysToKeep0 < pCfg->daysPerFile) return -1;
   if (pCfg->daysToKeep0 > pCfg->daysToKeep1) return -1;
   if (pCfg->daysToKeep1 > pCfg->daysToKeep2) return -1;
   if (pCfg->minRows < TSDB_MIN_MIN_ROW_FBLOCK || pCfg->minRows > TSDB_MAX_MIN_ROW_FBLOCK) return -1;
@@ -498,7 +498,7 @@ static int32_t mndProcessCreateDbReq(SMnodeMsg *pReq) {
   return TSDB_CODE_MND_ACTION_IN_PROGRESS;
 }
 
-static int32_t mndSetDbCfgFromAlterDbMsg(SDbObj *pDb, SAlterDbReq *pAlter) {
+static int32_t mndSetDbCfgFromAlterDbReq(SDbObj *pDb, SAlterDbReq *pAlter) {
   terrno = TSDB_CODE_MND_DB_OPTION_UNCHANGED;
 
   if (pAlter->totalBlocks >= 0 && pAlter->totalBlocks != pDb->cfg.totalBlocks) {
@@ -649,7 +649,7 @@ static int32_t mndProcessAlterDbReq(SMnodeMsg *pReq) {
   SDbObj dbObj = {0};
   memcpy(&dbObj, pDb, sizeof(SDbObj));
 
-  int32_t code = mndSetDbCfgFromAlterDbMsg(&dbObj, pAlter);
+  int32_t code = mndSetDbCfgFromAlterDbReq(&dbObj, pAlter);
   if (code != 0) {
     mndReleaseDb(pMnode, pDb);
     mError("db:%s, failed to alter since %s", pAlter->db, tstrerror(code));
@@ -1143,7 +1143,7 @@ static int32_t mndRetrieveDbs(SMnodeMsg *pReq, SShowObj *pShow, char *data, int3
         prec = TSDB_TIME_PRECISION_NANO_STR;
         break;
       default:
-        assert(false);
+        prec = "none";
         break;
     }
     STR_WITH_SIZE_TO_VARSTR(pWrite, prec, 2);

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

@@ -20,6 +20,7 @@
 #include "mndTrans.h"
 
 #define SDB_FUNC_VER 1
+#define SDB_FUNC_RESERVE_SIZE 64
 
 static SSdbRaw *mndFuncActionEncode(SFuncObj *pFunc);
 static SSdbRow *mndFuncActionDecode(SSdbRaw *pRaw);
@@ -60,7 +61,7 @@ void mndCleanupFunc(SMnode *pMnode) {}
 static SSdbRaw *mndFuncActionEncode(SFuncObj *pFunc) {
   terrno = TSDB_CODE_OUT_OF_MEMORY;
 
-  int32_t  size = pFunc->commentSize + pFunc->codeSize + sizeof(SFuncObj);
+  int32_t  size = pFunc->commentSize + pFunc->codeSize + sizeof(SFuncObj) + SDB_FUNC_RESERVE_SIZE;
   SSdbRaw *pRaw = sdbAllocRaw(SDB_FUNC, SDB_FUNC_VER, size);
   if (pRaw == NULL) goto FUNC_ENCODE_OVER;
 
@@ -78,6 +79,7 @@ static SSdbRaw *mndFuncActionEncode(SFuncObj *pFunc) {
   SDB_SET_INT32(pRaw, dataPos, pFunc->codeSize, FUNC_ENCODE_OVER)
   SDB_SET_BINARY(pRaw, dataPos, pFunc->pComment, pFunc->commentSize, FUNC_ENCODE_OVER)
   SDB_SET_BINARY(pRaw, dataPos, pFunc->pCode, pFunc->codeSize, FUNC_ENCODE_OVER)
+  SDB_SET_RESERVE(pRaw, dataPos, SDB_FUNC_RESERVE_SIZE, FUNC_ENCODE_OVER)
   SDB_SET_DATALEN(pRaw, dataPos, FUNC_ENCODE_OVER);
 
   terrno = 0;
@@ -131,6 +133,7 @@ static SSdbRow *mndFuncActionDecode(SSdbRaw *pRaw) {
 
   SDB_GET_BINARY(pRaw, dataPos, pFunc->pComment, pFunc->commentSize, FUNC_DECODE_OVER)
   SDB_GET_BINARY(pRaw, dataPos, pFunc->pCode, pFunc->codeSize, FUNC_DECODE_OVER)
+  SDB_GET_RESERVE(pRaw, dataPos, SDB_FUNC_RESERVE_SIZE, FUNC_DECODE_OVER)
 
   terrno = 0;
 

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

@@ -66,13 +66,13 @@ int32_t mndInitSubscribe(SMnode *pMnode) {
 
 static int32_t mndProcessGetSubEpReq(SMnodeMsg *pMsg) {
   SMnode           *pMnode = pMsg->pMnode;
-  SMqCMGetSubEpReq *pReq = (SMqCMGetSubEpReq *)pMsg->pCont;
+  SMqCMGetSubEpReq *pReq = (SMqCMGetSubEpReq *)pMsg->rpcMsg.pCont;
   SMqCMGetSubEpRsp  rsp;
   int64_t           consumerId = be64toh(pReq->consumerId);
 
   SMqConsumerObj *pConsumer = mndAcquireConsumer(pMsg->pMnode, consumerId);
   if (pConsumer == NULL) {
-    /*terrno = */
+    terrno = TSDB_CODE_MND_CONSUMER_NOT_EXIST;
     return -1;
   }
   ASSERT(strcmp(pReq->cgroup, pConsumer->cgroup) == 0);
@@ -91,9 +91,13 @@ static int32_t mndProcessGetSubEpReq(SMnodeMsg *pMsg) {
     int32_t          assignedSz = taosArrayGetSize(pSub->assigned);
     topicEp.vgs = taosArrayInit(assignedSz, sizeof(SMqSubVgEp));
     for (int32_t j = 0; j < assignedSz; j++) {
-      SMqConsumerEp *pCEp = taosArrayGet(pSub->assigned, i);
+      SMqConsumerEp *pCEp = taosArrayGet(pSub->assigned, j);
       if (pCEp->consumerId == consumerId) {
-        taosArrayPush(pSub->assigned, pCEp);
+        SMqSubVgEp vgEp = {
+          .epSet = pCEp->epSet,
+          .vgId = pCEp->vgId
+        };
+        taosArrayPush(topicEp.vgs, &vgEp);
       }
     }
     if (taosArrayGetSize(topicEp.vgs) != 0) {
@@ -101,7 +105,7 @@ static int32_t mndProcessGetSubEpReq(SMnodeMsg *pMsg) {
     }
   }
   int32_t tlen = tEncodeSMqCMGetSubEpRsp(NULL, &rsp);
-  void   *buf = malloc(tlen);
+  void   *buf = rpcMallocCont(tlen);
   if (buf == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
@@ -161,8 +165,6 @@ static int32_t mndProcessMqTimerMsg(SMnodeMsg *pMsg) {
         pReq->sql = strdup(pTopic->sql);
         pReq->logicalPlan = strdup(pTopic->logicalPlan);
         pReq->physicalPlan = strdup(pTopic->physicalPlan);
-        pReq->qmsgLen = pCEp->qmsgLen;
-        /*memcpy(pReq->qmsg, pCEp->qmsg, pCEp->qmsgLen);*/
         pReq->qmsg = strdup(pCEp->qmsg);
         int32_t tlen = tEncodeSMqSetCVgReq(NULL, pReq);
         void   *reqStr = malloc(tlen);
@@ -192,7 +194,7 @@ static int32_t mndProcessMqTimerMsg(SMnodeMsg *pMsg) {
       if (mndTransPrepare(pMnode, pTrans) != 0) {
         mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
       }
-      mndReleaseTopic(pMnode, pTopic);
+      /*mndReleaseTopic(pMnode, pTopic);*/
       mndTransDrop(pTrans);
     }
     pIter = sdbFetch(pSdb, SDB_SUBSCRIBE, NULL, (void **)&pSub);
@@ -220,6 +222,7 @@ static int mndInitUnassignedVg(SMnode *pMnode, SMqTopicObj *pTopic, SArray *unas
   for (int32_t i = 0; i < sz; i++) {
     SMqConsumerEp CEp;
     CEp.status = 0;
+    CEp.consumerId = -1;
     CEp.lastConsumerHbTs = CEp.lastVgHbTs = -1;
     STaskInfo *pTaskInfo = taosArrayGet(pArray, i);
     tConvertQueryAddrToEpSet(&CEp.epSet, &pTaskInfo->addr);
@@ -227,13 +230,6 @@ static int mndInitUnassignedVg(SMnode *pMnode, SMqTopicObj *pTopic, SArray *unas
      * CEp.epSet.fqdn[2], CEp.epSet.fqdn[3], CEp.epSet.fqdn[4]);*/
     CEp.vgId = pTaskInfo->addr.nodeId;
     CEp.qmsg = strdup(pTaskInfo->msg->msg);
-    CEp.qmsgLen = strlen(CEp.qmsg) + 1;
-    printf("abc:\n%s\n", CEp.qmsg);
-    /*CEp.qmsg = malloc(CEp.qmsgLen);*/
-    /*if (CEp.qmsg == NULL) {*/
-    /*return -1;*/
-    /*}*/
-    /*memcpy(CEp.qmsg, pTaskInfo->msg->msg, pTaskInfo->msg->contentLen);*/
     taosArrayPush(unassignedVg, &CEp);
   }
 
@@ -257,8 +253,7 @@ static int mndBuildMqSetConsumerVgReq(SMnode *pMnode, STrans *pTrans, SMqConsume
     req.sql = pTopic->sql;
     req.logicalPlan = pTopic->logicalPlan;
     req.physicalPlan = pTopic->physicalPlan;
-    req.qmsg = strdup(pCEp->qmsg);
-    req.qmsgLen = strlen(req.qmsg);
+    req.qmsg = pCEp->qmsg;
     int32_t tlen = tEncodeSMqSetCVgReq(NULL, &req);
     void   *buf = malloc(sizeof(SMsgHead) + tlen);
     if (buf == NULL) {
@@ -631,14 +626,14 @@ static int32_t mndProcessSubscribeReq(SMnodeMsg *pMsg) {
     mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
     if (newSub) taosArrayDestroy(newSub);
     mndTransDrop(pTrans);
-    mndReleaseConsumer(pMnode, pConsumer);
+    /*mndReleaseConsumer(pMnode, pConsumer);*/
     return -1;
   }
 
   if (newSub) taosArrayDestroy(newSub);
   mndTransDrop(pTrans);
-  mndReleaseConsumer(pMnode, pConsumer);
-  return 0;
+  /*mndReleaseConsumer(pMnode, pConsumer);*/
+  return TSDB_CODE_MND_ACTION_IN_PROGRESS;
 }
 
 static int32_t mndProcessSubscribeInternalRsp(SMnodeMsg *pRsp) {

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

@@ -237,7 +237,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SMnodeMsg *pMsg, SCMCreateTopicReq
   tstrncpy(topicObj.db, pDb->name, TSDB_DB_FNAME_LEN);
   topicObj.createTime = taosGetTimestampMs();
   topicObj.updateTime = topicObj.createTime;
-  topicObj.uid = mndGenerateUid(pCreate->name, TSDB_TABLE_FNAME_LEN);
+  topicObj.uid = mndGenerateUid(pCreate->name, strlen(pCreate->name));
   topicObj.dbUid = pDb->uid;
   topicObj.version = 1;
   topicObj.sql = strdup(pCreate->sql);

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

@@ -921,7 +921,7 @@ static int32_t mndProcessTransMsg(SMnodeMsg *pMsg) {
 
 void mndTransPullup(SMnode *pMnode) {
   STrans *pTrans = NULL;
-  void   *pIter = NULL;
+  void *  pIter = NULL;
 
   while (1) {
     pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
@@ -930,4 +930,6 @@ void mndTransPullup(SMnode *pMnode) {
     mndTransExecute(pMnode, pTrans);
     sdbRelease(pMnode->pSdb, pTrans);
   }
+
+  sdbWriteFile(pMnode->pSdb);
 }

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

@@ -178,7 +178,7 @@ static int32_t mndVgroupActionUpdate(SSdb *pSdb, SVgObj *pOld, SVgObj *pNew) {
 SVgObj *mndAcquireVgroup(SMnode *pMnode, int32_t vgId) {
   SSdb   *pSdb = pMnode->pSdb;
   SVgObj *pVgroup = sdbAcquire(pSdb, SDB_VGROUP, &vgId);
-  if (pVgroup == NULL) {
+  if (pVgroup == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
     terrno = TSDB_CODE_MND_VGROUP_NOT_EXIST;
   }
   return pVgroup;

source/dnode/mnode/sdb/src/sdb.c

@@ -64,11 +64,7 @@ SSdb *sdbInit(SSdbOpt *pOption) {
 void sdbCleanup(SSdb *pSdb) {
   mDebug("start to cleanup sdb");
 
-  if (pSdb->curVer > pSdb->lastCommitVer) {
-    mDebug("write sdb file for current ver:%" PRId64 " larger than last commit ver:%" PRId64, pSdb->curVer,
-           pSdb->lastCommitVer);
-    sdbWriteFile(pSdb);
-  }
+  sdbWriteFile(pSdb);
 
   if (pSdb->currDir != NULL) {
     tfree(pSdb->currDir);

source/dnode/mnode/sdb/src/sdbFile.c

@@ -221,7 +221,7 @@ PARSE_SDB_DATA_ERROR:
   return code;
 }
 
-int32_t sdbWriteFile(SSdb *pSdb) {
+static int32_t sdbWriteFileImp(SSdb *pSdb) {
   int32_t code = 0;
 
   char tmpfile[PATH_MAX] = {0};
@@ -229,7 +229,8 @@ int32_t sdbWriteFile(SSdb *pSdb) {
   char curfile[PATH_MAX] = {0};
   snprintf(curfile, sizeof(curfile), "%s%ssdb.data", pSdb->currDir, TD_DIRSEP);
 
-  mDebug("start to write file:%s", curfile);
+  mDebug("start to write file:%s, current ver:%" PRId64 ", commit ver:%" PRId64, curfile, pSdb->curVer,
+         pSdb->lastCommitVer);
 
   FileFd fd = taosOpenFileCreateWriteTrunc(tmpfile);
   if (fd <= 0) {
@@ -323,12 +324,20 @@ int32_t sdbWriteFile(SSdb *pSdb) {
   return code;
 }
 
+int32_t sdbWriteFile(SSdb *pSdb) {
+  if (pSdb->curVer == pSdb->lastCommitVer) {
+    return 0;
+  }
+
+  return sdbWriteFileImp(pSdb);
+}
+
 int32_t sdbDeploy(SSdb *pSdb) {
   if (sdbRunDeployFp(pSdb) != 0) {
     return -1;
   }
 
-  if (sdbWriteFile(pSdb) != 0) {
+  if (sdbWriteFileImp(pSdb) != 0) {
     return -1;
   }
 

source/dnode/vnode/inc/tsdb.h

@@ -75,7 +75,6 @@ typedef struct STsdbQueryCond {
 } STsdbQueryCond;
 
 typedef struct {
-  void    *pTable;
   TSKEY    lastKey;
   uint64_t uid;
 } STableKeyInfo;
@@ -141,7 +140,7 @@ bool isTsdbCacheLastRow(tsdbReadHandleT* pTsdbReadHandle);
  */
 int32_t tsdbQuerySTableByTagCond(STsdb* tsdb, uint64_t uid, TSKEY skey, const char* pTagCond, size_t len,
                                  int16_t tagNameRelType, const char* tbnameCond, STableGroupInfo* pGroupInfo,
-                                 SColIndex* pColIndex, int32_t numOfCols, uint64_t reqId);
+                                 SColIndex* pColIndex, int32_t numOfCols, uint64_t reqId, uint64_t taskId);
 /**
  * get num of rows in mem table
  *

source/dnode/vnode/inc/vnode.h

@@ -33,6 +33,7 @@ extern "C" {
 typedef struct SVnode SVnode;
 typedef struct SDnode SDnode;
 typedef int32_t (*PutReqToVQueryQFp)(SDnode *pDnode, struct SRpcMsg *pReq);
+typedef int32_t (*SendReqToDnodeFp)(SDnode *pDnode, struct SEpSet *epSet, struct SRpcMsg *rpcMsg);
 
 typedef struct STqCfg {
   // TODO
@@ -64,6 +65,7 @@ typedef struct {
   const char       *charset;
   uint16_t          nthreads;  // number of commit threads. 0 for no threads and a schedule queue should be given (TODO)
   PutReqToVQueryQFp putReqToVQueryQFp;
+  SendReqToDnodeFp  sendReqToDnodeFp;
 } SVnodeOpt;
 
 typedef struct STqReadHandle {

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

@@ -55,6 +55,7 @@ typedef struct SVnodeMgr {
   // For vnode Mgmt
   SDnode*           pDnode;
   PutReqToVQueryQFp putReqToVQueryQFp;
+  SendReqToDnodeFp  sendReqToDnodeFp;
 } SVnodeMgr;
 
 extern SVnodeMgr vnodeMgr;
@@ -85,6 +86,7 @@ struct SVnode {
 int vnodeScheduleTask(SVnodeTask* task);
 
 int32_t vnodePutReqToVQueryQ(SVnode* pVnode, struct SRpcMsg* pReq);
+void    vnodeSendReqToDnode(SVnode* pVnode, struct SEpSet* epSet, struct SRpcMsg* pReq);
 
 // For Log
 extern int32_t vDebugFlag;

source/dnode/vnode/src/tq/tq.c

@@ -679,6 +679,7 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
   int              rspLen = 0;
 
   STqConsumerHandle* pConsumer = tqHandleGet(pTq->tqMeta, consumerId);
+  ASSERT(pConsumer);
   int                sz = taosArrayGetSize(pConsumer->topics);
 
   for (int i = 0; i < sz; i++) {
@@ -735,23 +736,6 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
         break;
       }
       if (pDataBlock != NULL) {
-        SMqTbData tbData = {
-          .uid = pDataBlock->info.uid,
-          .numOfCols = pDataBlock->info.numOfCols,
-          .numOfRows = pDataBlock->info.rows,
-        };
-        for (int i = 0; i < pDataBlock->info.numOfCols; i++) {
-          SColumnInfoData* pColData = taosArrayGet(pDataBlock->pDataBlock, i);
-          int32_t sz = pColData->info.bytes * pDataBlock->info.rows;
-          SMqColData colData = {
-            .bytes = pColData->info.bytes,
-            .colId = pColData->info.colId,
-            .type = pColData->info.type,
-          };
-          memcpy(colData.data, pColData->pData, colData.bytes * pDataBlock->info.rows);
-          memcpy(&tbData.colData[i], &colData, sz);
-        }
-        /*pDataBlock->info.*/
         taosArrayPush(pRes, pDataBlock);
       } else {
         break;
@@ -773,12 +757,12 @@ int32_t tqProcessConsumeReq(STQ* pTq, SRpcMsg* pMsg) {
     if (pTopic->buffer.lastOffset == -1 || pReq->offset > pTopic->buffer.lastOffset) {
       pTopic->buffer.lastOffset = pReq->offset;
     }
-    // put output into rsp
-    SMqConsumeRsp rsp = {
-      .consumerId = consumerId,
-      .numOfTopics = 1
-    };
   }
+    // put output into rsp
+  SMqConsumeRsp rsp = {
+    .consumerId = consumerId,
+    .numOfTopics = 1
+  };
 
   return 0;
 }
@@ -792,6 +776,8 @@ int32_t tqProcessSetConnReq(STQ* pTq, char* msg) {
   }
   strcpy(pConsumer->cgroup, req.cgroup);
   pConsumer->topics = taosArrayInit(0, sizeof(STqTopicHandle));
+  pConsumer->consumerId = req.newConsumerId;
+  pConsumer->epoch = 0;
 
   STqTopicHandle* pTopic = calloc(sizeof(STqTopicHandle), 1);
   if (pTopic == NULL) {
@@ -802,6 +788,8 @@ int32_t tqProcessSetConnReq(STQ* pTq, char* msg) {
   pTopic->sql = strdup(req.sql);
   pTopic->logicalPlan = strdup(req.logicalPlan);
   pTopic->physicalPlan = strdup(req.physicalPlan);
+  pTopic->committedOffset = -1;
+  pTopic->currentOffset = -1;
 
   pTopic->buffer.firstOffset = -1;
   pTopic->buffer.lastOffset = -1;
@@ -814,6 +802,8 @@ int32_t tqProcessSetConnReq(STQ* pTq, char* msg) {
     pTopic->buffer.output[i].task = qCreateStreamExecTaskInfo(req.qmsg, pReadHandle);
   }
   taosArrayPush(pConsumer->topics, pTopic);
+  tqHandleMovePut(pTq->tqMeta, req.newConsumerId, pConsumer);
+  tqHandleCommit(pTq->tqMeta, req.newConsumerId);
   terrno = TSDB_CODE_SUCCESS;
   return 0;
 }
@@ -832,14 +822,29 @@ STqReadHandle* tqInitSubmitMsgScanner(SMeta* pMeta) {
 
 void tqReadHandleSetMsg(STqReadHandle* pReadHandle, SSubmitMsg* pMsg, int64_t ver) {
   pReadHandle->pMsg = pMsg;
+  pMsg->length = htonl(pMsg->length);
+  pMsg->numOfBlocks = htonl(pMsg->numOfBlocks);
   tInitSubmitMsgIter(pMsg, &pReadHandle->msgIter);
   pReadHandle->ver = ver;
   memset(&pReadHandle->blkIter, 0, sizeof(SSubmitBlkIter));
 }
 
 bool tqNextDataBlock(STqReadHandle* pHandle) {
-  while (tGetSubmitMsgNext(&pHandle->msgIter, &pHandle->pBlock) >= 0) {
-    if (pHandle->tbUid == pHandle->pBlock->uid) return true;
+  while (1) {
+    if (tGetSubmitMsgNext(&pHandle->msgIter, &pHandle->pBlock) < 0) {
+      return false;
+    }
+    if (pHandle->pBlock == NULL) return false;
+
+    pHandle->pBlock->uid = htobe64(pHandle->pBlock->uid);
+    if (pHandle->tbUid == pHandle->pBlock->uid){
+      pHandle->pBlock->tid = htonl(pHandle->pBlock->tid);
+      pHandle->pBlock->sversion = htonl(pHandle->pBlock->sversion);
+      pHandle->pBlock->dataLen = htonl(pHandle->pBlock->dataLen);
+      pHandle->pBlock->schemaLen = htonl(pHandle->pBlock->schemaLen);
+      pHandle->pBlock->numOfRows = htons(pHandle->pBlock->numOfRows);
+     return true;
+    }
   }
   return false;
 }
@@ -855,8 +860,18 @@ int tqRetrieveDataBlockInfo(STqReadHandle* pHandle, SDataBlockInfo* pBlockInfo)
 
 SArray* tqRetrieveDataBlock(STqReadHandle* pHandle) {
   int32_t         sversion = pHandle->pBlock->sversion;
-  SSchemaWrapper* pSchemaWrapper = metaGetTableSchema(pHandle->pMeta, pHandle->pBlock->uid, sversion, true);
-  STSchema*       pTschema = metaGetTbTSchema(pHandle->pMeta, pHandle->pBlock->uid, sversion);
+  //TODO : change sversion
+  STSchema*       pTschema = metaGetTbTSchema(pHandle->pMeta, pHandle->pBlock->uid, 0);
+
+  tb_uid_t quid;
+  STbCfg* pTbCfg = metaGetTbInfoByUid(pHandle->pMeta, pHandle->pBlock->uid);
+  if (pTbCfg->type == META_CHILD_TABLE) {
+    quid = pTbCfg->ctbCfg.suid;
+  } else {
+    quid = pHandle->pBlock->uid;
+  }
+
+  SSchemaWrapper* pSchemaWrapper = metaGetTableSchema(pHandle->pMeta, quid, 0, true);
   SArray*         pArray = taosArrayInit(pSchemaWrapper->nCols, sizeof(SColumnInfoData));
   if (pArray == NULL) {
     return NULL;

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

@@ -81,7 +81,6 @@ enum {
   CHECKINFO_CHOSEN_BOTH = 2    //for update=2(merge case)
 };
 
-
 typedef struct STableCheckInfo {
   uint64_t      tableId;
   TSKEY         lastKey;
@@ -209,31 +208,6 @@ static SArray* getDefaultLoadColumns(STsdbReadHandle* pTsdbReadHandle, bool load
   return pLocalIdList;
 }
 
-static void tsdbMayTakeMemSnapshot(STsdbReadHandle* pTsdbReadHandle, SArray* psTable) {
-//  assert(pTsdbReadHandle != NULL && pTsdbReadHandle->pMemRef != NULL);
-//
-//  STsdbMemTable* pMemRef = pTsdbReadHandle->pMemRef;
-//  if (pTsdbReadHandle->pMemRef->ref++ == 0) {
-//    tsdbTakeMemSnapshot(pTsdbReadHandle->pTsdb, &(pMemRef->snapshot), psTable);
-//  }
-//
-//  taosArrayDestroy(psTable);
-}
-
-static void tsdbMayUnTakeMemSnapshot(STsdbReadHandle* pTsdbReadHandle) {
-//  assert(pTsdbReadHandle != NULL);
-//  STsdbMemTable* pMemRef = pTsdbReadHandle->pMemRef;
-//  if (pMemRef == NULL) { // it has been freed
-//    return;
-//  }
-//
-//  if (--pMemRef->ref == 0) {
-//    tsdbUnTakeMemSnapShot(pTsdbReadHandle->pTsdb, &(pMemRef->snapshot));
-//  }
-//
-//  pTsdbReadHandle->pMemRef = NULL;
-}
-
 //int64_t tsdbGetNumOfRowsInMemTable(tsdbReadHandleT* pHandle) {
 //  STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*) pHandle;
 //
@@ -263,9 +237,9 @@ static void tsdbMayUnTakeMemSnapshot(STsdbReadHandle* pTsdbReadHandle) {
 //  return rows;
 //}
 
-static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, STableGroupInfo* pGroupList, SArray** psTable) {
-  size_t sizeOfGroup = taosArrayGetSize(pGroupList->pGroupList);
-  assert(sizeOfGroup >= 1);
+static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, STableGroupInfo* pGroupList) {
+  size_t numOfGroup = taosArrayGetSize(pGroupList->pGroupList);
+  assert(numOfGroup >= 1);
 
   // allocate buffer in order to load data blocks from file
   SArray* pTableCheckInfo = taosArrayInit(pGroupList->numOfTables, sizeof(STableCheckInfo));
@@ -273,14 +247,8 @@ static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, S
     return NULL;
   }
 
-  SArray* pTable = taosArrayInit(4, sizeof(STable*));
-  if (pTable == NULL) {
-    taosArrayDestroy(pTableCheckInfo);
-    return NULL;
-  }
-
   // todo apply the lastkey of table check to avoid to load header file
-  for (int32_t i = 0; i < sizeOfGroup; ++i) {
+  for (int32_t i = 0; i < numOfGroup; ++i) {
     SArray* group = *(SArray**) taosArrayGet(pGroupList->pGroupList, i);
 
     size_t gsize = taosArrayGetSize(group);
@@ -289,12 +257,7 @@ static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, S
     for (int32_t j = 0; j < gsize; ++j) {
       STableKeyInfo* pKeyInfo = (STableKeyInfo*) taosArrayGet(group, j);
 
-      STableCheckInfo info = { .lastKey = pKeyInfo->lastKey};
-//      assert(info.pTableObj != NULL && (info.pTableObj->type == TSDB_NORMAL_TABLE ||
-//                                        info.pTableObj->type == TSDB_CHILD_TABLE || info.pTableObj->type == TSDB_STREAM_TABLE));
-
-      info.tableId = pKeyInfo->uid;
-
+      STableCheckInfo info = { .lastKey = pKeyInfo->lastKey, .tableId = pKeyInfo->uid};
       if (ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
         if (info.lastKey == INT64_MIN || info.lastKey < pTsdbReadHandle->window.skey) {
           info.lastKey = pTsdbReadHandle->window.skey;
@@ -310,13 +273,8 @@ static SArray* createCheckInfoFromTableGroup(STsdbReadHandle* pTsdbReadHandle, S
     }
   }
 
-//  taosArraySort(pTableCheckInfo, tsdbCheckInfoCompar);
-  size_t gsize = taosArrayGetSize(pTableCheckInfo);
-//  for (int32_t i = 0; i < gsize; ++i) {
-//    STableCheckInfo* pInfo = (STableCheckInfo*) taosArrayGet(pTableCheckInfo, i);
-//  }
-
-  *psTable = pTable;
+  // TODO  group table according to the tag value.
+  taosArraySort(pTableCheckInfo, tsdbCheckInfoCompar);
   return pTableCheckInfo;
 }
 
@@ -484,18 +442,17 @@ tsdbReadHandleT* tsdbQueryTables(STsdb* tsdb, STsdbQueryCond* pCond, STableGroup
     return (tsdbReadHandleT*) pTsdbReadHandle;
   }
 
-  SArray* psTable = NULL;
-
   // todo apply the lastkey of table check to avoid to load header file
-  pTsdbReadHandle->pTableCheckInfo = createCheckInfoFromTableGroup(pTsdbReadHandle, groupList, &psTable);
+  pTsdbReadHandle->pTableCheckInfo = createCheckInfoFromTableGroup(pTsdbReadHandle, groupList);
   if (pTsdbReadHandle->pTableCheckInfo == NULL) {
 //    tsdbCleanupQueryHandle(pTsdbReadHandle);
-    taosArrayDestroy(psTable);
     terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
     return NULL;
   }
 
-  tsdbDebug("%p total numOfTable:%" PRIzu " in query, %s", pTsdbReadHandle, taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo), pTsdbReadHandle->idStr);
+  tsdbDebug("%p total numOfTable:%" PRIzu " in this query, group %"PRIzu" %s", pTsdbReadHandle, taosArrayGetSize(pTsdbReadHandle->pTableCheckInfo),
+      taosArrayGetSize(groupList->pGroupList), pTsdbReadHandle->idStr);
+
   return (tsdbReadHandleT) pTsdbReadHandle;
 }
 
@@ -2640,7 +2597,7 @@ static int32_t getAllTableList(SMeta* pMeta, uint64_t uid, SArray* list) {
       break;
     }
 
-    STableKeyInfo info = {.pTable = NULL, .lastKey = TSKEY_INITIAL_VAL, uid = id};
+    STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, uid = id};
     taosArrayPush(list, &info);
   }
 
@@ -3231,7 +3188,7 @@ STimeWindow updateLastrowForEachGroup(STableGroupInfo *groupList) {
       if (key < lastKey) {
         key = lastKey;
 
-        keyInfo.pTable  = pInfo->pTable;
+//        keyInfo.pTable  = pInfo->pTable;
         keyInfo.lastKey = key;
         pInfo->lastKey  = key;
 
@@ -3245,29 +3202,19 @@ STimeWindow updateLastrowForEachGroup(STableGroupInfo *groupList) {
       }
     }
 
-    // clear current group, unref unused table
-    for (int32_t i = 0; i < numOfTables; ++i) {
-      STableKeyInfo* pInfo = (STableKeyInfo*)taosArrayGet(pGroup, i);
-
-      // keyInfo.pTable may be NULL here.
-      if (pInfo->pTable != keyInfo.pTable) {
-//        tsdbUnRefTable(pInfo->pTable);
-      }
-    }
-
     // more than one table in each group, only one table left for each group
-    if (keyInfo.pTable != NULL) {
-      totalNumOfTable++;
-      if (taosArrayGetSize(pGroup) == 1) {
-        // do nothing
-      } else {
-        taosArrayClear(pGroup);
-        taosArrayPush(pGroup, &keyInfo);
-      }
-    } else {  // mark all the empty groups, and remove it later
-      taosArrayDestroy(pGroup);
-      taosArrayPush(emptyGroup, &j);
-    }
+//    if (keyInfo.pTable != NULL) {
+//      totalNumOfTable++;
+//      if (taosArrayGetSize(pGroup) == 1) {
+//        // do nothing
+//      } else {
+//        taosArrayClear(pGroup);
+//        taosArrayPush(pGroup, &keyInfo);
+//      }
+//    } else {  // mark all the empty groups, and remove it later
+//      taosArrayDestroy(pGroup);
+//      taosArrayPush(emptyGroup, &j);
+//    }
   }
 
   // window does not being updated, so set the original
@@ -3457,11 +3404,13 @@ void filterPrepare(void* expr, void* param) {
   }
 }
 
+#endif
 
 static int32_t tableGroupComparFn(const void *p1, const void *p2, const void *param) {
+#if 0
   STableGroupSupporter* pTableGroupSupp = (STableGroupSupporter*) param;
-  STable* pTable1 = ((STableKeyInfo*) p1)->pTable;
-  STable* pTable2 = ((STableKeyInfo*) p2)->pTable;
+  STable* pTable1 = ((STableKeyInfo*) p1)->uid;
+  STable* pTable2 = ((STableKeyInfo*) p2)->uid;
 
   for (int32_t i = 0; i < pTableGroupSupp->numOfCols; ++i) {
     SColIndex* pColIndex = &pTableGroupSupp->pCols[i];
@@ -3509,10 +3458,9 @@ static int32_t tableGroupComparFn(const void *p1, const void *p2, const void *pa
       return ret;
     }
   }
-
+#endif
   return 0;
 }
-#endif
 
 static int tsdbCheckInfoCompar(const void* key1, const void* key2) {
   if (((STableCheckInfo*)key1)->tableId < ((STableCheckInfo*)key2)->tableId) {
@@ -3528,10 +3476,9 @@ static int tsdbCheckInfoCompar(const void* key1, const void* key2) {
 void createTableGroupImpl(SArray* pGroups, SArray* pTableList, size_t numOfTables, TSKEY skey,
                           STableGroupSupporter* pSupp, __ext_compar_fn_t compareFn) {
   STable* pTable = taosArrayGetP(pTableList, 0);
-
   SArray* g = taosArrayInit(16, sizeof(STableKeyInfo));
 
-  STableKeyInfo info = {.pTable = pTable, .lastKey = skey};
+  STableKeyInfo info = {.lastKey = skey};
   taosArrayPush(g, &info);
 
   for (int32_t i = 1; i < numOfTables; ++i) {
@@ -3542,13 +3489,13 @@ void createTableGroupImpl(SArray* pGroups, SArray* pTableList, size_t numOfTable
     assert(ret == 0 || ret == -1);
 
     if (ret == 0) {
-      STableKeyInfo info1 = {.pTable = *p, .lastKey = skey};
+      STableKeyInfo info1 = {.lastKey = skey};
       taosArrayPush(g, &info1);
     } else {
       taosArrayPush(pGroups, &g);  // current group is ended, start a new group
       g = taosArrayInit(16, sizeof(STableKeyInfo));
 
-      STableKeyInfo info1 = {.pTable = *p, .lastKey = skey};
+      STableKeyInfo info1 = {.lastKey = skey};
       taosArrayPush(g, &info1);
     }
   }
@@ -3581,8 +3528,8 @@ SArray* createTableGroup(SArray* pTableList, SSchemaWrapper* pTagSchema, SColInd
     sup.pTagSchema = pTagSchema->pSchema;
     sup.pCols = pCols;
 
-//    taosqsort(pTableList->pData, size, sizeof(STableKeyInfo), &sup, tableGroupComparFn);
-//    createTableGroupImpl(pTableGroup, pTableList, size, skey, &sup, tableGroupComparFn);
+    taosqsort(pTableList->pData, size, sizeof(STableKeyInfo), &sup, tableGroupComparFn);
+    createTableGroupImpl(pTableGroup, pTableList, size, skey, &sup, tableGroupComparFn);
   }
 
   return pTableGroup;
@@ -3690,16 +3637,16 @@ SArray* createTableGroup(SArray* pTableList, SSchemaWrapper* pTagSchema, SColInd
 
 int32_t tsdbQuerySTableByTagCond(STsdb* tsdb, uint64_t uid, TSKEY skey, const char* pTagCond, size_t len,
                                  int16_t tagNameRelType, const char* tbnameCond, STableGroupInfo* pGroupInfo,
-                                 SColIndex* pColIndex, int32_t numOfCols, uint64_t reqId) {
+                                 SColIndex* pColIndex, int32_t numOfCols, uint64_t reqId, uint64_t taskId) {
   STbCfg* pTbCfg = metaGetTbInfoByUid(tsdb->pMeta, uid);
   if (pTbCfg == NULL) {
-    tsdbError("%p failed to get stable, uid:%"PRIu64", reqId:0x%"PRIx64, tsdb, uid, reqId);
+    tsdbError("%p failed to get stable, uid:%"PRIu64", TID:0x%"PRIx64" QID:0x%"PRIx64, tsdb, uid, taskId, reqId);
     terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
     goto _error;
   }
 
   if (pTbCfg->type != META_SUPER_TABLE) {
-    tsdbError("%p query normal tag not allowed, uid:%" PRIu64 ", reId:0x%"PRIx64, tsdb, uid, reqId);
+    tsdbError("%p query normal tag not allowed, uid:%" PRIu64 ", TID:0x%"PRIx64" QID:0x%"PRIx64, tsdb, uid, taskId, reqId);
     terrno = TSDB_CODE_OPS_NOT_SUPPORT; //basically, this error is caused by invalid sql issued by client
     goto _error;
   }
@@ -3718,8 +3665,8 @@ int32_t tsdbQuerySTableByTagCond(STsdb* tsdb, uint64_t uid, TSKEY skey, const ch
     pGroupInfo->numOfTables = (uint32_t) taosArrayGetSize(res);
     pGroupInfo->pGroupList  = createTableGroup(res, pTagSchema, pColIndex, numOfCols, skey);
 
-    tsdbDebug("%p no table name/tag condition, all tables qualified, numOfTables:%u, group:%zu", tsdb,
-              pGroupInfo->numOfTables, taosArrayGetSize(pGroupInfo->pGroupList));
+    tsdbDebug("%p no table name/tag condition, all tables qualified, numOfTables:%u, group:%zu, TID:0x%"PRIx64" QID:0x%"PRIx64, tsdb,
+              pGroupInfo->numOfTables, taosArrayGetSize(pGroupInfo->pGroupList), taskId, reqId);
 
     taosArrayDestroy(res);
     return ret;
@@ -3892,7 +3839,7 @@ void tsdbCleanupQueryHandle(tsdbReadHandleT queryHandle) {
   tfree(pTsdbReadHandle->statis);
 
   if (!emptyQueryTimewindow(pTsdbReadHandle)) {
-    tsdbMayUnTakeMemSnapshot(pTsdbReadHandle);
+//    tsdbMayUnTakeMemSnapshot(pTsdbReadHandle);
   } else {
     assert(pTsdbReadHandle->pTableCheckInfo == NULL);
   }

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

@@ -26,6 +26,7 @@ int vnodeInit(const SVnodeOpt *pOption) {
 
   vnodeMgr.stop = false;
   vnodeMgr.putReqToVQueryQFp = pOption->putReqToVQueryQFp;
+  vnodeMgr.sendReqToDnodeFp = pOption->sendReqToDnodeFp;
 
   // Start commit handers
   if (pOption->nthreads > 0) {
@@ -96,6 +97,10 @@ int32_t vnodePutReqToVQueryQ(SVnode* pVnode, struct SRpcMsg* pReq) {
   return (*vnodeMgr.putReqToVQueryQFp)(pVnode->pDnode, pReq);
 }
 
+void vnodeSendReqToDnode(SVnode* pVnode, struct SEpSet* epSet, struct SRpcMsg* pReq) {
+  (*vnodeMgr.sendReqToDnodeFp)(pVnode->pDnode, epSet, pReq);
+}
+
 /* ------------------------ STATIC METHODS ------------------------ */
 static void* loop(void* arg) {
   setThreadName("vnode-commit");

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

@@ -21,7 +21,7 @@ static int     vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg);
 
 int vnodeQueryOpen(SVnode *pVnode) {
   return qWorkerInit(NODE_TYPE_VNODE, pVnode->vgId, NULL, (void **)&pVnode->pQuery, pVnode,
-                     (putReqToQueryQFp)vnodePutReqToVQueryQ);
+                     (putReqToQueryQFp)vnodePutReqToVQueryQ, (sendReqToDnodeFp)vnodeSendReqToDnode);
 }
 
 int vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
@@ -43,6 +43,8 @@ int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg) {
   switch (pMsg->msgType) {
     case TDMT_VND_FETCH:
       return qWorkerProcessFetchMsg(pVnode, pVnode->pQuery, pMsg);
+    case TDMT_VND_FETCH_RSP:
+      return qWorkerProcessFetchRsp(pVnode, pVnode->pQuery, pMsg);
     case TDMT_VND_RES_READY:
       return qWorkerProcessReadyMsg(pVnode, pVnode->pQuery, pMsg);
     case TDMT_VND_TASKS_STATUS:

source/libs/executor/inc/executil.h

@@ -38,7 +38,7 @@
 #define GET_RES_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t))
 #define GET_RES_EXT_WINDOW_KEY_LEN(_l) ((_l) + sizeof(uint64_t) + POINTER_BYTES)
 
-#define GET_TASKID(_t)  (((SExecTaskInfo*)(_t))->id.idstr)
+#define GET_TASKID(_t)  (((SExecTaskInfo*)(_t))->id.str)
 
 #define curTimeWindowIndex(_winres)        ((_winres)->curIndex)
 

source/libs/executor/inc/executorimpl.h

@@ -239,8 +239,7 @@ typedef struct STaskIdInfo {
   uint64_t       queryId;    // this is also a request id
   uint64_t       subplanId;
   uint64_t       templateId;
-  uint64_t       taskId;     // this is a subplan id
-  char          *idstr;
+  char          *str;
 } STaskIdInfo;
 
 typedef struct SExecTaskInfo {
@@ -639,7 +638,6 @@ int32_t buildArithmeticExprFromMsg(SExprInfo *pArithExprInfo, void *pQueryMsg);
 bool isTaskKilled(SExecTaskInfo *pTaskInfo);
 int32_t checkForQueryBuf(size_t numOfTables);
 bool checkNeedToCompressQueryCol(SQInfo *pQInfo);
-bool doBuildResCheck(SQInfo* pQInfo);
 void setQueryStatus(STaskRuntimeEnv *pRuntimeEnv, int8_t status);
 
 bool onlyQueryTags(STaskAttr* pQueryAttr);

source/libs/executor/src/executor.c

@@ -18,20 +18,20 @@
 #include "executorimpl.h"
 #include "planner.h"
 
-static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, uint64_t reqId) {
+static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, char* id) {
   ASSERT(pOperator != NULL);
   if (pOperator->operatorType != OP_StreamScan) {
     if (pOperator->numOfDownstream == 0) {
-      qError("failed to find stream scan operator to set the input data block, reqId:0x%" PRIx64, reqId);
+      qError("failed to find stream scan operator to set the input data block, %s" PRIx64, id);
       return TSDB_CODE_QRY_APP_ERROR;
     }
 
     if (pOperator->numOfDownstream > 1) {  // not handle this in join query
-      qError("join not supported for stream block scan, reqId:0x%" PRIx64, reqId);
+      qError("join not supported for stream block scan, %s" PRIx64, id);
       return TSDB_CODE_QRY_APP_ERROR;
     }
 
-    return doSetStreamBlock(pOperator->pDownstream[0], input, reqId);
+    return doSetStreamBlock(pOperator->pDownstream[0], input, id);
   } else {
     SStreamBlockScanInfo* pInfo = pOperator->info;
     tqReadHandleSetMsg(pInfo->readerHandle, input, 0);
@@ -50,7 +50,7 @@ int32_t qSetStreamInput(qTaskInfo_t tinfo, void* input) {
 
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*) tinfo;
 
-  int32_t code = doSetStreamBlock(pTaskInfo->pRoot, input, pTaskInfo->id.queryId);
+  int32_t code = doSetStreamBlock(pTaskInfo->pRoot, input, GET_TASKID(pTaskInfo));
   if (code != TSDB_CODE_SUCCESS) {
     qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo));
   } else {

source/libs/executor/src/executorimpl.c

@@ -5115,7 +5115,7 @@ static void destroySendMsgInfo(SMsgSendInfo* pMsgBody) {
   tfree(pMsgBody);
 }
 
-void processRspMsg(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
+void qProcessFetchRsp(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
   SMsgSendInfo *pSendInfo = (SMsgSendInfo *) pMsg->ahandle;
   assert(pMsg->ahandle != NULL);
 
@@ -5146,7 +5146,7 @@ static SSDataBlock* doLoadRemoteData(void* param, bool* newgroup) {
 
   size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources);
   if (pExchangeInfo->current >= totalSources) {
-    qDebug("%s all %"PRIzu" source(s) are exhausted, total rows:%"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", pTaskInfo->id.idstr, totalSources,
+    qDebug("%s all %"PRIzu" source(s) are exhausted, total rows:%"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", GET_TASKID(pTaskInfo), totalSources,
            pExchangeInfo->totalRows, pExchangeInfo->totalSize, pExchangeInfo->totalElapsed/1000.0);
     return NULL;
   }
@@ -5208,7 +5208,7 @@ static SSDataBlock* doLoadRemoteData(void* param, bool* newgroup) {
         int64_t el = taosGetTimestampUs() - startTs;
         pExchangeInfo->totalElapsed += el;
 
-        qDebug("%s all %"PRIzu" sources are exhausted, total rows: %"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", pTaskInfo->id.idstr, totalSources,
+        qDebug("%s all %"PRIzu" sources are exhausted, total rows: %"PRIu64" bytes:%"PRIu64", elapsed:%.2f ms", GET_TASKID(pTaskInfo), totalSources,
                pExchangeInfo->totalRows, pExchangeInfo->totalSize, pExchangeInfo->totalElapsed/1000.0);
         return NULL;
       } else {
@@ -5296,13 +5296,14 @@ SOperatorInfo* createExchangeOperatorInfo(const SArray* pSources, const SArray*
   pOperator->exec         = doLoadRemoteData;
   pOperator->pTaskInfo    = pTaskInfo;
 
+#if 1
   { // todo refactor
     SRpcInit rpcInit;
     memset(&rpcInit, 0, sizeof(rpcInit));
     rpcInit.localPort = 0;
     rpcInit.label = "EX";
     rpcInit.numOfThreads = 1;
-    rpcInit.cfp = processRspMsg;
+    rpcInit.cfp = qProcessFetchRsp;
     rpcInit.sessions = tsMaxConnections;
     rpcInit.connType = TAOS_CONN_CLIENT;
     rpcInit.user = (char *)"root";
@@ -5316,7 +5317,7 @@ SOperatorInfo* createExchangeOperatorInfo(const SArray* pSources, const SArray*
       return NULL; // todo
     }
   }
-
+#endif
   return pOperator;
 }
 
@@ -5467,6 +5468,9 @@ SOperatorInfo* createStreamScanOperatorInfo(void *streamReadHandle, SArray* pExp
     return NULL;
   }
 
+  // todo dynamic set the value of 4096
+  pInfo->pRes = createOutputBuf_rv(pExprInfo, 4096);
+
   int32_t numOfOutput = (int32_t) taosArrayGetSize(pExprInfo);
   SArray* pColList = taosArrayInit(numOfOutput, sizeof(int32_t));
   for(int32_t i = 0; i < numOfOutput; ++i) {
@@ -7431,7 +7435,7 @@ SOperatorInfo* createTagScanOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SExprInfo
 
   SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
   pOperator->name         = "SeqTableTagScan";
-//  pOperator->operatorType = OP_TagScan;
+  pOperator->operatorType = OP_TagScan;
   pOperator->blockingOptr = false;
   pOperator->status       = OP_IN_EXECUTING;
   pOperator->info         = pInfo;
@@ -7741,11 +7745,10 @@ static SExecTaskInfo* createExecTaskInfo(uint64_t queryId, uint64_t taskId) {
 
   pTaskInfo->cost.created = taosGetTimestampMs();
   pTaskInfo->id.queryId = queryId;
-  pTaskInfo->id.taskId  = taskId;
 
   char* p = calloc(1, 128);
   snprintf(p, 128, "TID:0x%"PRIx64" QID:0x%"PRIx64, taskId, queryId);
-  pTaskInfo->id.idstr = strdup(p);
+  pTaskInfo->id.str = strdup(p);
 
   return pTaskInfo;
 }
@@ -7822,7 +7825,7 @@ static tsdbReadHandleT doCreateDataReadHandle(STableScanPhyNode* pTableScanNode,
 
   if (tableType == TSDB_SUPER_TABLE) {
     code =
-        tsdbQuerySTableByTagCond(readerHandle, uid, window.skey, NULL, 0, 0, NULL, &groupInfo, NULL, 0, queryId);
+        tsdbQuerySTableByTagCond(readerHandle, uid, window.skey, NULL, 0, 0, NULL, &groupInfo, NULL, 0, queryId, taskId);
     if (code != TSDB_CODE_SUCCESS) {
       goto _error;
     }
@@ -7832,7 +7835,7 @@ static tsdbReadHandleT doCreateDataReadHandle(STableScanPhyNode* pTableScanNode,
 
     SArray* pa = taosArrayInit(1, sizeof(STableKeyInfo));
 
-    STableKeyInfo info = {.pTable = NULL, .lastKey = 0, .uid = uid};
+    STableKeyInfo info = {.lastKey = 0, .uid = uid};
     taosArrayPush(pa, &info);
     taosArrayPush(groupInfo.pGroupList, &pa);
   }
@@ -8827,33 +8830,15 @@ void* freeColumnInfo(SColumnInfo* pColumnInfo, int32_t numOfCols) {
 }
 
 void doDestroyTask(SExecTaskInfo *pTaskInfo) {
-  qDebug("%s start to free execTask", GET_TASKID(pTaskInfo));
   doDestroyTableQueryInfo(&pTaskInfo->tableqinfoGroupInfo);
 //  taosArrayDestroy(pTaskInfo->summary.queryProfEvents);
 //  taosHashCleanup(pTaskInfo->summary.operatorProfResults);
 
+  tfree(pTaskInfo->sql);
+  tfree(pTaskInfo->id.str);
   qDebug("%s execTask is freed", GET_TASKID(pTaskInfo));
-  tfree(pTaskInfo);
-}
-
-bool doBuildResCheck(SQInfo* pQInfo) {
-  bool buildRes = false;
 
-  pthread_mutex_lock(&pQInfo->lock);
-
-  pQInfo->dataReady = QUERY_RESULT_READY;
-  buildRes = needBuildResAfterQueryComplete(pQInfo);
-
-  // clear qhandle owner, it must be in the secure area. other thread may run ahead before current, after it is
-  // put into task to be executed.
-  assert(pQInfo->owner == taosGetSelfPthreadId());
-  pQInfo->owner = 0;
-
-  pthread_mutex_unlock(&pQInfo->lock);
-
-  // used in retrieve blocking model.
-  tsem_post(&pQInfo->ready);
-  return buildRes;
+  tfree(pTaskInfo);
 }
 
 static void doSetTagValueToResultBuf(char* output, const char* val, int16_t type, int16_t bytes) {

source/libs/qworker/inc/qworkerInt.h

@@ -137,6 +137,7 @@ typedef struct SQWorkerMgmt {
   SHashObj        *ctxHash;       //key: queryId+taskId, value: SQWTaskCtx
   void            *nodeObj;
   putReqToQueryQFp putToQueueFp;
+  sendReqToDnodeFp sendReqFp;
 } SQWorkerMgmt;
 
 #define QW_FPARAMS_DEF SQWorkerMgmt *mgmt, uint64_t sId, uint64_t qId, uint64_t tId

source/libs/qworker/src/qworker.c

@@ -1313,12 +1313,13 @@ _return:
   QW_RET(code);
 }
 
-int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, void *nodeObj, putReqToQueryQFp fp) {
-  if (NULL == qWorkerMgmt || NULL == nodeObj || NULL == fp) {
+int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qWorkerMgmt, void *nodeObj,
+                    putReqToQueryQFp fp1, sendReqToDnodeFp fp2) {
+  if (NULL == qWorkerMgmt || NULL == nodeObj || NULL == fp1 || NULL == fp2) {
     qError("invalid param to init qworker");
     QW_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }
-  
+
   SQWorkerMgmt *mgmt = calloc(1, sizeof(SQWorkerMgmt));
   if (NULL == mgmt) {
     qError("calloc %d failed", (int32_t)sizeof(SQWorkerMgmt));
@@ -1361,7 +1362,8 @@ int32_t qWorkerInit(int8_t nodeType, int32_t nodeId, SQWorkerCfg *cfg, void **qW
   mgmt->nodeType = nodeType;
   mgmt->nodeId = nodeId;
   mgmt->nodeObj = nodeObj;
-  mgmt->putToQueueFp = fp;
+  mgmt->putToQueueFp = fp1;
+  mgmt->sendReqFp = fp2;
 
   *qWorkerMgmt = mgmt;
 

source/libs/qworker/src/qworkerMsg.c

@@ -421,6 +421,11 @@ int32_t qWorkerProcessFetchMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   return TSDB_CODE_SUCCESS;  
 }
 
+int32_t qWorkerProcessFetchRsp(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
+  qProcessFetchRsp(NULL, pMsg, NULL);
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   if (NULL == node || NULL == qWorkerMgmt || NULL == pMsg) {
     return TSDB_CODE_QRY_INVALID_INPUT;

source/libs/qworker/test/qworkerTests.cpp

@@ -1081,7 +1081,7 @@ TEST(rcTest, shortExecshortDelay) {
   qwtTestStop = false;
   qwtTestQuitThreadNum = 0;
 
-  code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue);
+  code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue, NULL);
   ASSERT_EQ(code, 0);
 
   qwtTestMaxExecTaskUsec = 0;
@@ -1162,7 +1162,7 @@ TEST(rcTest, longExecshortDelay) {
   qwtTestStop = false;
   qwtTestQuitThreadNum = 0;
 
-  code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue);
+  code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue, NULL);
   ASSERT_EQ(code, 0);
 
   qwtTestMaxExecTaskUsec = 1000000;
@@ -1245,7 +1245,7 @@ TEST(rcTest, shortExeclongDelay) {
   qwtTestStop = false;
   qwtTestQuitThreadNum = 0;
 
-  code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue);
+  code = qWorkerInit(NODE_TYPE_VNODE, 1, NULL, &mgmt, mockPointer, qwtPutReqToQueue, NULL);
   ASSERT_EQ(code, 0);
 
   qwtTestMaxExecTaskUsec = 0;

source/libs/transport/inc/transComm.h

@@ -202,6 +202,8 @@ bool transDecompressMsg(char* msg, int32_t len, int32_t* flen);
 void transConnCtxDestroy(STransConnCtx* ctx);
 
 void transFreeMsg(void* msg);
+
+//
 typedef struct SConnBuffer {
   char* buf;
   int   len;
@@ -209,4 +211,9 @@ typedef struct SConnBuffer {
   int   left;
 } SConnBuffer;
 
+int transInitBuffer(SConnBuffer* buf);
+int transClearBuffer(SConnBuffer* buf);
+int transDestroyBuffer(SConnBuffer* buf);
+int transAllocBuffer(SConnBuffer* connBuf, uv_buf_t* uvBuf);
+
 #endif

source/libs/transport/src/trans.c

@@ -30,7 +30,8 @@ void* rpcOpen(const SRpcInit* pInit) {
     tstrncpy(pRpc->label, pInit->label, strlen(pInit->label));
   }
   pRpc->cfp = pInit->cfp;
-  pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
+  // pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
+  pRpc->numOfThreads = pInit->numOfThreads;
   pRpc->connType = pInit->connType;
   pRpc->idleTime = pInit->idleTime;
   pRpc->tcphandle = (*taosInitHandle[pRpc->connType])(0, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc);
@@ -55,7 +56,13 @@ void* rpcMallocCont(int contLen) {
   }
   return start + sizeof(STransMsgHead);
 }
-void  rpcFreeCont(void* cont) { return; }
+void rpcFreeCont(void* cont) {
+  // impl
+  if (cont == NULL) {
+    return;
+  }
+  free((char*)cont - TRANS_MSG_OVERHEAD);
+}
 void* rpcReallocCont(void* ptr, int contLen) { return NULL; }
 
 void rpcSendRedirectRsp(void* pConn, const SEpSet* pEpSet) {}

source/libs/transport/src/transCli.c

@@ -31,6 +31,7 @@ typedef struct SCliConn {
   char         secured;
   uint64_t     expireTime;
   int8_t       notifyCount;  // timers already notify to client
+  int32_t      ref;
 } SCliConn;
 
 typedef struct SCliMsg {
@@ -94,8 +95,10 @@ static void clientHandleExcept(SCliConn* conn);
 // handle req from app
 static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd);
 
-static void clientMsgDestroy(SCliMsg* pMsg);
-static void destroyTransConnCtx(STransConnCtx* ctx);
+static void destroyUserdata(SRpcMsg* userdata);
+
+static void destroyCmsg(SCliMsg* cmsg);
+static void transDestroyConnCtx(STransConnCtx* ctx);
 // thread obj
 static SCliThrdObj* createThrdObj();
 static void         destroyThrdObj(SCliThrdObj* pThrd);
@@ -103,7 +106,8 @@ static void         destroyThrdObj(SCliThrdObj* pThrd);
 static void* clientThread(void* arg);
 
 static void clientHandleResp(SCliConn* conn) {
-  STransConnCtx* pCtx = ((SCliMsg*)conn->data)->ctx;
+  SCliMsg*       pMsg = conn->data;
+  STransConnCtx* pCtx = pMsg->ctx;
   SRpcInfo*      pRpc = pCtx->pTransInst;
 
   STransMsgHead* pHead = (STransMsgHead*)(conn->readBuf.buf);
@@ -112,41 +116,53 @@ static void clientHandleResp(SCliConn* conn) {
 
   SRpcMsg rpcMsg;
   rpcMsg.contLen = transContLenFromMsg(pHead->msgLen);
-  rpcMsg.pCont = transContFromHead(pHead);
+  rpcMsg.pCont = transContFromHead((char*)pHead);
   rpcMsg.code = pHead->code;
   rpcMsg.msgType = pHead->msgType;
   rpcMsg.ahandle = pCtx->ahandle;
 
+  tDebug("conn %p handle resp", conn);
   (pRpc->cfp)(NULL, &rpcMsg, NULL);
   conn->notifyCount += 1;
 
+  // buf's mem alread translated to rpcMsg.pCont
+  transClearBuffer(&conn->readBuf);
+
+  uv_read_start((uv_stream_t*)conn->stream, clientAllocBufferCb, clientReadCb);
+
   SCliThrdObj* pThrd = conn->hostThrd;
-  tfree(conn->data);
   addConnToPool(pThrd->pool, pCtx->ip, pCtx->port, conn);
 
+  destroyCmsg(pMsg);
+  conn->data = NULL;
   // start thread's timer of conn pool if not active
   if (!uv_is_active((uv_handle_t*)pThrd->pTimer) && pRpc->idleTime > 0) {
     uv_timer_start((uv_timer_t*)pThrd->pTimer, clientTimeoutCb, CONN_PERSIST_TIME(pRpc->idleTime) / 2, 0);
   }
-  destroyTransConnCtx(pCtx);
 }
 static void clientHandleExcept(SCliConn* pConn) {
+  if (pConn->data == NULL) {
+    // handle conn except in conn pool
+    clientConnDestroy(pConn, true);
+    return;
+  }
+  tDebug("conn %p start to destroy", pConn);
   SCliMsg* pMsg = pConn->data;
 
-  STransConnCtx* pCtx = pMsg->ctx;
-  SRpcInfo*      pRpc = pCtx->pTransInst;
+  destroyUserdata(&pMsg->msg);
 
-  transFreeMsg((pMsg->msg.pCont));
-  pMsg->msg.pCont = NULL;
+  STransConnCtx* pCtx = pMsg->ctx;
 
   SRpcMsg rpcMsg = {0};
   rpcMsg.ahandle = pCtx->ahandle;
   rpcMsg.code = -1;
   // SRpcInfo* pRpc = pMsg->ctx->pRpc;
-  (pRpc->cfp)(NULL, &rpcMsg, NULL);
-  tfree(pConn->data);
+  (pCtx->pTransInst->cfp)(NULL, &rpcMsg, NULL);
   pConn->notifyCount += 1;
-  destroyTransConnCtx(pCtx);
+
+  destroyCmsg(pMsg);
+  pConn->data = NULL;
+  // transDestroyConnCtx(pCtx);
   clientConnDestroy(pConn, true);
 }
 
@@ -213,14 +229,20 @@ static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) {
   }
   queue* h = QUEUE_HEAD(&plist->conn);
   QUEUE_REMOVE(h);
-  return QUEUE_DATA(h, SCliConn, conn);
+
+  SCliConn* conn = QUEUE_DATA(h, SCliConn, conn);
+  QUEUE_INIT(&conn->conn);
+  return conn;
 }
 static void addConnToPool(void* pool, char* ip, uint32_t port, SCliConn* conn) {
   char key[128] = {0};
+
   tstrncpy(key, ip, strlen(ip));
   tstrncpy(key + strlen(key), (char*)(&port), sizeof(port));
+  tDebug("conn %p added to conn pool, read buf cap: %d", conn, conn->readBuf.cap);
 
   SRpcInfo* pRpc = ((SCliThrdObj*)conn->hostThrd)->pTransInst;
+
   conn->expireTime = taosGetTimestampMs() + CONN_PERSIST_TIME(pRpc->idleTime);
   SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key));
   conn->notifyCount = 0;
@@ -237,40 +259,18 @@ static bool clientReadComplete(SConnBuffer* data) {
     if (msgLen > data->len) {
       data->left = msgLen - data->len;
       return false;
-    } else {
+    } else if (msgLen == data->len) {
+      data->left = 0;
       return true;
     }
   } else {
     return false;
   }
 }
-static void clientAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
-  // impl later
-  static const int CAPACITY = 512;
-
+static void clientAllocBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) {
   SCliConn*    conn = handle->data;
   SConnBuffer* pBuf = &conn->readBuf;
-  if (pBuf->cap == 0) {
-    pBuf->buf = (char*)calloc(1, CAPACITY * sizeof(char));
-    pBuf->len = 0;
-    pBuf->cap = CAPACITY;
-    pBuf->left = -1;
-
-    buf->base = pBuf->buf;
-    buf->len = CAPACITY;
-  } else {
-    if (pBuf->len >= pBuf->cap) {
-      if (pBuf->left == -1) {
-        pBuf->cap *= 2;
-        pBuf->buf = realloc(pBuf->buf, pBuf->cap);
-      } else if (pBuf->len + pBuf->left > pBuf->cap) {
-        pBuf->cap = pBuf->len + pBuf->left;
-        pBuf->buf = realloc(pBuf->buf, pBuf->cap);
-      }
-    }
-    buf->base = pBuf->buf + pBuf->len;
-    buf->len = pBuf->cap - pBuf->len;
-  }
+  transAllocBuffer(pBuf, buf);
 }
 static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf) {
   // impl later
@@ -279,6 +279,7 @@ static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf
   if (nread > 0) {
     pBuf->len += nread;
     if (clientReadComplete(pBuf)) {
+      uv_read_stop((uv_stream_t*)conn->stream);
       tDebug("conn %p read complete", conn);
       clientHandleResp(conn);
     } else {
@@ -288,10 +289,12 @@ static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf
   }
   assert(nread <= 0);
   if (nread == 0) {
-    tError("conn %p closed", conn);
+    // ref http://docs.libuv.org/en/v1.x/stream.html?highlight=uv_read_start#c.uv_read_cb
+    // nread might be 0, which does not indicate an error or EOF. This is equivalent to EAGAIN or EWOULDBLOCK under
+    // read(2).
     return;
   }
-  if (nread < 0) {
+  if (nread < 0 || nread == UV_EOF) {
     tError("conn %p read error: %s", conn, uv_err_name(nread));
     clientHandleExcept(conn);
   }
@@ -300,43 +303,47 @@ static void clientReadCb(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf
 }
 
 static void clientConnDestroy(SCliConn* conn, bool clear) {
-  tDebug("conn %p destroy", conn);
-  if (clear) {
-    uv_close((uv_handle_t*)conn->stream, NULL);
+  //
+  conn->ref--;
+  if (conn->ref == 0) {
+    tDebug("conn %p remove from conn pool", conn);
+    QUEUE_REMOVE(&conn->conn);
+    tDebug("conn %p remove from conn pool successfully", conn);
+    if (clear) {
+      uv_close((uv_handle_t*)conn->stream, clientDestroy);
+    }
   }
-  free(conn->stream);
-  free(conn->readBuf.buf);
-  free(conn->writeReq);
-  free(conn);
 }
 static void clientDestroy(uv_handle_t* handle) {
   SCliConn* conn = handle->data;
-  // QUEUE_REMOVE(&conn->conn);
-  clientConnDestroy(conn, false);
+  // transDestroyBuffer(&conn->readBuf);
+
+  free(conn->stream);
+  free(conn->writeReq);
+  tDebug("conn %p destroy successfully", conn);
+  free(conn);
+
+  // clientConnDestroy(conn, false);
 }
 
 static void clientWriteCb(uv_write_t* req, int status) {
   SCliConn* pConn = req->data;
-
-  SCliMsg* pMsg = pConn->data;
-  transFreeMsg((pMsg->msg.pCont));
-  pMsg->msg.pCont = NULL;
-
   if (status == 0) {
     tDebug("conn %p data already was written out", pConn);
+    SCliMsg* pMsg = pConn->data;
+    if (pMsg == NULL) {
+      destroy
+          // handle
+          return;
+    }
+    destroyUserdata(&pMsg->msg);
   } else {
     tError("conn %p failed to write: %s", pConn, uv_err_name(status));
     clientHandleExcept(pConn);
     return;
   }
   SCliThrdObj* pThrd = pConn->hostThrd;
-  // if (pConn->stream == NULL) {
-  //  pConn->stream = (uv_stream_t*)malloc(sizeof(uv_tcp_t));
-  //  uv_tcp_init(pThrd->loop, (uv_tcp_t*)pConn->stream);
-  //  pConn->stream->data = pConn;
-  //}
-  uv_read_start((uv_stream_t*)pConn->stream, clientAllocReadBufferCb, clientReadCb);
-  // impl later
+  uv_read_start((uv_stream_t*)pConn->stream, clientAllocBufferCb, clientReadCb);
 }
 
 static void clientWrite(SCliConn* pConn) {
@@ -381,14 +388,11 @@ static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
     tDebug("conn %p get from conn pool", conn);
     conn->data = pMsg;
     conn->writeReq->data = conn;
-
-    conn->readBuf.len = 0;
-    memset(conn->readBuf.buf, 0, conn->readBuf.cap);
-    conn->readBuf.left = -1;
+    transDestroyBuffer(&conn->readBuf);
     clientWrite(conn);
   } else {
     SCliConn* conn = calloc(1, sizeof(SCliConn));
-
+    conn->ref++;
     // read/write stream handle
     conn->stream = (uv_stream_t*)malloc(sizeof(uv_tcp_t));
     uv_tcp_init(pThrd->loop, (uv_tcp_t*)(conn->stream));
@@ -397,6 +401,7 @@ static void clientHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
     // write req handle
     conn->writeReq = malloc(sizeof(uv_write_t));
     conn->writeReq->data = conn;
+
     QUEUE_INIT(&conn->conn);
 
     conn->connReq.data = conn;
@@ -459,8 +464,20 @@ void* taosInitClient(uint32_t ip, uint32_t port, char* label, int numOfThreads,
   }
   return cli;
 }
-static void clientMsgDestroy(SCliMsg* pMsg) {
-  // impl later
+
+static void destroyUserdata(SRpcMsg* userdata) {
+  if (userdata->pCont == NULL) {
+    return;
+  }
+  transFreeMsg(userdata->pCont);
+  userdata->pCont = NULL;
+}
+static void destroyCmsg(SCliMsg* pMsg) {
+  if (pMsg == NULL) {
+    return;
+  }
+  transDestroyConnCtx(pMsg->ctx);
+  destroyUserdata(&pMsg->msg);
   free(pMsg);
 }
 static SCliThrdObj* createThrdObj() {
@@ -493,7 +510,7 @@ static void destroyThrdObj(SCliThrdObj* pThrd) {
   free(pThrd);
 }
 
-static void destroyTransConnCtx(STransConnCtx* ctx) {
+static void transDestroyConnCtx(STransConnCtx* ctx) {
   if (ctx != NULL) {
     free(ctx->ip);
   }

source/libs/transport/src/transComm.c

@@ -198,4 +198,51 @@ void transFreeMsg(void* msg) {
   }
   free((char*)msg - sizeof(STransMsgHead));
 }
+
+int transInitBuffer(SConnBuffer* buf) {
+  transClearBuffer(buf);
+  return 0;
+}
+int transClearBuffer(SConnBuffer* buf) {
+  memset(buf, 0, sizeof(*buf));
+  return 0;
+}
+int transAllocBuffer(SConnBuffer* connBuf, uv_buf_t* uvBuf) {
+  /*
+   * formate of data buffer:
+   * |<--------------------------data from socket------------------------------->|
+   * |<------STransMsgHead------->|<-------------------other data--------------->|
+   */
+  static const int CAPACITY = 1024;
+
+  SConnBuffer* p = connBuf;
+  if (p->cap == 0) {
+    p->buf = (char*)calloc(CAPACITY, sizeof(char));
+    p->len = 0;
+    p->cap = CAPACITY;
+    p->left = -1;
+
+    uvBuf->base = p->buf;
+    uvBuf->len = CAPACITY;
+  } else {
+    if (p->len >= p->cap) {
+      if (p->left == -1) {
+        p->cap *= 2;
+        p->buf = realloc(p->buf, p->cap);
+      } else if (p->len + p->left > p->cap) {
+        p->cap = p->len + p->left;
+        p->buf = realloc(p->buf, p->len + p->left);
+      }
+    }
+    uvBuf->base = p->buf + p->len;
+    uvBuf->len = p->cap - p->len;
+  }
+  return 0;
+}
+int transDestroyBuffer(SConnBuffer* buf) {
+  if (buf->cap > 0) {
+    tfree(buf->buf);
+  }
+  transClearBuffer(buf);
+}
 #endif

source/libs/transport/src/transSrv.c

@@ -17,7 +17,7 @@
 
 #include "transComm.h"
 
-typedef struct SConn {
+typedef struct SSrvConn {
   uv_tcp_t*   pTcp;
   uv_write_t* pWriter;
   uv_timer_t* pTimer;
@@ -26,13 +26,14 @@ typedef struct SConn {
   queue       queue;
   int         ref;
   int         persist;  // persist connection or not
-  SConnBuffer connBuf;  // read buf,
+  SConnBuffer readBuf;  // read buf,
   int         inType;
   void*       pTransInst;  // rpc init
   void*       ahandle;     //
   void*       hostThrd;
+  void*       pSrvMsg;
 
-  SRpcMsg sendMsg;
+  // SRpcMsg sendMsg;
   // del later
   char secured;
   int  spi;
@@ -40,7 +41,13 @@ typedef struct SConn {
   char user[TSDB_UNI_LEN];  // user ID for the link
   char secret[TSDB_PASSWORD_LEN];
   char ckey[TSDB_PASSWORD_LEN];  // ciphering key
-} SConn;
+} SSrvConn;
+
+typedef struct SSrvMsg {
+  SSrvConn* pConn;
+  SRpcMsg   msg;
+  queue     q;
+} SSrvMsg;
 
 typedef struct SWorkThrdObj {
   pthread_t       thread;
@@ -48,8 +55,8 @@ typedef struct SWorkThrdObj {
   int             fd;
   uv_loop_t*      loop;
   uv_async_t*     workerAsync;  //
-  queue           conn;
-  pthread_mutex_t connMtx;
+  queue           msg;
+  pthread_mutex_t msgMtx;
   void*           pTransInst;
 } SWorkThrdObj;
 
@@ -68,9 +75,9 @@ typedef struct SServerObj {
 static const char* notify = "a";
 
 // refactor later
-static int transAddAuthPart(SConn* pConn, char* msg, int msgLen);
+static int transAddAuthPart(SSrvConn* pConn, char* msg, int msgLen);
 
-static int uvAuthMsg(SConn* pConn, char* msg, int msgLen);
+static int uvAuthMsg(SSrvConn* pConn, char* msg, int msgLen);
 
 static void uvAllocConnBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
 static void uvAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf);
@@ -82,12 +89,13 @@ static void uvOnAcceptCb(uv_stream_t* stream, int status);
 static void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf);
 static void uvWorkerAsyncCb(uv_async_t* handle);
 
-static void uvPrepareSendData(SConn* conn, uv_buf_t* wb);
-
+static void uvPrepareSendData(SSrvMsg* msg, uv_buf_t* wb);
+static void uvStartSendResp(SSrvMsg* msg);
+static void destroySmsg(SSrvMsg* smsg);
 // check whether already read complete packet
-static bool   readComplete(SConnBuffer* buf);
-static SConn* createConn();
-static void   destroyConn(SConn* conn, bool clear /*clear handle or not*/);
+static bool      readComplete(SConnBuffer* buf);
+static SSrvConn* createConn();
+static void      destroyConn(SSrvConn* conn, bool clear /*clear handle or not*/);
 
 static void uvDestroyConn(uv_handle_t* handle);
 
@@ -105,31 +113,9 @@ void uvAllocReadBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* b
    * |<--------------------------data from socket------------------------------->|
    * |<------STransMsgHead------->|<-------------------other data--------------->|
    */
-  static const int CAPACITY = 1024;
-
-  SConn*       conn = handle->data;
-  SConnBuffer* pBuf = &conn->connBuf;
-  if (pBuf->cap == 0) {
-    pBuf->buf = (char*)calloc(CAPACITY, sizeof(char));
-    pBuf->len = 0;
-    pBuf->cap = CAPACITY;
-    pBuf->left = -1;
-
-    buf->base = pBuf->buf;
-    buf->len = CAPACITY;
-  } else {
-    if (pBuf->len >= pBuf->cap) {
-      if (pBuf->left == -1) {
-        pBuf->cap *= 2;
-        pBuf->buf = realloc(pBuf->buf, pBuf->cap);
-      } else if (pBuf->len + pBuf->left > pBuf->cap) {
-        pBuf->cap = pBuf->len + pBuf->left;
-        pBuf->buf = realloc(pBuf->buf, pBuf->len + pBuf->left);
-      }
-    }
-    buf->base = pBuf->buf + pBuf->len;
-    buf->len = pBuf->cap - pBuf->len;
-  }
+  SSrvConn*    conn = handle->data;
+  SConnBuffer* pBuf = &conn->readBuf;
+  transAllocBuffer(pBuf, buf);
 }
 
 // check data read from socket completely or not
@@ -159,7 +145,7 @@ static bool readComplete(SConnBuffer* data) {
 //  // impl later
 //  STransMsgHead* pHead = (STransMsgHead*)pRecv->msg;
 //  SRpcInfo*      pRpc = (SRpcInfo*)pRecv->shandle;
-//  SConn*         pConn = pRecv->thandle;
+//  SSrvConn*         pConn = pRecv->thandle;
 //  tDump(pRecv->msg, pRecv->msgLen);
 //  terrno = 0;
 //  // SRpcReqContext* pContest;
@@ -167,7 +153,7 @@ static bool readComplete(SConnBuffer* data) {
 //  // do auth and check
 //}
 
-static int uvAuthMsg(SConn* pConn, char* msg, int len) {
+static int uvAuthMsg(SSrvConn* pConn, char* msg, int len) {
   STransMsgHead* pHead = (STransMsgHead*)msg;
 
   int code = 0;
@@ -222,14 +208,14 @@ static int uvAuthMsg(SConn* pConn, char* msg, int len) {
 
 // refers specifically to query or insert timeout
 static void uvHandleActivityTimeout(uv_timer_t* handle) {
-  SConn* conn = handle->data;
+  SSrvConn* conn = handle->data;
   tDebug("%p timeout since no activity", conn);
 }
 
-static void uvHandleReq(SConn* pConn) {
+static void uvHandleReq(SSrvConn* pConn) {
   SRecvInfo    info;
   SRecvInfo*   p = &info;
-  SConnBuffer* pBuf = &pConn->connBuf;
+  SConnBuffer* pBuf = &pConn->readBuf;
   p->msg = pBuf->buf;
   p->msgLen = pBuf->len;
   p->ip = 0;
@@ -255,7 +241,6 @@ static void uvHandleReq(SConn* pConn) {
   pHead->code = htonl(pHead->code);
 
   int32_t dlen = 0;
-  SRpcMsg rpcMsg;
   if (transDecompressMsg(NULL, 0, NULL)) {
     // add compress later
     // pHead = rpcDecompressRpcMsg(pHead);
@@ -264,6 +249,8 @@ static void uvHandleReq(SConn* pConn) {
     // impl later
     //
   }
+
+  SRpcMsg rpcMsg;
   rpcMsg.contLen = transContLenFromMsg(pHead->msgLen);
   rpcMsg.pCont = pHead->content;
   rpcMsg.msgType = pHead->msgType;
@@ -271,6 +258,7 @@ static void uvHandleReq(SConn* pConn) {
   rpcMsg.ahandle = NULL;
   rpcMsg.handle = pConn;
 
+  transClearBuffer(&pConn->readBuf);
   pConn->ref++;
   (*(pRpc->cfp))(pRpc->parent, &rpcMsg, NULL);
   // uv_timer_start(pConn->pTimer, uvHandleActivityTimeout, pRpc->idleTime * 10000, 0);
@@ -280,8 +268,8 @@ static void uvHandleReq(SConn* pConn) {
 
 void uvOnReadCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) {
   // opt
-  SConn*       conn = cli->data;
-  SConnBuffer* pBuf = &conn->connBuf;
+  SSrvConn*    conn = cli->data;
+  SConnBuffer* pBuf = &conn->readBuf;
   if (nread > 0) {
     pBuf->len += nread;
     tDebug("conn %p read summroy, total read: %d, current read: %d", conn, pBuf->len, (int)nread);
@@ -294,11 +282,12 @@ void uvOnReadCb(uv_stream_t* cli, ssize_t nread, const uv_buf_t* buf) {
     return;
   }
   if (nread == 0) {
-    tDebug("conn %p except read", conn);
-    // destroyConn(conn, true);
     return;
   }
-  if (nread != UV_EOF) {
+  if (nread < 0 || nread != UV_EOF) {
+    if (conn->ref > 1) {
+      conn->ref++;  // ref > 1 signed that write is in progress
+    }
     tDebug("conn %p read error: %s", conn, uv_err_name(nread));
     destroyConn(conn, true);
   }
@@ -310,25 +299,23 @@ void uvAllocConnBufferCb(uv_handle_t* handle, size_t suggested_size, uv_buf_t* b
 
 void uvOnTimeoutCb(uv_timer_t* handle) {
   // opt
-  SConn* pConn = handle->data;
+  SSrvConn* pConn = handle->data;
   tDebug("conn %p time out", pConn);
 }
 
 void uvOnWriteCb(uv_write_t* req, int status) {
-  SConn* conn = req->data;
+  SSrvConn* conn = req->data;
 
-  SConnBuffer* buf = &conn->connBuf;
-  buf->len = 0;
-  memset(buf->buf, 0, buf->cap);
-  buf->left = -1;
-
-  SRpcMsg* pMsg = &conn->sendMsg;
-  transFreeMsg(pMsg->pCont);
+  SSrvMsg* smsg = conn->pSrvMsg;
+  destroySmsg(smsg);
+  conn->pSrvMsg = NULL;
 
+  transClearBuffer(&conn->readBuf);
   if (status == 0) {
     tDebug("conn %p data already was written on stream", conn);
   } else {
     tDebug("conn %p failed to write data, %s", conn, uv_err_name(status));
+    //
     destroyConn(conn, true);
   }
   // opt
@@ -341,16 +328,16 @@ static void uvOnPipeWriteCb(uv_write_t* req, int status) {
   }
 }
 
-static void uvPrepareSendData(SConn* conn, uv_buf_t* wb) {
+static void uvPrepareSendData(SSrvMsg* smsg, uv_buf_t* wb) {
   // impl later;
-  tDebug("conn %p prepare to send resp", conn);
-  SRpcMsg* pMsg = &conn->sendMsg;
+  tDebug("conn %p prepare to send resp", smsg->pConn);
+  SRpcMsg* pMsg = &smsg->msg;
   if (pMsg->pCont == 0) {
     pMsg->pCont = (void*)rpcMallocCont(0);
     pMsg->contLen = 0;
   }
   STransMsgHead* pHead = transHeadFromCont(pMsg->pCont);
-  pHead->msgType = conn->inType + 1;
+  pHead->msgType = smsg->pConn->inType + 1;
   // add more info
   char*   msg = (char*)pHead;
   int32_t len = transMsgLenFromCont(pMsg->contLen);
@@ -361,28 +348,53 @@ static void uvPrepareSendData(SConn* conn, uv_buf_t* wb) {
   wb->base = msg;
   wb->len = len;
 }
+static void uvStartSendResp(SSrvMsg* smsg) {
+  // impl
+  uv_buf_t wb;
+  uvPrepareSendData(smsg, &wb);
+
+  SSrvConn* pConn = smsg->pConn;
+  uv_timer_stop(pConn->pTimer);
+
+  pConn->pSrvMsg = smsg;
+  // conn->pWriter->data = smsg;
+  uv_write(pConn->pWriter, (uv_stream_t*)pConn->pTcp, &wb, 1, uvOnWriteCb);
+
+  // SRpcMsg* rpcMsg = smsg->msg;
+
+  return;
+}
+static void destroySmsg(SSrvMsg* smsg) {
+  if (smsg == NULL) {
+    return;
+  }
+  transFreeMsg(smsg->msg.pCont);
+  free(smsg);
+}
 void uvWorkerAsyncCb(uv_async_t* handle) {
   SWorkThrdObj* pThrd = handle->data;
-  SConn*        conn = NULL;
+  SSrvConn*     conn = NULL;
   queue         wq;
   // batch process to avoid to lock/unlock frequently
-  pthread_mutex_lock(&pThrd->connMtx);
-  QUEUE_MOVE(&pThrd->conn, &wq);
-  pthread_mutex_unlock(&pThrd->connMtx);
+  pthread_mutex_lock(&pThrd->msgMtx);
+  QUEUE_MOVE(&pThrd->msg, &wq);
+  pthread_mutex_unlock(&pThrd->msgMtx);
 
   while (!QUEUE_IS_EMPTY(&wq)) {
     queue* head = QUEUE_HEAD(&wq);
     QUEUE_REMOVE(head);
-    SConn* conn = QUEUE_DATA(head, SConn, queue);
-    if (conn == NULL) {
-      tError("except occurred, do nothing");
-      return;
+
+    SSrvMsg* msg = QUEUE_DATA(head, SSrvMsg, q);
+    if (msg == NULL) {
+      tError("except occurred, continue");
+      continue;
     }
-    uv_buf_t wb;
-    uvPrepareSendData(conn, &wb);
-    uv_timer_stop(conn->pTimer);
+    uvStartSendResp(msg);
+    // uv_buf_t wb;
+    // uvPrepareSendData(msg, &wb);
+    // uv_timer_stop(conn->pTimer);
 
-    uv_write(conn->pWriter, (uv_stream_t*)conn->pTcp, &wb, 1, uvOnWriteCb);
+    // uv_write(conn->pWriter, (uv_stream_t*)conn->pTcp, &wb, 1, uvOnWriteCb);
   }
 }
 
@@ -435,7 +447,7 @@ void uvOnConnectionCb(uv_stream_t* q, ssize_t nread, const uv_buf_t* buf) {
   uv_handle_type pending = uv_pipe_pending_type(pipe);
   assert(pending == UV_TCP);
 
-  SConn* pConn = createConn();
+  SSrvConn* pConn = createConn();
 
   pConn->pTransInst = pThrd->pTransInst;
   /* init conn timer*/
@@ -484,8 +496,8 @@ static bool addHandleToWorkloop(void* arg) {
 
   pThrd->pipe->data = pThrd;
 
-  QUEUE_INIT(&pThrd->conn);
-  pthread_mutex_init(&pThrd->connMtx, NULL);
+  QUEUE_INIT(&pThrd->msg);
+  pthread_mutex_init(&pThrd->msgMtx, NULL);
 
   pThrd->workerAsync = malloc(sizeof(uv_async_t));
   uv_async_init(pThrd->loop, pThrd->workerAsync, uvWorkerAsyncCb);
@@ -523,34 +535,43 @@ void* workerThread(void* arg) {
   uv_run(pThrd->loop, UV_RUN_DEFAULT);
 }
 
-static SConn* createConn() {
-  SConn* pConn = (SConn*)calloc(1, sizeof(SConn));
+static SSrvConn* createConn() {
+  SSrvConn* pConn = (SSrvConn*)calloc(1, sizeof(SSrvConn));
+  tDebug("conn %p created", pConn);
   ++pConn->ref;
   return pConn;
 }
 
-static void destroyConn(SConn* conn, bool clear) {
+static void destroyConn(SSrvConn* conn, bool clear) {
   if (conn == NULL) {
     return;
   }
-  if (--conn->ref == 0) {
+  // SRpcMsg* pMsg = &conn->sendMsg;
+  // transFreeMsg(pMsg->pCont);
+  // pMsg->pCont = NULL;
+
+  tDebug("conn %p try to destroy", conn);
+  if (--conn->ref > 0) {
     return;
   }
+  transDestroyBuffer(&conn->readBuf);
+  destroySmsg(conn->pSrvMsg);
+  conn->pSrvMsg = NULL;
+
   if (clear) {
-    uv_close((uv_handle_t*)conn->pTcp, NULL);
+    uv_close((uv_handle_t*)conn->pTcp, uvDestroyConn);
   }
+}
+static void uvDestroyConn(uv_handle_t* handle) {
+  SSrvConn* conn = handle->data;
+  tDebug("conn %p destroy", conn);
   uv_timer_stop(conn->pTimer);
   free(conn->pTimer);
-  free(conn->pTcp);
-  free(conn->connBuf.buf);
+  // free(conn->pTcp);
   free(conn->pWriter);
   free(conn);
 }
-static void uvDestroyConn(uv_handle_t* handle) {
-  SConn* conn = handle->data;
-  destroyConn(conn, false);
-}
-static int transAddAuthPart(SConn* pConn, char* msg, int msgLen) {
+static int transAddAuthPart(SSrvConn* pConn, char* msg, int msgLen) {
   STransMsgHead* pHead = (STransMsgHead*)msg;
 
   if (pConn->spi && pConn->secured == 0) {
@@ -632,6 +653,7 @@ void destroyWorkThrd(SWorkThrdObj* pThrd) {
   pthread_join(pThrd->thread, NULL);
   // free(srv->pipe[i]);
   free(pThrd->loop);
+  pthread_mutex_destroy(&pThrd->msgMtx);
   free(pThrd);
 }
 void taosCloseServer(void* arg) {
@@ -648,17 +670,20 @@ void taosCloseServer(void* arg) {
 }
 
 void rpcSendResponse(const SRpcMsg* pMsg) {
-  SConn*        pConn = pMsg->handle;
+  SSrvConn*     pConn = pMsg->handle;
   SWorkThrdObj* pThrd = pConn->hostThrd;
 
-  // opt later
-  pConn->sendMsg = *pMsg;
-  pthread_mutex_lock(&pThrd->connMtx);
-  QUEUE_PUSH(&pThrd->conn, &pConn->queue);
-  pthread_mutex_unlock(&pThrd->connMtx);
+  SSrvMsg* srvMsg = calloc(1, sizeof(SSrvMsg));
+  srvMsg->pConn = pConn;
+  srvMsg->msg = *pMsg;
+
+  pthread_mutex_lock(&pThrd->msgMtx);
+  QUEUE_PUSH(&pThrd->msg, &srvMsg->q);
+  pthread_mutex_unlock(&pThrd->msgMtx);
+
   tDebug("conn %p start to send resp", pConn);
 
-  uv_async_send(pConn->pWorkerAsync);
+  uv_async_send(pThrd->workerAsync);
 }
 
 #endif

source/libs/transport/test/rclient.c

@@ -64,6 +64,7 @@ static void *sendRequest(void *param) {
     // tsem_wait(&pInfo->rspSem);
     tsem_wait(&pInfo->rspSem);
     tDebug("recv response succefully");
+
     // usleep(100000000);
   }
 

source/libs/transport/test/rserver.c

@@ -165,6 +165,7 @@ int main(int argc, char *argv[]) {
     tError("failed to start RPC server");
     return -1;
   }
+  // sleep(5);
 
   tInfo("RPC server is running, ctrl-c to exit");
 
@@ -172,7 +173,6 @@ int main(int argc, char *argv[]) {
     dataFd = open(dataName, O_APPEND | O_CREAT | O_WRONLY, S_IRWXU | S_IRWXG | S_IRWXO);
     if (dataFd < 0) tInfo("failed to open data file, reason:%s", strerror(errno));
   }
-
   qhandle = taosOpenQueue();
   qset = taosOpenQset();
   taosAddIntoQset(qset, qhandle, NULL);

source/libs/wal/CMakeLists.txt

@@ -11,6 +11,7 @@ target_link_libraries(
     PUBLIC cjson
     PUBLIC os
     PUBLIC util
+    PUBLIC common
 )
 
 if(${BUILD_TEST})

source/libs/wal/src/walWrite.c

@@ -257,7 +257,7 @@ static int walWriteIndex(SWal *pWal, int64_t ver, int64_t offset) {
   return 0;
 }
 
-int64_t walWrite(SWal *pWal, int64_t index, uint8_t msgType, const void *body, int32_t bodyLen) {
+int64_t walWrite(SWal *pWal, int64_t index, tmsg_t msgType, const void *body, int32_t bodyLen) {
   if (pWal == NULL) return -1;
   int code = 0;
 

tests/script/tsim/dnode/basic1.sim

@@ -178,9 +178,9 @@ if $rows != 3 then
 endi
 
 sql select * from st
-#if $rows != 15 then 
-#  return -1
-#endi
+if $rows != 15 then 
+  return -1
+endi
 
 print =============== drop dnode
 sql drop dnode 2;

tests/script/tsim/table/basic1.sim

@@ -139,9 +139,9 @@ endi
 
 print =============== query data frpm st
 sql select * from st
-#if $rows != 21 then 
-#  return -1
-#endi
+if $rows != 21 then 
+  return -1
+endi
 
 system sh/exec.sh -n dnode1 -s stop -x SIGINT
 system sh/exec.sh -n dnode1 -s start
@@ -200,8 +200,8 @@ endi
 
 print =============== query data frpm st
 sql select * from st
-#if $rows != 21 then 
-#  return -1
-#endi
+if $rows != 21 then 
+  return -1
+endi
 
 system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file