diff --git a/cmake/taostools_CMakeLists.txt.in b/cmake/taostools_CMakeLists.txt.in
index 87f0579f44045f07794e522df4ee4f5ae1a0644d..897ccdd1584ab49aedd7ea0680891d3858a9ccf3 100644
--- a/cmake/taostools_CMakeLists.txt.in
+++ b/cmake/taostools_CMakeLists.txt.in
@@ -2,7 +2,7 @@
 # taos-tools
 ExternalProject_Add(taos-tools
         GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
-        GIT_TAG d11f210
+        GIT_TAG 04296a5
         SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
         BINARY_DIR ""
         #BUILD_IN_SOURCE TRUE
diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c
index 77bbd0be1a77bd0d673a52bb4b1842459fae76eb..111ca28cdc4153cde2ee07f54b630efaa312ba84 100644
--- a/source/client/src/clientTmq.c
+++ b/source/client/src/clientTmq.c
@@ -1116,6 +1116,7 @@ _failed:
 }
 
 int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
+  const int32_t   MAX_RETRY_COUNT = 120 * 2;  // let's wait for 2 mins at most
   const SArray*   container = &topic_list->container;
   int32_t         sz = taosArrayGetSize(container);
   void*           buf = NULL;
@@ -1209,7 +1210,7 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
 
   int32_t retryCnt = 0;
   while (TSDB_CODE_MND_CONSUMER_NOT_READY == tmqAskEp(tmq, false)) {
-    if (retryCnt++ > 40) {
+    if (retryCnt++ > MAX_RETRY_COUNT) {
       goto FAIL;
     }
 
@@ -1811,7 +1812,6 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
     if (pRspWrapper == NULL) {
       taosReadAllQitems(tmq->mqueue, tmq->qall);
       taosGetQitem(tmq->qall, (void**)&pRspWrapper);
-
       if (pRspWrapper == NULL) {
         return NULL;
       }
@@ -1831,7 +1831,6 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
       SMqDataRsp* pDataRsp = &pollRspWrapper->dataRsp;
 
       if (pDataRsp->head.epoch == consumerEpoch) {
-        // todo fix it: race condition
         SMqClientVg* pVg = pollRspWrapper->vgHandle;
 
         // update the epset
@@ -1843,6 +1842,7 @@ static void* tmqHandleAllRsp(tmq_t* tmq, int64_t timeout, bool pollIfReset) {
           pVg->epSet = *pollRspWrapper->pEpset;
         }
 
+        // update the local offset value only for the returned values.
         pVg->currentOffset = pDataRsp->rspOffset;
         atomic_store_32(&pVg->vgStatus, TMQ_VG_STATUS__IDLE);
 
diff --git a/source/dnode/vnode/src/inc/tq.h b/source/dnode/vnode/src/inc/tq.h
index bf390c163296a4575580f6f2f2974aa883e5f5f0..9037644602034819fd1da260c1cef02b7ae29a97 100644
--- a/source/dnode/vnode/src/inc/tq.h
+++ b/source/dnode/vnode/src/inc/tq.h
@@ -109,23 +109,18 @@ typedef struct {
 } STqPushEntry;
 
 struct STQ {
-  SVnode* pVnode;
-  char*   path;
-  int64_t walLogLastVer;
-
-  SRWLatch pushLock;
-
-  SHashObj* pPushMgr;    // consumerId -> STqPushEntry
-  SHashObj* pHandle;     // subKey -> STqHandle
-  SHashObj* pCheckInfo;  // topic -> SAlterCheckInfo
-
+  SVnode*         pVnode;
+  char*           path;
+  int64_t         walLogLastVer;
+  SRWLatch        lock;
+  SHashObj*       pPushMgr;    // consumerId -> STqPushEntry
+  SHashObj*       pHandle;     // subKey -> STqHandle
+  SHashObj*       pCheckInfo;  // topic -> SAlterCheckInfo
   STqOffsetStore* pOffsetStore;
-
-  TDB* pMetaDB;
-  TTB* pExecStore;
-  TTB* pCheckStore;
-
-  SStreamMeta* pStreamMeta;
+  TDB*            pMetaDB;
+  TTB*            pExecStore;
+  TTB*            pCheckStore;
+  SStreamMeta*    pStreamMeta;
 };
 
 typedef struct {
@@ -163,7 +158,7 @@ typedef struct {
   int32_t size;
 } STqOffsetHead;
 
-STqOffsetStore* tqOffsetOpen();
+STqOffsetStore* tqOffsetOpen(STQ* pTq);
 void            tqOffsetClose(STqOffsetStore*);
 STqOffset*      tqOffsetRead(STqOffsetStore* pStore, const char* subscribeKey);
 int32_t         tqOffsetWrite(STqOffsetStore* pStore, const STqOffset* pOffset);
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index 437870ebd61d7e045df509cde259bd8724760e59..81fa058f24333a4be9f77e622550a273d03fb665 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -51,7 +51,7 @@ void tqCleanUp() {
   }
 }
 
-static void destroySTqHandle(void* data) {
+static void destroyTqHandle(void* data) {
   STqHandle* pData = (STqHandle*)data;
   qDestroyTask(pData->execHandle.task);
   if (pData->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
@@ -89,9 +89,9 @@ STQ* tqOpen(const char* path, SVnode* pVnode) {
   pTq->walLogLastVer = pVnode->pWal->vers.lastVer;
 
   pTq->pHandle = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);
-  taosHashSetFreeFp(pTq->pHandle, destroySTqHandle);
+  taosHashSetFreeFp(pTq->pHandle, destroyTqHandle);
 
-  taosInitRWLatch(&pTq->pushLock);
+  taosInitRWLatch(&pTq->lock);
   pTq->pPushMgr = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
   taosHashSetFreeFp(pTq->pPushMgr, tqPushEntryFree);
 
@@ -236,38 +236,6 @@ int32_t tqPushDataRsp(STQ* pTq, STqPushEntry* pPushEntry) {
   }
 #endif
 
-//  int32_t len = 0;
-//  int32_t code = 0;
-//  tEncodeSize(tEncodeSMqDataRsp, pRsp, len, code);
-//  if (code < 0) {
-//    return -1;
-//  }
-//
-//  int32_t tlen = sizeof(SMqRspHead) + len;
-//  void*   buf = rpcMallocCont(tlen);
-//  if (buf == NULL) {
-//    return -1;
-//  }
-//
-//  memcpy(buf, &pPushEntry->dataRsp.head, sizeof(SMqRspHead));
-//
-//  void* abuf = POINTER_SHIFT(buf, sizeof(SMqRspHead));
-//
-//  SEncoder encoder = {0};
-//  tEncoderInit(&encoder, abuf, len);
-//  tEncodeSMqDataRsp(&encoder, pRsp);
-//  tEncoderClear(&encoder);
-//
-//  SRpcMsg rsp = {
-//      .info = pPushEntry->pInfo,
-//      .pCont = buf,
-//      .contLen = tlen,
-//      .code = 0,
-//  };
-//
-//  tmsgSendRsp(&rsp);
-//
-
   SMqRspHead* pHeader = &pPushEntry->pDataRsp->head;
   doSendDataRsp(&pPushEntry->info, pRsp, pHeader->epoch, pHeader->consumerId, pHeader->mqMsgType);
 
@@ -444,7 +412,7 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand
 
     char formatBuf[80];
     tFormatOffset(formatBuf, 80, pOffsetVal);
-    tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s, vgId:%d, prev offset found, offset reset to %s and continue.",
+    tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s, vgId:%d, existed offset found, offset reset to %s and continue.",
             consumerId, pHandle->subKey, vgId, formatBuf);
     return 0;
   } else {
@@ -502,7 +470,45 @@ static int32_t extractResetOffsetVal(STqOffsetVal* pOffsetVal, STQ* pTq, STqHand
   return 0;
 }
 
-static int32_t extractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest, SRpcMsg* pMsg) {
+#define IS_OFFSET_RESET_TYPE(_t)  ((_t) < 0)
+
+static int32_t extractDataAndRspForNormalSubscribe(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest,
+                                                   SRpcMsg* pMsg, STqOffsetVal* pOffset) {
+  int32_t  code = 0;
+  uint64_t consumerId = pRequest->consumerId;
+  int32_t  vgId = TD_VID(pTq->pVnode);
+
+  SMqDataRsp dataRsp = {0};
+  tqInitDataRsp(&dataRsp, pRequest, pHandle->execHandle.subType);
+
+  // lock
+  taosWLockLatch(&pTq->lock);
+
+  qSetTaskId(pHandle->execHandle.task, consumerId, pRequest->reqId);
+  code = tqScanData(pTq, pHandle, &dataRsp, pOffset);
+
+  // till now, all data has been transferred to consumer, new data needs to push client once arrived.
+  if (dataRsp.blockNum == 0 && dataRsp.reqOffset.type == TMQ_OFFSET__LOG &&
+      dataRsp.reqOffset.version == dataRsp.rspOffset.version && pHandle->consumerId == pRequest->consumerId) {
+    code = tqRegisterPushEntry(pTq, pHandle, pRequest, pMsg, &dataRsp, TMQ_MSG_TYPE__POLL_RSP);
+    taosWUnLockLatch(&pTq->lock);
+    return code;
+  }
+
+  taosWUnLockLatch(&pTq->lock);
+  code = tqSendDataRsp(pTq, pMsg, pRequest, (SMqDataRsp*)&dataRsp, TMQ_MSG_TYPE__POLL_RSP);
+
+  // NOTE: this pHandle->consumerId may have been changed already.
+  tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s, vgId:%d, rsp block:%d, offset type:%d, uid/version:%" PRId64
+          ", ts:%" PRId64 ", reqId:0x%" PRIx64,
+          consumerId, pHandle->subKey, vgId, dataRsp.blockNum, dataRsp.rspOffset.type, dataRsp.rspOffset.uid,
+          dataRsp.rspOffset.ts, pRequest->reqId);
+
+  tDeleteSMqDataRsp(&dataRsp);
+  return code;
+}
+
+static int32_t doPollDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest, SRpcMsg* pMsg) {
   int32_t      code = -1;
   STqOffsetVal offset = {0};
   SWalCkHead*  pCkHead = NULL;
@@ -512,9 +518,8 @@ static int32_t extractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq*
   uint64_t     consumerId = pRequest->consumerId;
 
   // 1. reset the offset if needed
-  if (reqOffset.type > 0) {
-    offset = reqOffset;
-  } else { // handle the reset offset cases, according to the consumer's choice.
+  if (IS_OFFSET_RESET_TYPE(reqOffset.type)) {
+    // handle the reset offset cases, according to the consumer's choice.
     bool blockReturned = false;
     code = extractResetOffsetVal(&offset, pTq, pHandle, pRequest, pMsg, &blockReturned);
     if (code != 0) {
@@ -525,38 +530,14 @@ static int32_t extractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq*
     if (blockReturned) {
       return 0;
     }
+  } else { // use the consumer specified offset
+    // the offset value can not be monotonious increase??
+    offset = reqOffset;
   }
 
-  // this is a normal subscription requirement
+  // this is a normal subscribe requirement
   if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
-    SMqDataRsp dataRsp = {0};
-    tqInitDataRsp(&dataRsp, pRequest, pHandle->execHandle.subType);
-
-    // lock
-    taosWLockLatch(&pTq->pushLock);
-
-    qSetTaskId(pHandle->execHandle.task, consumerId, pRequest->reqId);
-    code = tqScanData(pTq, pHandle, &dataRsp, &offset);
-
-    // till now, all data has been transferred to consumer, new data needs to push client once arrived.
-    if (dataRsp.blockNum == 0 && dataRsp.reqOffset.type == TMQ_OFFSET__LOG &&
-        dataRsp.reqOffset.version == dataRsp.rspOffset.version && pHandle->consumerId == pRequest->consumerId) {
-      code = tqRegisterPushEntry(pTq, pHandle, pRequest, pMsg, &dataRsp, TMQ_MSG_TYPE__POLL_RSP);
-      taosWUnLockLatch(&pTq->pushLock);
-      return code;
-    }
-
-    taosWUnLockLatch(&pTq->pushLock);
-    code = tqSendDataRsp(pTq, pMsg, pRequest, (SMqDataRsp*)&dataRsp, TMQ_MSG_TYPE__POLL_RSP);
-
-    // NOTE: this pHandle->consumerId may have been changed already.
-    tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s, vgId:%d, rsp block:%d, offset type:%d, uid/version:%" PRId64
-            ", ts:%" PRId64", reqId:0x%"PRIx64,
-            consumerId, pHandle->subKey, vgId, dataRsp.blockNum, dataRsp.rspOffset.type, dataRsp.rspOffset.uid,
-            dataRsp.rspOffset.ts, pRequest->reqId);
-
-    tDeleteSMqDataRsp(&dataRsp);
-    return code;
+    return extractDataAndRspForNormalSubscribe(pTq, pHandle, pRequest, pMsg, &offset);
   }
 
   // todo handle the case where re-balance occurs.
@@ -573,7 +554,7 @@ static int32_t extractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq*
     if (metaRsp.metaRspLen > 0) {
       code = tqSendMetaPollRsp(pTq, pMsg, pRequest, &metaRsp);
       tqDebug("tmq poll: consumer:0x%" PRIx64 " subkey:%s vgId:%d, send meta offset type:%d,uid:%" PRId64
-              ",ts:%" PRId64,
+                  ",ts:%" PRId64,
               consumerId, pHandle->subKey, vgId, metaRsp.rspOffset.type, metaRsp.rspOffset.uid,
               metaRsp.rspOffset.ts);
       taosMemoryFree(metaRsp.metaRsp);
@@ -590,7 +571,7 @@ static int32_t extractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq*
     }
 
     tqDebug("taosx poll: consumer:0x%" PRIx64 " subkey:%s vgId:%d, send data blockNum:%d, offset type:%d,uid:%" PRId64
-            ",version:%" PRId64,
+                ",version:%" PRId64,
             consumerId, pHandle->subKey, vgId, taosxRsp.blockNum, taosxRsp.rspOffset.type, taosxRsp.rspOffset.uid,
             taosxRsp.rspOffset.version);
   }
@@ -610,7 +591,7 @@ static int32_t extractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq*
       int32_t savedEpoch = atomic_load_32(&pHandle->epoch);
       if (savedEpoch > pRequest->epoch) {
         tqWarn("tmq poll: consumer:0x%" PRIx64 " (epoch %d), subkey:%s vgId:%d offset %" PRId64
-               ", found new consumer epoch %d, discard req epoch %d",
+                   ", found new consumer epoch %d, discard req epoch %d",
                consumerId, pRequest->epoch, pHandle->subKey, vgId, fetchVer, savedEpoch, pRequest->epoch);
         break;
       }
@@ -708,31 +689,31 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
   }
 
   // 2. check re-balance status
-  taosRLockLatch(&pTq->pushLock);
+  taosRLockLatch(&pTq->lock);
   if (pHandle->consumerId != consumerId) {
     tqDebug("ERROR tmq poll: consumer:0x%" PRIx64 " vgId:%d, subkey %s, mismatch for saved handle consumer:0x%" PRIx64,
             consumerId, TD_VID(pTq->pVnode), req.subKey, pHandle->consumerId);
     terrno = TSDB_CODE_TMQ_CONSUMER_MISMATCH;
-    taosRUnLockLatch(&pTq->pushLock);
+    taosRUnLockLatch(&pTq->lock);
     return -1;
   }
-  taosRUnLockLatch(&pTq->pushLock);
+  taosRUnLockLatch(&pTq->lock);
 
-  taosWLockLatch(&pTq->pushLock);
   // 3. update the epoch value
+  taosWLockLatch(&pTq->lock);
   int32_t savedEpoch = pHandle->epoch;
   if (savedEpoch < reqEpoch) {
     tqDebug("tmq poll: consumer:0x%" PRIx64 " epoch update from %d to %d by poll req", consumerId, savedEpoch, reqEpoch);
     pHandle->epoch = reqEpoch;
   }
-  taosWUnLockLatch(&pTq->pushLock);
+  taosWUnLockLatch(&pTq->lock);
 
   char buf[80];
   tFormatOffset(buf, 80, &reqOffset);
   tqDebug("tmq poll: consumer:0x%" PRIx64 " (epoch %d), subkey %s, recv poll req vgId:%d, req:%s, reqId:0x%" PRIx64,
           consumerId, req.epoch, pHandle->subKey, vgId, buf, req.reqId);
 
-  return extractDataForMq(pTq, pHandle, &req, pMsg);
+  return doPollDataForMq(pTq, pHandle, &req, pMsg);
 }
 
 int32_t tqProcessDeleteSubReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
@@ -740,12 +721,12 @@ int32_t tqProcessDeleteSubReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
 
   tqDebug("vgId:%d, tq process delete sub req %s", pTq->pVnode->config.vgId, pReq->subKey);
 
-  taosWLockLatch(&pTq->pushLock);
+  taosWLockLatch(&pTq->lock);
   int32_t code = taosHashRemove(pTq->pPushMgr, pReq->subKey, strlen(pReq->subKey));
   if (code != 0) {
     tqDebug("vgId:%d, tq remove push handle %s", pTq->pVnode->config.vgId, pReq->subKey);
   }
-  taosWUnLockLatch(&pTq->pushLock);
+  taosWUnLockLatch(&pTq->lock);
 
   STqHandle* pHandle = taosHashGet(pTq->pHandle, pReq->subKey, strlen(pReq->subKey));
   if (pHandle) {
@@ -809,18 +790,18 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
   SVnode* pVnode = pTq->pVnode;
   int32_t vgId = TD_VID(pVnode);
 
-  tqDebug("vgId:%d, tq process sub req %s, Id:0x%" PRIx64 " -> Id:0x%" PRIx64, pVnode->config.vgId, req.subKey,
+  tqDebug("vgId:%d, tq process sub req:%s, Id:0x%" PRIx64 " -> Id:0x%" PRIx64, pVnode->config.vgId, req.subKey,
           req.oldConsumerId, req.newConsumerId);
 
   STqHandle* pHandle = taosHashGet(pTq->pHandle, req.subKey, strlen(req.subKey));
   if (pHandle == NULL) {
     if (req.oldConsumerId != -1) {
-      tqError("vgId:%d, build new consumer handle %s for consumer:0x%" PRIx64 ", but old consumerId is %" PRId64 "",
+      tqError("vgId:%d, build new consumer handle %s for consumer:0x%" PRIx64 ", but old consumerId:0x%" PRIx64,
               req.vgId, req.subKey, req.newConsumerId, req.oldConsumerId);
     }
 
     if (req.newConsumerId == -1) {
-      tqError("vgId:%d, tq invalid rebalance request, new consumerId %" PRId64 "", req.vgId, req.newConsumerId);
+      tqError("vgId:%d, tq invalid re-balance request, new consumerId %" PRId64 "", req.vgId, req.newConsumerId);
       taosMemoryFree(req.qmsg);
       return 0;
     }
@@ -910,28 +891,28 @@ int32_t tqProcessSubscribeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msg
       atomic_add_fetch_32(&pHandle->epoch, 1);
       taosMemoryFree(req.qmsg);
       return tqMetaSaveHandle(pTq, req.subKey, pHandle);
-    }
-
-    tqInfo("vgId:%d switch consumer from Id:0x%" PRIx64 " to Id:0x%" PRIx64, req.vgId, pHandle->consumerId,
-           req.newConsumerId);
+    } else {
+      tqInfo("vgId:%d switch consumer from Id:0x%" PRIx64 " to Id:0x%" PRIx64, req.vgId, pHandle->consumerId,
+             req.newConsumerId);
 
-    taosWLockLatch(&pTq->pushLock);
-    atomic_store_32(&pHandle->epoch, -1);
+      taosWLockLatch(&pTq->lock);
+      atomic_store_32(&pHandle->epoch, -1);
 
-    // remove if it has been register in the push manager, and return one empty block to consumer
-    tqRemovePushEntry(pTq, req.subKey, (int32_t) strlen(req.subKey), pHandle->consumerId, true);
+      // remove if it has been register in the push manager, and return one empty block to consumer
+      tqRemovePushEntry(pTq, req.subKey, (int32_t)strlen(req.subKey), pHandle->consumerId, true);
 
-    atomic_store_64(&pHandle->consumerId, req.newConsumerId);
-    atomic_add_fetch_32(&pHandle->epoch, 1);
+      atomic_store_64(&pHandle->consumerId, req.newConsumerId);
+      atomic_add_fetch_32(&pHandle->epoch, 1);
 
-    if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
-      qStreamCloseTsdbReader(pHandle->execHandle.task);
-    }
+      if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) {
+        qStreamCloseTsdbReader(pHandle->execHandle.task);
+      }
 
-    taosWUnLockLatch(&pTq->pushLock);
-    if (tqMetaSaveHandle(pTq, req.subKey, pHandle) < 0) {
-      taosMemoryFree(req.qmsg);
-      return -1;
+      taosWUnLockLatch(&pTq->lock);
+      if (tqMetaSaveHandle(pTq, req.subKey, pHandle) < 0) {
+        taosMemoryFree(req.qmsg);
+        return -1;
+      }
     }
   }
 
diff --git a/source/dnode/vnode/src/tq/tqPush.c b/source/dnode/vnode/src/tq/tqPush.c
index 01d8e7cf14790f18285095914c2f8ba26a885f45..797aeb3f04357787718c6597933fd9c920980b4f 100644
--- a/source/dnode/vnode/src/tq/tqPush.c
+++ b/source/dnode/vnode/src/tq/tqPush.c
@@ -213,7 +213,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
 
   if (msgType == TDMT_VND_SUBMIT) {
     // lock push mgr to avoid potential msg lost
-    taosWLockLatch(&pTq->pushLock);
+    taosWLockLatch(&pTq->lock);
 
     int32_t numOfRegisteredPush = taosHashGetSize(pTq->pPushMgr);
     if (numOfRegisteredPush > 0) {
@@ -231,7 +231,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
         taosArrayDestroy(cachedKeyLens);
 
         // unlock
-        taosWUnLockLatch(&pTq->pushLock);
+        taosWUnLockLatch(&pTq->lock);
         return -1;
       }
 
@@ -320,7 +320,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
       taosMemoryFree(data);
     }
     // unlock
-    taosWUnLockLatch(&pTq->pushLock);
+    taosWUnLockLatch(&pTq->lock);
   }
 
   if (!tsDisableStream && vnodeIsRoleLeader(pTq->pVnode)) {
diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c
index 5ad9276c6c06c28b72fcb93bd11a28e77ff329da..96bce02b674476b760114099de9d93eb0a735d72 100644
--- a/source/dnode/vnode/src/tsdb/tsdbRead.c
+++ b/source/dnode/vnode/src/tsdb/tsdbRead.c
@@ -4507,6 +4507,7 @@ static void doFillNullColSMA(SBlockLoadSuppInfo* pSup, int32_t numOfRows, int32_
   int32_t i = 0, j = 0;
   int32_t size = (int32_t)taosArrayGetSize(pSup->pColAgg);
   taosArrayInsert(pSup->pColAgg, 0, pTsAgg);
+  size++;
 
   while (j < numOfCols && i < size) {
     SColumnDataAgg* pAgg = taosArrayGet(pSup->pColAgg, i);
@@ -4519,10 +4520,21 @@ static void doFillNullColSMA(SBlockLoadSuppInfo* pSup, int32_t numOfRows, int32_
       if (pSup->colId[j] != PRIMARYKEY_TIMESTAMP_COL_ID) {
         SColumnDataAgg nullColAgg = {.colId = pSup->colId[j], .numOfNull = numOfRows};
         taosArrayInsert(pSup->pColAgg, i, &nullColAgg);
+        i += 1;
+        size++;
       }
       j += 1;
     }
   }
+
+  while (j < numOfCols) {
+    if (pSup->colId[j] != PRIMARYKEY_TIMESTAMP_COL_ID) {
+      SColumnDataAgg nullColAgg = {.colId = pSup->colId[j], .numOfNull = numOfRows};
+      taosArrayInsert(pSup->pColAgg, i, &nullColAgg);
+      i += 1;
+    }
+    j++;
+  }
 }
 
 int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SSDataBlock* pDataBlock, bool* allHave) {
@@ -4602,8 +4614,8 @@ int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SSDataBlock* pDataBlock,
     } else if (pAgg->colId < pSup->colId[j]) {
       i += 1;
     } else if (pSup->colId[j] < pAgg->colId) {
-      // ASSERT(pSup->colId[j] == PRIMARYKEY_TIMESTAMP_COL_ID);
-      pResBlock->pBlockAgg[pSup->slotId[j]] = &pSup->tsColAgg;
+      pResBlock->pBlockAgg[pSup->slotId[j]] = NULL;
+      *allHave = false;
       j += 1;
     }
   }
