From 3d9e27edb83180d7741cd6baa9cbcebfb0f2dd95 Mon Sep 17 00:00:00 2001
From: Shengliang Guan <slguan@taosdata.com>
Date: Mon, 25 Apr 2022 10:53:06 +0800
Subject: [PATCH 01/13] fix: remove un-necessary logs

---
 tools/shell/src/shellCommand.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tools/shell/src/shellCommand.c b/tools/shell/src/shellCommand.c
index c34ee7a22c..fcdbf2d020 100644
--- a/tools/shell/src/shellCommand.c
+++ b/tools/shell/src/shellCommand.c
@@ -395,7 +395,7 @@ void shellClearScreen(int32_t ecmd_pos, int32_t cursor_pos) {
 void shellShowOnScreen(SShellCmd *cmd) {
   struct winsize w;
   if (ioctl(0, TIOCGWINSZ, &w) < 0 || w.ws_col == 0 || w.ws_row == 0) {
-    fprintf(stderr, "No stream device\n");
+    // fprintf(stderr, "No stream device\n");
     w.ws_col = 120;
     w.ws_row = 30;
   }

From ecf00f7f579b1ff45263bfd32b8864ca51fd81e9 Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Mon, 25 Apr 2022 11:41:42 +0800
Subject: [PATCH 02/13] enable hb

---
 example/src/tmq.c            |  1 -
 source/client/src/clientHb.c | 64 ++++++++++++++++++------------------
 2 files changed, 32 insertions(+), 33 deletions(-)

diff --git a/example/src/tmq.c b/example/src/tmq.c
index b2cbf856c0..32390f71c9 100644
--- a/example/src/tmq.c
+++ b/example/src/tmq.c
@@ -60,7 +60,6 @@ int32_t init_env() {
   pRes =
       taos_query(pConn, "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(10)) tags(t1 int)");
   if (taos_errno(pRes) != 0) {
-    assert(0);
     printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes));
     return -1;
   }
diff --git a/source/client/src/clientHb.c b/source/client/src/clientHb.c
index 6159da9cb1..653f9a0634 100644
--- a/source/client/src/clientHb.c
+++ b/source/client/src/clientHb.c
@@ -14,9 +14,9 @@
  */
 
 #include "catalog.h"
-#include "scheduler.h"
 #include "clientInt.h"
 #include "clientLog.h"
+#include "scheduler.h"
 #include "trpc.h"
 
 static SClientHbMgr clientHbMgr = {0};
@@ -110,7 +110,8 @@ static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalo
 static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
   SHbConnInfo *info = taosHashGet(pAppHbMgr->connInfo, &pRsp->connKey, sizeof(SClientHbKey));
   if (NULL == info) {
-    tscWarn("fail to get connInfo, may be dropped, refId:%" PRIx64 ", type:%d", pRsp->connKey.tscRid, pRsp->connKey.connType);
+    tscWarn("fail to get connInfo, may be dropped, refId:%" PRIx64 ", type:%d", pRsp->connKey.tscRid,
+            pRsp->connKey.connType);
     return TSDB_CODE_SUCCESS;
   }
 
@@ -121,7 +122,7 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
     } else {
       updateEpSet_s(&pTscObj->pAppInfo->mgmtEp, &pRsp->query->epSet);
       pTscObj->connId = pRsp->query->connId;
-      
+
       if (pRsp->query->killRid) {
         SRequestObj *pRequest = acquireRequest(pRsp->query->killRid);
         if (NULL == pRequest) {
@@ -131,7 +132,7 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
           releaseRequest(pRsp->query->killRid);
         }
       }
-      
+
       if (pRsp->query->killConnection) {
         taos_close(pTscObj);
       }
@@ -139,7 +140,7 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
       releaseTscObj(pRsp->connKey.tscRid);
     }
   }
-  
+
   int32_t kvNum = pRsp->info ? taosArrayGetSize(pRsp->info) : 0;
 
   tscDebug("hb got %d rsp kv", kvNum);
@@ -236,24 +237,24 @@ static int32_t hbAsyncCallBack(void *param, const SDataBuf *pMsg, int32_t code)
 }
 
 int32_t hbBuildQueryDesc(SQueryHbReqBasic *hbBasic, STscObj *pObj) {
-  int64_t now = taosGetTimestampUs();
+  int64_t    now = taosGetTimestampUs();
   SQueryDesc desc = {0};
-  int32_t code = 0;
+  int32_t    code = 0;
 
-  void   *pIter = taosHashIterate(pObj->pRequests, NULL);
+  void *pIter = taosHashIterate(pObj->pRequests, NULL);
   while (pIter != NULL) {
-    int64_t *rid = pIter;
+    int64_t     *rid = pIter;
     SRequestObj *pRequest = acquireRequest(*rid);
     if (NULL == pRequest) {
       continue;
     }
 
     tstrncpy(desc.sql, pRequest->sqlstr, sizeof(desc.sql));
-    desc.stime    = pRequest->metric.start;
-    desc.queryId  = pRequest->requestId;
+    desc.stime = pRequest->metric.start;
+    desc.queryId = pRequest->requestId;
     desc.useconds = now - pRequest->metric.start;
-    desc.reqRid   = pRequest->self;
-    desc.pid      = hbBasic->pid;
+    desc.reqRid = pRequest->self;
+    desc.pid = hbBasic->pid;
     taosGetFqdn(desc.fqdn);
     desc.subPlanNum = pRequest->body.pDag ? pRequest->body.pDag->numOfSubplans : 0;
 
@@ -271,9 +272,9 @@ int32_t hbBuildQueryDesc(SQueryHbReqBasic *hbBasic, STscObj *pObj) {
       }
     }
 
-    releaseRequest(*rid);    
+    releaseRequest(*rid);
     taosArrayPush(hbBasic->queryDesc, &desc);
-    
+
     pIter = taosHashIterate(pObj->pRequests, pIter);
   }
 
@@ -286,14 +287,14 @@ int32_t hbGetQueryBasicInfo(SClientHbKey *connKey, SClientHbReq *req) {
     tscWarn("tscObj rid %" PRIx64 " not exist", connKey->tscRid);
     return TSDB_CODE_QRY_APP_ERROR;
   }
-  
+
   int32_t numOfQueries = pTscObj->pRequests ? taosHashGetSize(pTscObj->pRequests) : 0;
   if (numOfQueries <= 0) {
     releaseTscObj(connKey->tscRid);
     tscDebug("no queries on connection");
     return TSDB_CODE_QRY_APP_ERROR;
   }
-  
+
   SQueryHbReqBasic *hbBasic = (SQueryHbReqBasic *)taosMemoryCalloc(1, sizeof(SQueryHbReqBasic));
   if (NULL == hbBasic) {
     tscError("calloc %d failed", (int32_t)sizeof(SQueryHbReqBasic));
@@ -308,7 +309,7 @@ int32_t hbGetQueryBasicInfo(SClientHbKey *connKey, SClientHbReq *req) {
     taosMemoryFree(hbBasic);
     return TSDB_CODE_QRY_OUT_OF_MEMORY;
   }
-  
+
   hbBasic->connId = pTscObj->connId;
   hbBasic->pid = taosGetPId();
   taosGetAppName(hbBasic->app, NULL);
@@ -405,7 +406,7 @@ int32_t hbQueryHbReqHandle(SClientHbKey *connKey, void *param, SClientHbReq *req
   }
 
   hbGetQueryBasicInfo(connKey, req);
-  
+
   code = hbGetExpiredDBInfo(connKey, pCatalog, req);
   if (TSDB_CODE_SUCCESS != code) {
     return code;
@@ -471,10 +472,10 @@ SClientHbBatchReq *hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
     pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter);
   }
 
-//  if (code) {
-//    taosArrayDestroyEx(pBatchReq->reqs, hbFreeReq);
-//    taosMemoryFreeClear(pBatchReq);
-//  }
+  //  if (code) {
+  //    taosArrayDestroyEx(pBatchReq->reqs, hbFreeReq);
+  //    taosMemoryFreeClear(pBatchReq);
+  //  }
 
   return pBatchReq;
 }
@@ -630,24 +631,23 @@ void appHbMgrCleanup(void) {
   int sz = taosArrayGetSize(clientHbMgr.appHbMgrs);
   for (int i = 0; i < sz; i++) {
     SAppHbMgr *pTarget = taosArrayGetP(clientHbMgr.appHbMgrs, i);
-    
-    void   *pIter = taosHashIterate(pTarget->activeInfo, NULL);
+
+    void *pIter = taosHashIterate(pTarget->activeInfo, NULL);
     while (pIter != NULL) {
       SClientHbReq *pOneReq = pIter;
       hbFreeReq(pOneReq);
       taosHashCleanup(pOneReq->info);
       pIter = taosHashIterate(pTarget->activeInfo, pIter);
-    }    
+    }
     taosHashCleanup(pTarget->activeInfo);
     pTarget->activeInfo = NULL;
 
-
     pIter = taosHashIterate(pTarget->connInfo, NULL);
     while (pIter != NULL) {
       SHbConnInfo *info = pIter;
       taosMemoryFree(info->param);
       pIter = taosHashIterate(pTarget->connInfo, pIter);
-    }    
+    }
     taosHashCleanup(pTarget->connInfo);
     pTarget->connInfo = NULL;
 
@@ -668,13 +668,13 @@ int hbMgrInit() {
   hbMgrInitHandle();
 
   // init backgroud thread
-  //hbCreateThread();
+  hbCreateThread();
 
   return 0;
 }
 
 void hbMgrCleanUp() {
-  //hbStopThread();
+  // hbStopThread();
 
   // destroy all appHbMgr
   int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 1, 0);
@@ -747,11 +747,11 @@ void hbDeregisterConn(SAppHbMgr *pAppHbMgr, SClientHbKey connKey) {
     taosMemoryFree(info->param);
     taosHashRemove(pAppHbMgr->connInfo, &connKey, sizeof(SClientHbKey));
   }
-  
+
   if (NULL == pReq || NULL == info) {
     return;
   }
-  
+
   atomic_sub_fetch_32(&pAppHbMgr->connKeyCnt, 1);
 }
 

From 8864a9ed592fea8af1008188ea1e140849dce056 Mon Sep 17 00:00:00 2001
From: Shengliang Guan <slguan@taosdata.com>
Date: Mon, 25 Apr 2022 15:39:52 +0800
Subject: [PATCH 03/13] fix(raft): crash while start vnode

---
 source/dnode/mgmt/mgmt_vnode/src/vmInt.c | 47 +++++++++++++++++++++---
 source/dnode/vnode/inc/vnode.h           |  8 ++--
 source/dnode/vnode/src/vnd/vnodeOpen.c   | 11 ++++++
 3 files changed, 55 insertions(+), 11 deletions(-)

diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
index 687a799c57..38f6d7648f 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
@@ -74,12 +74,6 @@ int32_t vmOpenVnode(SVnodesMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
     return -1;
   }
 
-  // sync integration
-  vnodeSyncSetQ(pImpl, NULL);
-  vnodeSyncSetRpc(pImpl, NULL);
-  int32_t ret = vnodeSyncStart(pImpl);
-  assert(ret == 0);
-
   taosWLockLatch(&pMgmt->latch);
   int32_t code = taosHashPut(pMgmt->hash, &pVnode->vgId, sizeof(int32_t), &pVnode, sizeof(SVnodeObj *));
   taosWUnLockLatch(&pMgmt->latch);
@@ -153,6 +147,7 @@ static void *vmOpenVnodeFunc(void *param) {
       pThread->failed++;
     } else {
       vmOpenVnode(pMgmt, pCfg, pImpl);
+      vnodeStart(pImpl);
       dDebug("vgId:%d, is opened by thread:%d", pCfg->vgId, pThread->threadIndex);
       pThread->opened++;
     }
@@ -364,10 +359,50 @@ static int32_t vmRequire(SMgmtWrapper *pWrapper, bool *required) {
   return 0;
 }
 
+static int32_t vmStart(SMgmtWrapper *pWrapper) {
+  dDebug("vnode-mgmt start to run");
+  SVnodesMgmt *pMgmt = pWrapper->pMgmt;
+
+  taosRLockLatch(&pMgmt->latch);
+
+  void *pIter = taosHashIterate(pMgmt->hash, NULL);
+  while (pIter) {
+    SVnodeObj **ppVnode = pIter;
+    if (ppVnode == NULL || *ppVnode == NULL) continue;
+
+    SVnodeObj *pVnode = *ppVnode;
+    vnodeStart(pVnode->pImpl);
+    pIter = taosHashIterate(pMgmt->hash, pIter);
+  }
+
+  taosRUnLockLatch(&pMgmt->latch);
+  return 0;
+}
+
+static void vmStop(SMgmtWrapper *pWrapper) {
+  dDebug("vnode-mgmt start to stop");
+  SVnodesMgmt *pMgmt = pWrapper->pMgmt;
+  taosRLockLatch(&pMgmt->latch);
+
+  void *pIter = taosHashIterate(pMgmt->hash, NULL);
+  while (pIter) {
+    SVnodeObj **ppVnode = pIter;
+    if (ppVnode == NULL || *ppVnode == NULL) continue;
+
+    SVnodeObj *pVnode = *ppVnode;
+    vnodeStop(pVnode->pImpl);
+    pIter = taosHashIterate(pMgmt->hash, pIter);
+  }
+
+  taosRUnLockLatch(&pMgmt->latch);
+}
+
 void vmSetMgmtFp(SMgmtWrapper *pWrapper) {
   SMgmtFp mgmtFp = {0};
   mgmtFp.openFp = vmInit;
   mgmtFp.closeFp = vmCleanup;
+  mgmtFp.startFp = vmStart;
+  mgmtFp.stopFp = vmStop;
   mgmtFp.requiredFp = vmRequire;
 
   vmInitMsgHandle(pWrapper);
diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h
index f4975c183e..9eab4d3376 100644
--- a/source/dnode/vnode/inc/vnode.h
+++ b/source/dnode/vnode/inc/vnode.h
@@ -61,6 +61,9 @@ int32_t vnodeSync(SVnode *pVnode);
 int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad);
 int     vnodeValidateTableHash(SVnodeCfg *pVnodeOptions, char *tableFName);
 