@@ -4996,4 +5008,4 @@ void tsdbUntakeReadSnap(STsdbReader* pReader, STsdbReadSnap* pSnap, bool proacti
 void tsdbReaderSetId(STsdbReader* pReader, const char* idstr) {
   taosMemoryFreeClear(pReader->idStr);
   pReader->idStr = taosStrdup(idstr);
-}
\ No newline at end of file
+}
diff --git a/source/libs/parser/src/parUtil.c b/source/libs/parser/src/parUtil.c
index 793d05721ec5e35cee4a3afd85375e2826ccffc2..563bc5e7800ddbb8752b4070bd203ad3410bd5c1 100644
--- a/source/libs/parser/src/parUtil.c
+++ b/source/libs/parser/src/parUtil.c
@@ -444,7 +444,7 @@ static int32_t getInsTagsTableTargetNameFromOp(int32_t acctId, SOperatorNode* pO
   } else if (QUERY_NODE_VALUE == nodeType(pOper->pRight)) {
     pVal = (SValueNode*)pOper->pRight;
   }
-  if (NULL == pCol || NULL == pVal) {
+  if (NULL == pCol || NULL == pVal || NULL == pVal->literal || 0 == strcmp(pVal->literal, "")) {
     return TSDB_CODE_SUCCESS;
   }
 
diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task
index 8c8d41d5b617f64c1cbf4173d2bb59df99220243..58fa0cbfab4f0dc5ff4d63495564f29b8d7c82c1 100644
--- a/tests/parallel_test/cases.task
+++ b/tests/parallel_test/cases.task
@@ -871,6 +871,7 @@
 ,,y,script,./test.sh -f tsim/query/emptyTsRange.sim
 ,,y,script,./test.sh -f tsim/query/partitionby.sim
 ,,y,script,./test.sh -f tsim/query/tableCount.sim 
+,,y,script,./test.sh -f tsim/query/nullColSma.sim
 ,,y,script,./test.sh -f tsim/qnode/basic1.sim
 ,,y,script,./test.sh -f tsim/snode/basic1.sim
 ,,y,script,./test.sh -f tsim/mnode/basic1.sim
diff --git a/tests/script/tsim/query/nullColSma.sim b/tests/script/tsim/query/nullColSma.sim
new file mode 100644
index 0000000000000000000000000000000000000000..886274e7e59ca22a557a8ed861585592d5402fbb
--- /dev/null
+++ b/tests/script/tsim/query/nullColSma.sim
@@ -0,0 +1,139 @@
+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start
+sql connect
+
+$dbPrefix = m_in_db
+$tbPrefix = m_in_tb
+$mtPrefix = m_in_mt
+$tbNum = 1
+$rowNum = 200
+$totalNum = 400
+
+print =============== step1
+$i = 0
+$db = $dbPrefix . $i
+$mt = $mtPrefix . $i
+
+sql drop database if exists $db
+sql create database $db vgroups 1 maxrows 200 minrows 10;
+sql use $db
+sql create table $mt (ts timestamp, f1 int, f2 float) TAGS(tgcol int)
+
+print ====== start create child tables and insert data
+$i = 0
+while $i < $tbNum
+  $tb = $tbPrefix . $i
+  sql create table $tb using $mt tags( $i )
+
+  $x = 0
+  while $x < $rowNum
+    $cc = $x * 1
+    $ms = 1601481600000 + $cc
+
+    sql insert into $tb values ($ms , NULL , $x )
+    $x = $x + 1
+  endw
+
+  $i = $i + 1
+endw
+
+$i = 1
+$tb = $tbPrefix . $i
+sql create table $tb using $mt tags( $i )
+
+$x = 0
+while $x < $rowNum
+  $cc = $x * 1
+  $ms = 1601481600000 + $cc
+
+  sql insert into $tb values ($ms , $x , NULL )
+  $x = $x + 1
+endw
+
+sql flush database $db
+
+print =============== step2
+$i = 0
+$tb = $tbPrefix . $i
+sql select max(f1) from $tb
+if $rows != 1 then
+  return -1
+endi
+if $data00 != NULL then
+  return -1
+endi
+
+$i = 1
+$tb = $tbPrefix . $i
+sql select max(f2) from $tb
+if $rows != 1 then
+  return -1
+endi
+if $data00 != NULL then
+  return -1
+endi
+
+$rowNum = 10
+
+print ====== insert more data
+$i = 0
+while $i < $tbNum
+  $tb = $tbPrefix . $i
+
+  $x = 0
+  while $x < $rowNum
+    $cc = $x * 1
+    $ms = 1601481700000 + $cc
+
+    sql insert into $tb values ($ms , $x , $x )
+    $x = $x + 1
+  endw
+
+  $i = $i + 1
+endw
+
+$i = 1
+$tb = $tbPrefix . $i
+$x = 0
+while $x < $rowNum
+  $cc = $x * 1
+  $ms = 1601481700000 + $cc
+
+  sql insert into $tb values ($ms , $x , $x )
+  $x = $x + 1
+endw
+
+sql flush database $db
+
+print =============== step3
+$i = 0
+$tb = $tbPrefix . $i
+sql select max(f1) from $tb
+if $rows != 1 then
+  return -1
+endi
+if $data00 != 9 then
+  return -1
+endi
+
+$i = 1
+$tb = $tbPrefix . $i
+sql select max(f2) from $tb
+if $rows != 1 then
+  return -1
+endi
+if $data00 != 9.00000 then
+  print $data00
+  return -1
+endi
+
+
+print =============== clear
+#sql drop database $db
+#sql select * from information_schema.ins_databases
+#if $rows != 0 then
+#  return -1
+#endi
+
+system sh/exec.sh -n dnode1 -s stop -x SIGINT
diff --git a/tests/script/tsim/query/sys_tbname.sim b/tests/script/tsim/query/sys_tbname.sim
index 7b3953129a81d6fa56b093c8b048b2071b6fbe70..c676f2b1e0b7168c5e83e9b5d2c9b07a1164effb 100644
--- a/tests/script/tsim/query/sys_tbname.sim
+++ b/tests/script/tsim/query/sys_tbname.sim
@@ -51,6 +51,11 @@ if $data00 != @ins_stables@ then
 	return -1
 endi
 
+sql select * from information_schema.ins_tables where table_name='';
+if $rows != 0 then 
+        return -1
+endi
+
 sql select tbname from information_schema.ins_tables;
 print $rows $data00
 if $rows != 33 then
diff --git a/tools/shell/src/shellEngine.c b/tools/shell/src/shellEngine.c
index 3080b15b8c5b644781bfd1b77bf03bd0f731cd7c..f9602aaa4dfb0f2ea7b68216c239df66ceae62e1 100644
--- a/tools/shell/src/shellEngine.c
+++ b/tools/shell/src/shellEngine.c
@@ -541,7 +541,7 @@ void shellPrintField(const char *val, TAOS_FIELD *field, int32_t width, int32_t
       printf("%*" PRIu64, width, *((uint64_t *)val));
       break;
     case TSDB_DATA_TYPE_FLOAT:
-      printf("%*.5f", width, GET_FLOAT_VAL(val));
+      printf("%*ef", width, GET_FLOAT_VAL(val));
       break;
     case TSDB_DATA_TYPE_DOUBLE:
       n = snprintf(buf, TSDB_MAX_BYTES_PER_ROW, "%*.9f", width, GET_DOUBLE_VAL(val));