+int32_t vnodeStart(SVnode *pVnode);
+void    vnodeStop(SVnode *pVnode);
+
 int64_t vnodeGetSyncHandle(SVnode *pVnode);
 void    vnodeGetSnapshot(SVnode *pVnode, SSnapshot *pSnapshot);
 
@@ -171,11 +174,6 @@ typedef struct {
   uint64_t uid;
 } STableKeyInfo;
 
-// sync integration
-void    vnodeSyncSetQ(SVnode *pVnode, void *qHandle);
-void    vnodeSyncSetRpc(SVnode *pVnode, void *rpcHandle);
-int32_t vnodeSyncStart(SVnode *pVnode);
-
 #ifdef __cplusplus
 }
 #endif
diff --git a/source/dnode/vnode/src/vnd/vnodeOpen.c b/source/dnode/vnode/src/vnd/vnodeOpen.c
index 2b10afa5ed..11136bc7ba 100644
--- a/source/dnode/vnode/src/vnd/vnodeOpen.c
+++ b/source/dnode/vnode/src/vnd/vnodeOpen.c
@@ -14,6 +14,7 @@
  */
 
 #include "vnodeInt.h"
+#include "vnodeSync.h"
 
 int vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs) {
   SVnodeInfo info = {0};
@@ -171,6 +172,16 @@ void vnodeClose(SVnode *pVnode) {
   }
 }
 
+// start the sync timer after the queue is ready
+int32_t vnodeStart(SVnode *pVnode) {
+  vnodeSyncSetQ(pVnode, NULL);
+  vnodeSyncSetRpc(pVnode, NULL);
+  vnodeSyncStart(pVnode);
+  return 0;
+}
+
+void vnodeStop(SVnode *pVnode) {}
+
 int64_t vnodeGetSyncHandle(SVnode *pVnode) { return pVnode->sync; }
 
 void vnodeGetSnapshot(SVnode *pVnode, SSnapshot *pSnapshot) { pSnapshot->lastApplyIndex = pVnode->state.committed; }
\ No newline at end of file

From 609b21647bb771bbb4487c18fc6f68103de33e9e Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Mon, 25 Apr 2022 15:40:25 +0800
Subject: [PATCH 04/13] fix: msg with schema

---
 example/src/tmq.c                          |   7 +-
 include/common/tmsg.h                      | 110 ++++++---------------
 source/client/inc/clientInt.h              |   4 +
 source/client/src/clientImpl.c             |  29 +++---
 source/client/src/tmq.c                    |   6 +-
 source/dnode/mnode/impl/src/mndScheduler.c |  91 ++++++++---------
 source/dnode/mnode/impl/src/mndTopic.c     |   7 +-
 source/dnode/vnode/src/tq/tq.c             |  30 ++++--
 source/dnode/vnode/src/tq/tqRead.c         |  66 +++++++++----
 9 files changed, 178 insertions(+), 172 deletions(-)

diff --git a/example/src/tmq.c b/example/src/tmq.c
index 32390f71c9..56f210081b 100644
--- a/example/src/tmq.c
+++ b/example/src/tmq.c
@@ -103,8 +103,8 @@ int32_t create_topic() {
   }
   taos_free_result(pRes);
 
-  /*pRes = taos_query(pConn, "create topic topic_ctb_column as abc1");*/
-  pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from ct1");
+  pRes = taos_query(pConn, "create topic topic_ctb_column as abc1");
+  /*pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from ct1");*/
   if (taos_errno(pRes) != 0) {
     printf("failed to create topic topic_ctb_column, reason:%s\n", taos_errstr(pRes));
     return -1;
@@ -162,9 +162,10 @@ tmq_t* build_consumer() {
   tmq_conf_set(conf, "group.id", "tg2");
   tmq_conf_set(conf, "td.connect.user", "root");
   tmq_conf_set(conf, "td.connect.pass", "taosdata");
-  tmq_conf_set(conf, "td.connect.db", "abc1");
+  /*tmq_conf_set(conf, "td.connect.db", "abc1");*/
   tmq_conf_set_offset_commit_cb(conf, tmq_commit_cb_print);
   tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
+  assert(tmq);
   return tmq;
 }
 
diff --git a/include/common/tmsg.h b/include/common/tmsg.h
index 5a2dd502c1..a07101f404 100644
--- a/include/common/tmsg.h
+++ b/include/common/tmsg.h
@@ -2052,80 +2052,6 @@ static FORCE_INLINE void* tDecodeSMqSetCVgReq(void* buf, SMqSetCVgReq* pReq) {
   return buf;
 }
 
-typedef struct {
-  int64_t leftForVer;
-  int32_t vgId;
-  int32_t epoch;
-  int64_t consumerId;
-  char    topicName[TSDB_TOPIC_FNAME_LEN];
-} SMqCancelConnReq;
-
-static FORCE_INLINE int32_t tEncodeSMqCancelConnReq(void** buf, const SMqCancelConnReq* pReq) {
-  int32_t tlen = 0;
-  tlen += taosEncodeFixedI64(buf, pReq->leftForVer);
-  tlen += taosEncodeFixedI32(buf, pReq->vgId);
-  tlen += taosEncodeFixedI32(buf, pReq->epoch);
-  tlen += taosEncodeFixedI64(buf, pReq->consumerId);
-  tlen += taosEncodeString(buf, pReq->topicName);
-  return tlen;
-}
-
-static FORCE_INLINE void* tDecodeSMqCancelConnReq(void* buf, SMqCancelConnReq* pReq) {
-  buf = taosDecodeFixedI64(buf, &pReq->leftForVer);
-  buf = taosDecodeFixedI32(buf, &pReq->vgId);
-  buf = taosDecodeFixedI32(buf, &pReq->epoch);
-  buf = taosDecodeFixedI64(buf, &pReq->consumerId);
-  buf = taosDecodeStringTo(buf, pReq->topicName);
-  return buf;
-}
-
-typedef struct {
-  int8_t reserved;
-} SMqCancelConnRsp;
-
-typedef struct {
-  int64_t leftForVer;
-  int32_t vgId;
-  int64_t oldConsumerId;
-  int64_t newConsumerId;
-  char*   topic;
-} SMqMVRebReq;
-
-static FORCE_INLINE int32_t tEncodeSMqMVRebReq(void** buf, const SMqMVRebReq* pReq) {
-  int32_t tlen = 0;
-  tlen += taosEncodeFixedI64(buf, pReq->leftForVer);
-  tlen += taosEncodeFixedI32(buf, pReq->vgId);
-  tlen += taosEncodeFixedI64(buf, pReq->oldConsumerId);
-  tlen += taosEncodeFixedI64(buf, pReq->newConsumerId);
-  tlen += taosEncodeString(buf, pReq->topic);
-  return tlen;
-}
-
-static FORCE_INLINE void* tDecodeSMqMVRebReq(void* buf, SMqMVRebReq* pReq) {
-  buf = taosDecodeFixedI64(buf, &pReq->leftForVer);
-  buf = taosDecodeFixedI32(buf, &pReq->vgId);
-  buf = taosDecodeFixedI64(buf, &pReq->oldConsumerId);
-  buf = taosDecodeFixedI64(buf, &pReq->newConsumerId);
-  buf = taosDecodeString(buf, &pReq->topic);
-  return buf;
-}
-
-typedef struct {
-  SMsgHead header;
-  int32_t  vgId;
-  int64_t  consumerId;
-  char     topicName[TSDB_TOPIC_FNAME_LEN];
-  char     cgroup[TSDB_CGROUP_LEN];
-} SMqSetCVgRsp;
-
-typedef struct {
-  SMsgHead header;
-  int32_t  vgId;
-  int64_t  consumerId;
-  char     topicName[TSDB_TOPIC_FNAME_LEN];
-  char     cgroup[TSDB_CGROUP_LEN];
-} SMqMVRebRsp;
-
 typedef struct {
   int32_t vgId;
   int64_t offset;
@@ -2152,6 +2078,24 @@ typedef struct {
   SSchema* pSchema;
 } SSchemaWrapper;
 
+static FORCE_INLINE SSchemaWrapper* tCloneSSchemaWrapper(const SSchemaWrapper* pSchemaWrapper) {
+  SSchemaWrapper* pSW = (SSchemaWrapper*)taosMemoryMalloc(sizeof(SSchemaWrapper));
+  if (pSW == NULL) return pSW;
+  pSW->nCols = pSchemaWrapper->nCols;
+  pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema));
+  if (pSW->pSchema == NULL) {
+    taosMemoryFree(pSW);
+    return NULL;
+  }
+  memcpy(pSW->pSchema, pSchemaWrapper->pSchema, pSW->nCols * sizeof(SSchema));
+  return pSW;
+}
+
+static FORCE_INLINE void tDeleteSSchemaWrapper(SSchemaWrapper* pSchemaWrapper) {
+  taosMemoryFree(pSchemaWrapper->pSchema);
+  taosMemoryFree(pSchemaWrapper);
+}
+
 static FORCE_INLINE int32_t taosEncodeSSchema(void** buf, const SSchema* pSchema) {
   int32_t tlen = 0;
   tlen += taosEncodeFixedI8(buf, pSchema->type);
@@ -2162,13 +2106,13 @@ static FORCE_INLINE int32_t taosEncodeSSchema(void** buf, const SSchema* pSchema
   return tlen;
 }
 
-static FORCE_INLINE void* taosDecodeSSchema(void* buf, SSchema* pSchema) {
+static FORCE_INLINE void* taosDecodeSSchema(const void* buf, SSchema* pSchema) {
   buf = taosDecodeFixedI8(buf, &pSchema->type);
   buf = taosDecodeFixedI8(buf, &pSchema->flags);
   buf = taosDecodeFixedI32(buf, &pSchema->bytes);
   buf = taosDecodeFixedI16(buf, &pSchema->colId);
   buf = taosDecodeStringTo(buf, pSchema->name);
-  return buf;
+  return (void*)buf;
 }
 
 static FORCE_INLINE int32_t tEncodeSSchema(SCoder* pEncoder, const SSchema* pSchema) {
@@ -2198,7 +2142,7 @@ static FORCE_INLINE int32_t taosEncodeSSchemaWrapper(void** buf, const SSchemaWr
   return tlen;
 }
 
-static FORCE_INLINE void* taosDecodeSSchemaWrapper(void* buf, SSchemaWrapper* pSW) {
+static FORCE_INLINE void* taosDecodeSSchemaWrapper(const void* buf, SSchemaWrapper* pSW) {
   buf = taosDecodeFixedU32(buf, &pSW->nCols);
   pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema));
   if (pSW->pSchema == NULL) {
@@ -2208,7 +2152,7 @@ static FORCE_INLINE void* taosDecodeSSchemaWrapper(void* buf, SSchemaWrapper* pS
   for (int32_t i = 0; i < pSW->nCols; i++) {
     buf = taosDecodeSSchema(buf, &pSW->pSchema[i]);
   }
-  return buf;
+  return (void*)buf;
 }
 
 static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SCoder* pEncoder, const SSchemaWrapper* pSW) {
@@ -2615,6 +2559,10 @@ static FORCE_INLINE int32_t tEncodeSMqDataBlkRsp(void** buf, const SMqDataBlkRsp
       void*   data = taosArrayGetP(pRsp->blockData, i);
       tlen += taosEncodeFixedI32(buf, bLen);
       tlen += taosEncodeBinary(buf, data, bLen);
+      if (pRsp->withSchema) {
+        SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(pRsp->blockSchema, i);
+        tlen += taosEncodeSSchemaWrapper(buf, pSW);
+      }
     }
   }
   return tlen;
@@ -2627,6 +2575,7 @@ static FORCE_INLINE void* tDecodeSMqDataBlkRsp(const void* buf, SMqDataBlkRsp* p
   buf = taosDecodeFixedI32(buf, &pRsp->blockNum);
   pRsp->blockData = taosArrayInit(pRsp->blockNum, sizeof(void*));
   pRsp->blockDataLen = taosArrayInit(pRsp->blockNum, sizeof(void*));
+  pRsp->blockSchema = taosArrayInit(pRsp->blockNum, sizeof(void*));
   if (pRsp->blockNum != 0) {
     buf = taosDecodeFixedI8(buf, &pRsp->withTbName);
     buf = taosDecodeFixedI8(buf, &pRsp->withSchema);
@@ -2639,6 +2588,11 @@ static FORCE_INLINE void* tDecodeSMqDataBlkRsp(const void* buf, SMqDataBlkRsp* p
       buf = taosDecodeBinary(buf, &data, bLen);
       taosArrayPush(pRsp->blockDataLen, &bLen);
       taosArrayPush(pRsp->blockData, &data);
+      if (pRsp->withSchema) {
+        SSchemaWrapper* pSW = (SSchemaWrapper*)taosMemoryMalloc(sizeof(SSchemaWrapper));
+        buf = taosDecodeSSchemaWrapper(buf, pSW);
+        taosArrayPush(pRsp->blockSchema, &pSW);
+      }
     }
   }
   return (void*)buf;
diff --git a/source/client/inc/clientInt.h b/source/client/inc/clientInt.h
index 814caf330a..c0237a184a 100644
--- a/source/client/inc/clientInt.h
+++ b/source/client/inc/clientInt.h
@@ -231,6 +231,10 @@ static FORCE_INLINE SReqResultInfo* tmqGetNextResInfo(TAOS_RES* res, bool conver
   msg->resIter++;
   if (msg->resIter < msg->rsp.blockNum) {
     SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)taosArrayGetP(msg->rsp.blockData, msg->resIter);
+    if (msg->rsp.withSchema) {
+      SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(msg->rsp.blockSchema, msg->resIter);
+      setResSchemaInfo(&msg->resInfo, pSW->pSchema, pSW->nCols);
+    }
     setQueryResultFromRsp(&msg->resInfo, pRetrieve, convertUcs4);
     return &msg->resInfo;
   }
diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c
index 2ae9a11803..a40edbb65a 100644
--- a/source/client/src/clientImpl.c
+++ b/source/client/src/clientImpl.c
@@ -226,17 +226,15 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQuery* pQuery) {
 
 int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList) {
   pRequest->type = pQuery->msgType;
-  SPlanContext cxt = {
-    .queryId = pRequest->requestId,
-    .acctId = pRequest->pTscObj->acctId,
-    .mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp),
-    .pAstRoot = pQuery->pRoot,
-    .showRewrite = pQuery->showRewrite,
-    .pTransporter = pRequest->pTscObj->pAppInfo->pTransporter,
-    .pMsg = pRequest->msgBuf,
-    .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE
-  };
-  int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &cxt.pCatalog);
+  SPlanContext cxt = {.queryId = pRequest->requestId,
+                      .acctId = pRequest->pTscObj->acctId,
+                      .mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp),
+                      .pAstRoot = pQuery->pRoot,
+                      .showRewrite = pQuery->showRewrite,
+                      .pTransporter = pRequest->pTscObj->pAppInfo->pTransporter,
+                      .pMsg = pRequest->msgBuf,
+                      .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE};
+  int32_t      code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &cxt.pCatalog);
   if (TSDB_CODE_SUCCESS == code) {
     code = qCreateQueryPlan(&cxt, pPlan, pNodeList);
   }
@@ -247,6 +245,7 @@ void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t
   ASSERT(pSchema != NULL && numOfCols > 0);
 
   pResInfo->numOfCols = numOfCols;
+  // TODO handle memory leak
   pResInfo->fields = taosMemoryCalloc(numOfCols, sizeof(TAOS_FIELD));
   pResInfo->userFields = taosMemoryCalloc(numOfCols, sizeof(TAOS_FIELD));
 
@@ -282,7 +281,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
 
   SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf};
   int32_t      code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr,
-                                  pRequest->metric.start, &res);
+                                       pRequest->metric.start, &res);
   if (code != TSDB_CODE_SUCCESS) {
     if (pRequest->body.queryJob != 0) {
       schedulerFreeJob(pRequest->body.queryJob);
@@ -840,12 +839,12 @@ int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32
     return code;
   }
 
-  char* p = (char*) pResultInfo->pData;
+  char* p = (char*)pResultInfo->pData;
 
-  int32_t dataLen = *(int32_t*) p;
+  int32_t dataLen = *(int32_t*)p;
   p += sizeof(int32_t);
 
-  uint64_t groupId = *(uint64_t*) p;
+  uint64_t groupId = *(uint64_t*)p;
   p += sizeof(uint64_t);
 
   int32_t* colLength = (int32_t*)p;
diff --git a/source/client/src/tmq.c b/source/client/src/tmq.c
index d576f28cb4..988406c84d 100644
--- a/source/client/src/tmq.c
+++ b/source/client/src/tmq.c
@@ -376,7 +376,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
 
   ASSERT(user);
   ASSERT(pass);
-  ASSERT(conf->db);
+  /*ASSERT(conf->db);*/
   ASSERT(conf->groupId[0]);
 
   pTmq->pTscObj = taos_connect_internal(conf->ip, user, pass, NULL, conf->db, conf->port, CONN_TYPE__TMQ);
@@ -1118,7 +1118,9 @@ SMqRspObj* tmqBuildRspFromWrapper(SMqPollRspWrapper* pWrapper) {
 
   pRspObj->resInfo.totalRows = 0;
   pRspObj->resInfo.precision = TSDB_TIME_PRECISION_MILLI;
-  setResSchemaInfo(&pRspObj->resInfo, pWrapper->topicHandle->schema.pSchema, pWrapper->topicHandle->schema.nCols);
+  if (!pWrapper->msg.withSchema) {
+    setResSchemaInfo(&pRspObj->resInfo, pWrapper->topicHandle->schema.pSchema, pWrapper->topicHandle->schema.nCols);
+  }
 
   taosFreeQitem(pWrapper);
   return pRspObj;
diff --git a/source/dnode/mnode/impl/src/mndScheduler.c b/source/dnode/mnode/impl/src/mndScheduler.c
index 73583058f1..3dff65866c 100644
--- a/source/dnode/mnode/impl/src/mndScheduler.c
+++ b/source/dnode/mnode/impl/src/mndScheduler.c
@@ -476,33 +476,37 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
 int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscribeObj* pSub) {
   SSdb*       pSdb = pMnode->pSdb;
   SVgObj*     pVgroup = NULL;
-  SQueryPlan* pPlan = qStringToQueryPlan(pTopic->physicalPlan);
-  if (pPlan == NULL) {
-    terrno = TSDB_CODE_QRY_INVALID_INPUT;
-    return -1;
+  SQueryPlan* pPlan = NULL;
+  SSubplan*   plan = NULL;
+  if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {
+    pPlan = qStringToQueryPlan(pTopic->physicalPlan);
+    if (pPlan == NULL) {
+      terrno = TSDB_CODE_QRY_INVALID_INPUT;
+      return -1;
+    }
+
+    ASSERT(pSub->vgNum == -1);
+
+    pSub->vgNum = 0;
+
+    int32_t levelNum = LIST_LENGTH(pPlan->pSubplans);
+    if (levelNum != 1) {
+      qDestroyQueryPlan(pPlan);
+      terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC;
+      return -1;
+    }
+
+    SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0);
+
+    int32_t opNum = LIST_LENGTH(inner->pNodeList);
+    if (opNum != 1) {
+      qDestroyQueryPlan(pPlan);
+      terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC;
+      return -1;
+    }
+    plan = nodesListGetNode(inner->pNodeList, 0);
   }
 
-  ASSERT(pSub->vgNum == -1);
-
-  pSub->vgNum = 0;
-
-  int32_t levelNum = LIST_LENGTH(pPlan->pSubplans);
-  if (levelNum != 1) {
-    qDestroyQueryPlan(pPlan);
-    terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC;
-    return -1;
-  }
-
-  SNodeListNode* inner = nodesListGetNode(pPlan->pSubplans, 0);
-
-  int32_t opNum = LIST_LENGTH(inner->pNodeList);
-  if (opNum != 1) {
-    qDestroyQueryPlan(pPlan);
-    terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC;
-    return -1;
-  }
-  SSubplan* plan = nodesListGetNode(inner->pNodeList, 0);
-
   int64_t             unexistKey = -1;
   SMqConsumerEpInSub* pEpInSub = taosHashGet(pSub->consumerHash, &unexistKey, sizeof(int64_t));
   ASSERT(pEpInSub);
@@ -519,38 +523,35 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
     }
 
     pSub->vgNum++;
-    plan->execNode.nodeId = pVgroup->vgId;
-    plan->execNode.epSet = mndGetVgroupEpset(pMnode, pVgroup);
 
     SMqVgEp* pVgEp = taosMemoryMalloc(sizeof(SMqVgEp));
-    pVgEp->epSet = plan->execNode.epSet;
-    pVgEp->vgId = plan->execNode.nodeId;
-
-#if 0
-    SMqConsumerEp consumerEp = {0};
-    consumerEp.status = 0;
-    consumerEp.consumerId = -1;
-    consumerEp.epSet = plan->execNode.epSet;
-    consumerEp.vgId = plan->execNode.nodeId;
-#endif
+    pVgEp->epSet = mndGetVgroupEpset(pMnode, pVgroup);
+    pVgEp->vgId = pVgroup->vgId;
+    taosArrayPush(pEpInSub->vgs, &pVgEp);
 
     mDebug("init subscribption %s, assign vg: %d", pSub->key, pVgEp->vgId);
 
-    int32_t msgLen;
-    if (qSubPlanToString(plan, &pVgEp->qmsg, &msgLen) < 0) {
-      sdbRelease(pSdb, pVgroup);
-      qDestroyQueryPlan(pPlan);
-      terrno = TSDB_CODE_QRY_INVALID_INPUT;
-      return -1;
+    if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {
+      int32_t msgLen;
+
+      plan->execNode.epSet = pVgEp->epSet;
+      plan->execNode.nodeId = pVgEp->vgId;
+
+      if (qSubPlanToString(plan, &pVgEp->qmsg, &msgLen) < 0) {
+        sdbRelease(pSdb, pVgroup);
+        qDestroyQueryPlan(pPlan);
+        terrno = TSDB_CODE_QRY_INVALID_INPUT;
+        return -1;
+      }
+    } else {
+      pVgEp->qmsg = strdup("");
     }
-    taosArrayPush(pEpInSub->vgs, &pVgEp);
 
     ASSERT(taosHashGetSize(pSub->consumerHash) == 1);
 
     /*taosArrayPush(pSub->unassignedVg, &consumerEp);*/
   }
 
-  ASSERT(pEpInSub->vgs->size > 0);
   pEpInSub = taosHashGet(pSub->consumerHash, &unexistKey, sizeof(int64_t));
 
   ASSERT(pEpInSub->vgs->size > 0);
diff --git a/source/dnode/mnode/impl/src/mndTopic.c b/source/dnode/mnode/impl/src/mndTopic.c
index faddc3a1c5..e7a37658dc 100644
--- a/source/dnode/mnode/impl/src/mndTopic.c
+++ b/source/dnode/mnode/impl/src/mndTopic.c
@@ -282,10 +282,10 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq
   topicObj.version = 1;
   topicObj.sql = strdup(pCreate->sql);
   topicObj.sqlLen = strlen(pCreate->sql) + 1;
-  topicObj.ast = strdup(pCreate->ast);
-  topicObj.astLen = strlen(pCreate->ast) + 1;
 
   if (pCreate->ast && pCreate->ast[0]) {
+    topicObj.ast = strdup(pCreate->ast);
+    topicObj.astLen = strlen(pCreate->ast) + 1;
     topicObj.subType = TOPIC_SUB_TYPE__TABLE;
     topicObj.withTbName = 0;
     topicObj.withSchema = 0;
@@ -314,6 +314,9 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq
       return -1;
     }
   } else {
+    topicObj.ast = strdup("");
+    topicObj.astLen = 1;
+    topicObj.physicalPlan = strdup("");
     topicObj.subType = TOPIC_SUB_TYPE__DB;
     topicObj.withTbName = 1;
     topicObj.withSchema = 1;
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index 434c5bc5bb..fdb43a761b 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -411,8 +411,10 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
 
   SMqDataBlkRsp rsp = {0};
   rsp.reqOffset = pReq->currentOffset;
+  rsp.withSchema = pExec->withSchema;
   rsp.blockData = taosArrayInit(0, sizeof(void*));
   rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));
+  rsp.blockSchema = taosArrayInit(0, sizeof(void*));
 
   while (1) {
     consumerEpoch = atomic_load_32(&pExec->epoch);
@@ -512,6 +514,12 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
           ASSERT(actualLen <= dataStrLen);
           taosArrayPush(rsp.blockDataLen, &actualLen);
           taosArrayPush(rsp.blockData, &buf);
+
+          if (pExec->withSchema) {
+            SSchemaWrapper* pSW = tCloneSSchemaWrapper(pExec->pExecReader[workerId]->pSchemaWrapper);
+            taosArrayPush(rsp.blockSchema, &pSW);
+          }
+
           rsp.blockNum++;
         }
         // db subscribe
@@ -540,6 +548,10 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
           ASSERT(actualLen <= dataStrLen);
           taosArrayPush(rsp.blockDataLen, &actualLen);
           taosArrayPush(rsp.blockData, &buf);
+
+          SSchemaWrapper* pSW = tCloneSSchemaWrapper(pExec->pExecReader[workerId]->pSchemaWrapper);
+          taosArrayPush(rsp.blockSchema, &pSW);
+
           rsp.blockNum++;
         }
       } else {
@@ -586,6 +598,8 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
   // TODO destroy
   taosArrayDestroy(rsp.blockData);
   taosArrayDestroy(rsp.blockDataLen);
+  taosArrayDestroyP(rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
+
   return 0;
 }
 
@@ -827,12 +841,16 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
     pExec->pWalReader = walOpenReadHandle(pTq->pVnode->pWal);
     for (int32_t i = 0; i < 5; i++) {
       pExec->pExecReader[i] = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
-      SReadHandle handle = {
-          .reader = pExec->pExecReader[i],
-          .meta = pTq->pVnode->pMeta,
-      };
-      pExec->task[i] = qCreateStreamExecTaskInfo(pExec->qmsg, &handle);
-      ASSERT(pExec->task[i]);
+      if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
+        SReadHandle handle = {
+            .reader = pExec->pExecReader[i],
+            .meta = pTq->pVnode->pMeta,
+        };
+        pExec->task[i] = qCreateStreamExecTaskInfo(pExec->qmsg, &handle);
+        ASSERT(pExec->task[i]);
+      } else {
+        pExec->task[i] = NULL;
+      }
     }
     taosHashPut(pTq->execs, req.subKey, strlen(req.subKey), pExec, sizeof(STqExec));
     return 0;
diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c
index eb45577e0a..ef7e346867 100644
--- a/source/dnode/vnode/src/tq/tqRead.c
+++ b/source/dnode/vnode/src/tq/tqRead.c
@@ -65,7 +65,9 @@ bool tqNextDataBlock(STqReadHandle* pHandle) {
 
     /*pHandle->pBlock->uid = htobe64(pHandle->pBlock->uid);*/
     /*if (pHandle->tbUid == pHandle->pBlock->uid) {*/
-    ASSERT(pHandle->tbIdHash);
+    if (pHandle->tbIdHash == NULL) {
+      return true;
+    }
     void* ret = taosHashGet(pHandle->tbIdHash, &pHandle->pBlock->uid, sizeof(int64_t));
     if (ret != NULL) {
       /*printf("retrieve one tb %ld\n", pHandle->pBlock->uid);*/
@@ -107,26 +109,15 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
   *pNumOfRows = pHandle->pBlock->numOfRows;
   int32_t colNumNeed = taosArrayGetSize(pHandle->pColIdList);
 
-  if (colNumNeed > pSchemaWrapper->nCols) {
-    colNumNeed = pSchemaWrapper->nCols;
-  }
+  if (colNumNeed == 0) {
+    *ppCols = taosArrayInit(pSchemaWrapper->nCols, sizeof(SColumnInfoData));
+    if (*ppCols == NULL) {
+      return -1;
+    }
 
-  *ppCols = taosArrayInit(colNumNeed, sizeof(SColumnInfoData));
-  if (*ppCols == NULL) {
-    return -1;
-  }
-
-  int32_t colMeta = 0;
-  int32_t colNeed = 0;
-  while (colMeta < pSchemaWrapper->nCols && colNeed < colNumNeed) {
-    SSchema* pColSchema = &pSchemaWrapper->pSchema[colMeta];
-    col_id_t colIdSchema = pColSchema->colId;
-    col_id_t colIdNeed = *(col_id_t*)taosArrayGet(pHandle->pColIdList, colNeed);
-    if (colIdSchema < colIdNeed) {
-      colMeta++;
-    } else if (colIdSchema > colIdNeed) {
-      colNeed++;
-    } else {
+    int32_t colMeta = 0;
+    while (colMeta < pSchemaWrapper->nCols) {
+      SSchema*        pColSchema = &pSchemaWrapper->pSchema[colMeta];
       SColumnInfoData colInfo = {0};
       colInfo.info.bytes = pColSchema->bytes;
       colInfo.info.colId = pColSchema->colId;
@@ -137,7 +128,40 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
       }
       taosArrayPush(*ppCols, &colInfo);
       colMeta++;
-      colNeed++;
+    }
+  } else {
+    if (colNumNeed > pSchemaWrapper->nCols) {
+      colNumNeed = pSchemaWrapper->nCols;
+    }
+
+    *ppCols = taosArrayInit(colNumNeed, sizeof(SColumnInfoData));
+    if (*ppCols == NULL) {
+      return -1;
+    }
+
+    int32_t colMeta = 0;
+    int32_t colNeed = 0;
+    while (colMeta < pSchemaWrapper->nCols && colNeed < colNumNeed) {
+      SSchema* pColSchema = &pSchemaWrapper->pSchema[colMeta];
+      col_id_t colIdSchema = pColSchema->colId;
+      col_id_t colIdNeed = *(col_id_t*)taosArrayGet(pHandle->pColIdList, colNeed);
+      if (colIdSchema < colIdNeed) {
+        colMeta++;
+      } else if (colIdSchema > colIdNeed) {
+        colNeed++;
+      } else {
+        SColumnInfoData colInfo = {0};
+        colInfo.info.bytes = pColSchema->bytes;
+        colInfo.info.colId = pColSchema->colId;
+        colInfo.info.type = pColSchema->type;
+
+        if (colInfoDataEnsureCapacity(&colInfo, 0, *pNumOfRows) < 0) {
+          goto FAIL;
+        }
+        taosArrayPush(*ppCols, &colInfo);
+        colMeta++;
+        colNeed++;
+      }
     }
   }
 

From 56f3c9bf612f8a3c1b0bf3413f6ec2c6bceb7b0e Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Mon, 25 Apr 2022 17:12:28 +0800
Subject: [PATCH 05/13] fix(query): get the correct data block column number.

---
 source/libs/executor/src/executorimpl.c | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c
index 321e613f82..d0da970b2c 100644
--- a/source/libs/executor/src/executorimpl.c
+++ b/source/libs/executor/src/executorimpl.c
@@ -268,7 +268,6 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
   int32_t numOfCols = LIST_LENGTH(pNode->pSlots);
 
   SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
-  pBlock->info.numOfCols = numOfCols;
   pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
 
   pBlock->info.blockId = pNode->dataBlockId;
@@ -294,6 +293,7 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
     taosArrayPush(pBlock->pDataBlock, &idata);
   }
 
+  pBlock->info.numOfCols = taosArrayGetSize(pBlock->pDataBlock);
   return pBlock;
 }
 
@@ -1032,6 +1032,8 @@ static int32_t doCreateConstantValColumnInfo(SInputColumnInfoData* pInput, SFunc
     pColInfo->info.bytes = tDataTypes[type].bytes;
 
     pInput->pData[paramIndex] = pColInfo;
+  } else {
+    pColInfo = pInput->pData[paramIndex];
   }
 
   ASSERT(!IS_VAR_DATA_TYPE(type));

From 1d22c07188c102a7e55b15803ad2a9f8df02fe21 Mon Sep 17 00:00:00 2001
From: Liu Jicong <liujicong@qq.com>
Date: Mon, 25 Apr 2022 17:16:28 +0800
Subject: [PATCH 06/13] remove hb

---
 source/client/src/clientHb.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/client/src/clientHb.c b/source/client/src/clientHb.c
index 653f9a0634..b11a49fa1a 100644
--- a/source/client/src/clientHb.c
+++ b/source/client/src/clientHb.c
@@ -668,7 +668,7 @@ int hbMgrInit() {
   hbMgrInitHandle();
 
   // init backgroud thread
-  hbCreateThread();
+  /*hbCreateThread();*/
 
   return 0;
 }

From 8c36142a9c63b7180764be1b642ac6f04fc6d001 Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Mon, 25 Apr 2022 17:16:47 +0800
Subject: [PATCH 07/13] fix(query): fix scalar function input is constant
 return var type value contains illegal characters.

TD-15116
---
 source/libs/function/src/builtins.c | 2 +-
 source/libs/scalar/src/scalar.c     | 8 +++++---
 2 files changed, 6 insertions(+), 4 deletions(-)

diff --git a/source/libs/function/src/builtins.c b/source/libs/function/src/builtins.c
index a52e39bd4f..70087ee46b 100644
--- a/source/libs/function/src/builtins.c
+++ b/source/libs/function/src/builtins.c
@@ -355,7 +355,7 @@ static int32_t translateToIso8601(SFunctionNode* pFunc, char* pErrBuf, int32_t l
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
-  pFunc->node.resType = (SDataType) { .bytes = 24, .type = TSDB_DATA_TYPE_BINARY};
+  pFunc->node.resType = (SDataType) { .bytes = 64, .type = TSDB_DATA_TYPE_BINARY};
   return TSDB_CODE_SUCCESS;
 }
 
diff --git a/source/libs/scalar/src/scalar.c b/source/libs/scalar/src/scalar.c
index 820a4894b5..d6c0989f19 100644
--- a/source/libs/scalar/src/scalar.c
+++ b/source/libs/scalar/src/scalar.c
@@ -489,7 +489,7 @@ EDealRes sclRewriteFunction(SNode** pNode, SScalarCtx *ctx) {
   if (fmIsUserDefinedFunc(node->funcId)) {
     return DEAL_RES_CONTINUE;
   }
-  
+
   FOREACH(tnode, node->pParameterList) {
     if (!SCL_IS_CONST_NODE(tnode)) {
       return DEAL_RES_CONTINUE;
@@ -517,8 +517,10 @@ EDealRes sclRewriteFunction(SNode** pNode, SScalarCtx *ctx) {
     res->node.resType = node->node.resType;
     int32_t type = output.columnData->info.type;
     if (IS_VAR_DATA_TYPE(type)) {
-      res->datum.p = output.columnData->pData;
-      output.columnData->pData = NULL;
+      size_t tLen = varDataLen(output.columnData->pData);
+      ASSERT(tLen <= res->node.resType.bytes);
+      res->datum.p = taosMemoryCalloc(res->node.resType.bytes + VARSTR_HEADER_SIZE + 1, 1);
+      memcpy(res->datum.p, output.columnData->pData, tLen);
     } else {
       memcpy(nodesGetValueFromNode(res), output.columnData->pData, tDataTypes[type].bytes);
     }

From 28855b81b1e3b96ec15967f05b9eda0cc8b625b1 Mon Sep 17 00:00:00 2001
From: Ganlin Zhao <ganlinzhao@hotmail.com>
Date: Mon, 25 Apr 2022 17:16:47 +0800
Subject: [PATCH 08/13] fix(query): fix scalar function input is constant
 return var type value contains illegal characters.

TD-15116
---
 source/libs/scalar/src/scalar.c | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/source/libs/scalar/src/scalar.c b/source/libs/scalar/src/scalar.c
index d6c0989f19..6773f8192b 100644
--- a/source/libs/scalar/src/scalar.c
+++ b/source/libs/scalar/src/scalar.c
@@ -517,10 +517,8 @@ EDealRes sclRewriteFunction(SNode** pNode, SScalarCtx *ctx) {
     res->node.resType = node->node.resType;
     int32_t type = output.columnData->info.type;
     if (IS_VAR_DATA_TYPE(type)) {
-      size_t tLen = varDataLen(output.columnData->pData);
-      ASSERT(tLen <= res->node.resType.bytes);
       res->datum.p = taosMemoryCalloc(res->node.resType.bytes + VARSTR_HEADER_SIZE + 1, 1);
-      memcpy(res->datum.p, output.columnData->pData, tLen);
+      memcpy(res->datum.p, output.columnData->pData, varDataTLen(output.columnData->pData));
     } else {
       memcpy(nodesGetValueFromNode(res), output.columnData->pData, tDataTypes[type].bytes);
     }

From f2ea7c1fbd2979fe0154c1d69d191b217d54ea49 Mon Sep 17 00:00:00 2001
From: Shengliang Guan <slguan@taosdata.com>
Date: Mon, 25 Apr 2022 18:17:06 +0800
Subject: [PATCH 09/13] fix: add errstr for sync

---
 source/util/src/terror.c | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/source/util/src/terror.c b/source/util/src/terror.c
index 9332cb481e..0e2c322317 100644
--- a/source/util/src/terror.c
+++ b/source/util/src/terror.c
@@ -403,6 +403,9 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_CHECKSUM,         "Invalid msg checksum"
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_MSGLEN,           "Invalid msg length")
 TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INVALID_MSGTYPE,          "Invalid msg type")
 
+TAOS_DEFINE_ERROR(TSDB_CODE_SYN_NOT_LEADER,               "Sync not leader")
+TAOS_DEFINE_ERROR(TSDB_CODE_SYN_INTERNAL_ERROR,           "Sync internal error")
+
 // wal
 TAOS_DEFINE_ERROR(TSDB_CODE_WAL_APP_ERROR,                "Unexpected generic error in wal")
 TAOS_DEFINE_ERROR(TSDB_CODE_WAL_FILE_CORRUPTED,           "WAL file is corrupted")

From 3fc1ce71b3e5ad1855c73806d244a1ca37c4c843 Mon Sep 17 00:00:00 2001
From: Li Minghao <castermode@gmail.com>
Date: Mon, 25 Apr 2022 03:26:13 -0700
Subject: [PATCH 10/13] call vnodeStart after vnodeOpen

---
 source/dnode/mgmt/mgmt_vnode/src/vmHandle.c | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
index 4cc1b8527c..7fc263c93c 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c
@@ -202,6 +202,17 @@ int32_t vmProcessCreateVnodeReq(SVnodesMgmt *pMgmt, SNodeMsg *pMsg) {
     return code;
   }
 
+  code = vnodeStart(pImpl);
+  if (code != 0) {
+    tFreeSCreateVnodeReq(&createReq);
+    dError("vgId:%d, failed to start sync since %s", createReq.vgId, terrstr());
+    vnodeClose(pImpl);
+    vnodeDestroy(path, pMgmt->pTfs);
+    terrno = code;
+    return code;
+  }
+
+
   code = vmWriteVnodesToFile(pMgmt);
   if (code != 0) {
     tFreeSCreateVnodeReq(&createReq);

From 39daf0bc98353c3fb217902419138fb6c91ed1e3 Mon Sep 17 00:00:00 2001
From: Li Minghao <castermode@gmail.com>
Date: Mon, 25 Apr 2022 03:34:29 -0700
Subject: [PATCH 11/13] call vnodeStart after vnodeOpen

---
 source/dnode/mgmt/mgmt_vnode/src/vmInt.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
index 38f6d7648f..f995979ffd 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
@@ -147,7 +147,7 @@ static void *vmOpenVnodeFunc(void *param) {
       pThread->failed++;
     } else {
       vmOpenVnode(pMgmt, pCfg, pImpl);
-      vnodeStart(pImpl);
+      //vnodeStart(pImpl);
       dDebug("vgId:%d, is opened by thread:%d", pCfg->vgId, pThread->threadIndex);
       pThread->opened++;
     }
@@ -431,4 +431,4 @@ void vmGetVnodeLoads(SMgmtWrapper *pWrapper, SMonVloadInfo *pInfo) {
   }
 
   taosRUnLockLatch(&pMgmt->latch);
-}
\ No newline at end of file
+}

From c2b06004aa78ed69a355b347673abf47b855a0a2 Mon Sep 17 00:00:00 2001
From: Shengliang Guan <slguan@taosdata.com>
Date: Mon, 25 Apr 2022 19:21:07 +0800
Subject: [PATCH 12/13] fix: crash in multiprocess mode

---
 source/dnode/mgmt/mgmt_vnode/src/vmInt.c | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
index f995979ffd..3088c5dea4 100644
--- a/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
+++ b/source/dnode/mgmt/mgmt_vnode/src/vmInt.c
@@ -380,6 +380,7 @@ static int32_t vmStart(SMgmtWrapper *pWrapper) {
 }
 
 static void vmStop(SMgmtWrapper *pWrapper) {
+#if 0
   dDebug("vnode-mgmt start to stop");
   SVnodesMgmt *pMgmt = pWrapper->pMgmt;
   taosRLockLatch(&pMgmt->latch);
@@ -395,6 +396,7 @@ static void vmStop(SMgmtWrapper *pWrapper) {
   }
 
   taosRUnLockLatch(&pMgmt->latch);
+#endif
 }
 
 void vmSetMgmtFp(SMgmtWrapper *pWrapper) {

From 5a91ecfb7c04771ebf4a9a28e7ceae36973da7a9 Mon Sep 17 00:00:00 2001
From: Shengliang Guan <slguan@taosdata.com>
Date: Mon, 25 Apr 2022 19:29:15 +0800
Subject: [PATCH 13/13] fix: write histroy in shell

---
 tools/shell/src/shellEngine.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tools/shell/src/shellEngine.c b/tools/shell/src/shellEngine.c
index ac2be3c3cf..1f036ab25b 100644
--- a/tools/shell/src/shellEngine.c
+++ b/tools/shell/src/shellEngine.c
@@ -750,7 +750,7 @@ void shellReadHistory() {
 
 void shellWriteHistory() {
   SShellHistory *pHistory = &shell.history;
-  TdFilePtr      pFile = taosOpenFile(pHistory->file, TD_FILE_WRITE | TD_FILE_STREAM);
+  TdFilePtr      pFile = taosOpenFile(pHistory->file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_STREAM | TD_FILE_APPEND);
   if (pFile == NULL) return;
 
   for (int32_t i = pHistory->hstart; i != pHistory->hend;) {