Merge pull request #20422 from taosdata/fix/liaohj

fix(tmq): fix some race condition and do some internal refactor.

include/common/tcommon.h

@@ -25,13 +25,6 @@
 extern "C" {
 #endif
 
-// TODO remove it
-enum {
-  TMQ_CONF__RESET_OFFSET__NONE = -3,
-  TMQ_CONF__RESET_OFFSET__EARLIEAST = -2,
-  TMQ_CONF__RESET_OFFSET__LATEST = -1,
-};
-
 // clang-format off
 #define IS_META_MSG(x) ( \
      x == TDMT_VND_CREATE_STB     \

include/common/tmsg.h

@@ -3186,6 +3186,7 @@ typedef struct {
   SArray*      blockData;
   SArray*      blockTbName;
   SArray*      blockSchema;
+  // the following attributes are extended from SMqDataRsp
   int32_t      createTableNum;
   SArray*      createTableLen;
   SArray*      createTableReq;

source/client/test/clientTests.cpp

@@ -898,7 +898,86 @@ TEST(clientCase, update_test) {
   }
 }
 
-TEST(clientCase, subscription_test) {
+TEST(clientCase, sub_db_test) {
+  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  ASSERT_NE(pConn, nullptr);
+
+  //  TAOS_RES* pRes = taos_query(pConn, "create topic topic_t1 as select * from t1");
+  //  if (taos_errno(pRes) != TSDB_CODE_SUCCESS) {
+  //    printf("failed to create topic, code:%s", taos_errstr(pRes));
+  //    taos_free_result(pRes);
+  //    return;
+  //  }
+
+  tmq_conf_t* conf = tmq_conf_new();
+  tmq_conf_set(conf, "enable.auto.commit", "true");
+  tmq_conf_set(conf, "auto.commit.interval.ms", "1000");
+  tmq_conf_set(conf, "group.id", "cgrpNamedb");
+  tmq_conf_set(conf, "td.connect.user", "root");
+  tmq_conf_set(conf, "td.connect.pass", "taosdata");
+  tmq_conf_set(conf, "auto.offset.reset", "earliest");
+  tmq_conf_set(conf, "experimental.snapshot.enable", "true");
+  tmq_conf_set(conf, "msg.with.table.name", "true");
+  tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL);
+
+  tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
+  tmq_conf_destroy(conf);
+
+  // 创建订阅 topics 列表
+  tmq_list_t* topicList = tmq_list_new();
+  tmq_list_append(topicList, "topic_db1");
+
+  // 启动订阅
+  tmq_subscribe(tmq, topicList);
+  tmq_list_destroy(topicList);
+
+  TAOS_FIELD* fields = NULL;
+  int32_t     numOfFields = 0;
+  int32_t     precision = 0;
+  int32_t     totalRows = 0;
+  int32_t     msgCnt = 0;
+  int32_t     timeout = 5000;
+
+  int32_t count = 0;
+
+  while (1) {
+    TAOS_RES* pRes = tmq_consumer_poll(tmq, timeout);
+    if (pRes) {
+      char    buf[1024];
+      int32_t rows = 0;
+
+      const char* topicName = tmq_get_topic_name(pRes);
+      const char* dbName = tmq_get_db_name(pRes);
+      int32_t     vgroupId = tmq_get_vgroup_id(pRes);
+
+      printf("topic: %s\n", topicName);
+      printf("db: %s\n", dbName);
+      printf("vgroup id: %d\n", vgroupId);
+
+      if (count ++ > 200) {
+        tmq_unsubscribe(tmq);
+        break;
+      }
+
+      while (1) {
+        TAOS_ROW row = taos_fetch_row(pRes);
+        if (row == NULL) break;
+
+        fields = taos_fetch_fields(pRes);
+        numOfFields = taos_field_count(pRes);
+        precision = taos_result_precision(pRes);
+        rows++;
+        taos_print_row(buf, row, fields, numOfFields);
+        printf("precision: %d, row content: %s\n", precision, buf);
+      }
+    }
+//      return rows;
+  }
+
+  fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
+}
+
+TEST(clientCase, sub_tb_test) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
   ASSERT_NE(pConn, nullptr);
 
@@ -925,7 +1004,7 @@ TEST(clientCase, subscription_test) {
 
   // 创建订阅 topics 列表
   tmq_list_t* topicList = tmq_list_new();
-//  tmq_list_append(topicList, "topic_t1");
+  tmq_list_append(topicList, "topic_t1");
 
   // 启动订阅
   tmq_subscribe(tmq, topicList);
@@ -954,7 +1033,7 @@ TEST(clientCase, subscription_test) {
       printf("db: %s\n", dbName);
       printf("vgroup id: %d\n", vgroupId);
 
-      if (count ++ > 20) {
+      if (count ++ > 200) {
         tmq_unsubscribe(tmq);
         break;
       }

source/common/src/tmsg.c

@@ -6639,8 +6639,9 @@ int32_t tFormatOffset(char *buf, int32_t maxLen, const STqOffsetVal *pVal) {
   } else if (pVal->type == TMQ_OFFSET__SNAPSHOT_DATA || pVal->type == TMQ_OFFSET__SNAPSHOT_META) {
     snprintf(buf, maxLen, "offset(snapshot) uid:%" PRId64 " ts:%" PRId64, pVal->uid, pVal->ts);
   } else {
-    ASSERT(0);
+    return TSDB_CODE_INVALID_PARA;
   }
+
   return 0;
 }
 
@@ -6823,8 +6824,7 @@ int32_t tDecodeSMqDataRsp(SDecoder *pDecoder, SMqDataRsp *pRsp) {
 }
 
 void tDeleteSMqDataRsp(SMqDataRsp *pRsp) {
-  taosArrayDestroy(pRsp->blockDataLen);
-  pRsp->blockDataLen = NULL;
+  pRsp->blockDataLen = taosArrayDestroy(pRsp->blockDataLen);;
   taosArrayDestroyP(pRsp->blockData, (FDelete)taosMemoryFree);
   pRsp->blockData = NULL;
   taosArrayDestroyP(pRsp->blockSchema, (FDelete)tDeleteSSchemaWrapper);

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

@@ -24,10 +24,10 @@ extern "C" {
 
 enum {
   MQ_CONSUMER_STATUS__MODIFY = 1,
-  MQ_CONSUMER_STATUS__MODIFY_IN_REB,
+  MQ_CONSUMER_STATUS__MODIFY_IN_REB,     // this value is not used anymore
   MQ_CONSUMER_STATUS__READY,
   MQ_CONSUMER_STATUS__LOST,
-  MQ_CONSUMER_STATUS__LOST_IN_REB,
+  MQ_CONSUMER_STATUS__LOST_IN_REB,       // this value is not used anymore
   MQ_CONSUMER_STATUS__LOST_REBD,
   MQ_CONSUMER_STATUS__REMOVED,
 };

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

@@ -317,7 +317,7 @@ static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg) {
         taosArrayPush(pRebSub->removedConsumers, &pConsumer->consumerId);
       }
       taosRUnLockLatch(&pConsumer->lock);
-    } else if (status == MQ_CONSUMER_STATUS__MODIFY) {
+    } else if (status == MQ_CONSUMER_STATUS__MODIFY || status == MQ_CONSUMER_STATUS__MODIFY_IN_REB) {
       taosRLockLatch(&pConsumer->lock);
 
       int32_t newTopicNum = taosArrayGetSize(pConsumer->rebNewTopics);

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

@@ -224,7 +224,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
             .pVgEp = pVgEp,
         };
         taosHashPut(pHash, &pVgEp->vgId, sizeof(int32_t), &outputVg, sizeof(SMqRebOutputVg));
-        mInfo("sub:%s mq re-balance remove vgId:%d from consumer:%" PRIx64, sub, pVgEp->vgId, consumerId);
+        mInfo("sub:%s mq re-balance remove vgId:%d from consumer:0x%" PRIx64, sub, pVgEp->vgId, consumerId);
       }
       taosArrayDestroy(pConsumerEp->vgs);
       taosHashRemove(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t));
@@ -329,7 +329,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
       newConsumerEp.vgs = taosArrayInit(0, sizeof(void *));
       taosHashPut(pOutput->pSub->consumerHash, &consumerId, sizeof(int64_t), &newConsumerEp, sizeof(SMqConsumerEp));
       taosArrayPush(pOutput->newConsumers, &consumerId);
-      mInfo("sub:%s mq rebalance add new consumer:%" PRIx64, sub, consumerId);
+      mInfo("sub:%s mq rebalance add new consumer:0x%" PRIx64, sub, consumerId);
     }
   }
 
@@ -357,7 +357,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
       taosArrayPush(pConsumerEp->vgs, &pRebVg->pVgEp);
       pRebVg->newConsumerId = pConsumerEp->consumerId;
       taosArrayPush(pOutput->rebVgs, pRebVg);
-      mInfo("mq rebalance: add vgId:%d to consumer:%" PRIx64 " (second scan) (not enough)", pRebVg->pVgEp->vgId,
+      mInfo("mq rebalance: add vgId:%d to consumer:0x%" PRIx64 " (second scan) (not enough)", pRebVg->pVgEp->vgId,
             pConsumerEp->consumerId);
     }
   }
@@ -387,12 +387,12 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR
       taosArrayPush(pConsumerEp->vgs, &pRebVg->pVgEp);
       pRebVg->newConsumerId = pConsumerEp->consumerId;
       if (pRebVg->newConsumerId == pRebVg->oldConsumerId) {
-        mInfo("mq rebalance: skip vg %d for same consumer:%" PRIx64 " (second scan)", pRebVg->pVgEp->vgId,
+        mInfo("mq rebalance: skip vg %d for same consumer:0x%" PRIx64 " (second scan)", pRebVg->pVgEp->vgId,
               pConsumerEp->consumerId);
         continue;
       }
       taosArrayPush(pOutput->rebVgs, pRebVg);
-      mInfo("mq rebalance: add vgId:%d to consumer:%" PRIx64 " (second scan) (unassigned)", pRebVg->pVgEp->vgId,
+      mInfo("mq rebalance: add vgId:%d to consumer:0x%" PRIx64 " (second scan) (unassigned)", pRebVg->pVgEp->vgId,
             pConsumerEp->consumerId);
     }
   } else {
@@ -1019,7 +1019,7 @@ int32_t mndRetrieveSubscribe(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock
         pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
         colDataSetVal(pColInfo, numOfRows, (const char *)&pConsumerEp->consumerId, false);
 
-        mDebug("mnd show subscriptions: topic %s, consumer:%" PRIx64 " cgroup %s vgid %d", varDataVal(topic),
+        mDebug("mnd show subscriptions: topic %s, consumer:0x%" PRIx64 " cgroup %s vgid %d", varDataVal(topic),
                pConsumerEp->consumerId, varDataVal(cgroup), pVgEp->vgId);
 
         // offset

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

@@ -79,43 +79,33 @@ typedef struct {
 } STqExecDb;
 
 typedef struct {
-  int8_t subType;
-
-  STqReader*  pExecReader;
-  qTaskInfo_t task;
+  int8_t       subType;
+  STqReader*   pExecReader;
+  qTaskInfo_t  task;
   union {
     STqExecCol execCol;
     STqExecTb  execTb;
     STqExecDb  execDb;
   };
-  int32_t numOfCols;  // number of out pout column, temporarily used
+  int32_t      numOfCols;  // number of out pout column, temporarily used
 } STqExecHandle;
 
 typedef struct {
-  // info
-  char    subKey[TSDB_SUBSCRIBE_KEY_LEN];
-  int64_t consumerId;
-  int32_t epoch;
-  int8_t  fetchMeta;
-
-  int64_t snapshotVer;
-
-  SWalReader* pWalReader;
-
-  SWalRef* pRef;
-
-  // push
-  STqPushHandle pushHandle;
-
-  // exec
-  STqExecHandle execHandle;
-
+  char          subKey[TSDB_SUBSCRIBE_KEY_LEN];
+  int64_t       consumerId;
+  int32_t       epoch;
+  int8_t        fetchMeta;
+  int64_t       snapshotVer;
+  SWalReader*   pWalReader;
+  SWalRef*      pRef;
+  STqPushHandle pushHandle;    // push
+  STqExecHandle execHandle;    // exec
 } STqHandle;
 
 typedef struct {
-  SMqDataRsp     dataRsp;
+  SMqDataRsp*    pDataRsp;
   char           subKey[TSDB_SUBSCRIBE_KEY_LEN];
-  SRpcHandleInfo pInfo;
+  SRpcHandleInfo info;
 } STqPushEntry;
 
 struct STQ {
@@ -151,13 +141,13 @@ int32_t tDecodeSTqHandle(SDecoder* pDecoder, STqHandle* pHandle);
 // tqRead
 int32_t tqScanTaosx(STQ* pTq, const STqHandle* pHandle, STaosxRsp* pRsp, SMqMetaRsp* pMetaRsp, STqOffsetVal* offset);
 int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVal* pOffset);
-int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHead** pHeadWithCkSum);
+int32_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHead** pHeadWithCkSum);
 
 // tqExec
 int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SPackedData submit, STaosxRsp* pRsp);
 // int32_t tqTaosxScanLog(STQ* pTq, STqHandle* pHandle, SSubmitReq* pReq, STaosxRsp* pRsp);
 int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataRsp* pRsp, int32_t numOfCols, int8_t precision);
-int32_t tqSendDataRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataRsp* pRsp);
+int32_t tqSendDataRsp(STQ* pTq, const SRpcMsg* pMsg, const SMqPollReq* pReq, const SMqDataRsp* pRsp, int32_t type);
 int32_t tqPushDataRsp(STQ* pTq, STqPushEntry* pPushEntry);
 
 // tqMeta

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

@@ -192,6 +192,9 @@ void    tqCleanUp();
 STQ*    tqOpen(const char* path, SVnode* pVnode);
 void    tqClose(STQ*);
 int     tqPushMsg(STQ*, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver);
+int     tqRegisterPushEntry(STQ* pTq, void* pHandle, const SMqPollReq* pRequest, SRpcMsg* pRpcMsg, SMqDataRsp* pDataRsp, int32_t type);
+int     tqRemovePushEntry(STQ* pTq, const char* pKey, int32_t keyLen, uint64_t consumerId, bool rspConsumer);
+
 int     tqCommit(STQ*);
 int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd);
 int32_t tqCheckColModifiable(STQ* pTq, int64_t tbUid, int32_t colId);

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

@@ -46,11 +46,13 @@ static int32_t tqAddBlockSchemaToRsp(const STqExecHandle* pExec, STaosxRsp* pRsp
 static int32_t tqAddTbNameToRsp(const STQ* pTq, int64_t uid, STaosxRsp* pRsp, int32_t n) {
   SMetaReader mr = {0};
   metaReaderInit(&mr, pTq->pVnode->pMeta, 0);
+
   // TODO add reference to gurantee success
   if (metaGetTableEntryByUidCache(&mr, uid) < 0) {
     metaReaderClear(&mr);
     return -1;
   }
+
   for (int32_t i = 0; i < n; i++) {
     char* tbName = taosStrdup(mr.me.name);
     taosArrayPush(pRsp->blockTbName, &tbName);
@@ -83,13 +85,16 @@ int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs
   while (1) {
     SSDataBlock* pDataBlock = NULL;
     uint64_t     ts = 0;
+
     tqDebug("vgId:%d, tmq task start to execute", pTq->pVnode->config.vgId);
     if (qExecTask(task, &pDataBlock, &ts) < 0) {
       tqError("vgId:%d, task exec error since %s", pTq->pVnode->config.vgId, terrstr());
       return -1;
     }
+
     tqDebug("vgId:%d, tmq task executed, get %p", pTq->pVnode->config.vgId, pDataBlock);
 
+    // current scan should be stopped asap, since the rebalance occurs.
     if (pDataBlock == NULL) {
       break;
     }
@@ -99,7 +104,9 @@ int32_t tqScanData(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffs
 
     if (pOffset->type == TMQ_OFFSET__SNAPSHOT_DATA) {
       rowCnt += pDataBlock->info.rows;
-      if (rowCnt >= 4096) break;
+      if (rowCnt >= 4096) {
+        break;
+      }
     }
   }
 

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

@@ -283,7 +283,7 @@ int32_t tqMetaRestoreHandle(STQ* pTq) {
   tdbTbcMoveToFirst(pCur);
 
   while (tdbTbcNext(pCur, &pKey, &kLen, &pVal, &vLen) == 0) {
-    STqHandle handle;
+    STqHandle handle = {0};
     tDecoderInit(&decoder, (uint8_t*)pVal, vLen);
     tDecodeSTqHandle(&decoder, &handle);
     tDecoderClear(&decoder);

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

@@ -209,6 +209,7 @@ int32_t tqPushMsgNew(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_
 int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver) {
   void*   pReq = POINTER_SHIFT(msg, sizeof(SSubmitReq2Msg));
   int32_t len = msgLen - sizeof(SSubmitReq2Msg);
+  int32_t vgId = TD_VID(pTq->pVnode);
 
   if (msgType == TDMT_VND_SUBMIT) {
     // lock push mgr to avoid potential msg lost
@@ -217,7 +218,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
     int32_t numOfRegisteredPush = taosHashGetSize(pTq->pPushMgr);
     if (numOfRegisteredPush > 0) {
       tqDebug("vgId:%d tq push msg version:%" PRId64 " type:%s, head:%p, body:%p len:%d, numOfPushed consumers:%d",
-          pTq->pVnode->config.vgId, ver, TMSG_INFO(msgType), msg, pReq, len, numOfRegisteredPush);
+          vgId, ver, TMSG_INFO(msgType), msg, pReq, len, numOfRegisteredPush);
 
       SArray* cachedKeys = taosArrayInit(0, sizeof(void*));
       SArray* cachedKeyLens = taosArrayInit(0, sizeof(size_t));
@@ -239,7 +240,10 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
       void* pIter = NULL;
       while (1) {
         pIter = taosHashIterate(pTq->pPushMgr, pIter);
-        if (pIter == NULL) break;
+        if (pIter == NULL) {
+          break;
+        }
+
         STqPushEntry* pPushEntry = *(STqPushEntry**)pIter;
 
         STqHandle* pHandle = taosHashGet(pTq->pHandle, pPushEntry->subKey, strlen(pPushEntry->subKey));
@@ -248,17 +252,16 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
           continue;
         }
 
-        if (pPushEntry->dataRsp.reqOffset.version >= ver) {
-          tqDebug("vgId:%d, push entry req version %" PRId64 ", while push version %" PRId64 ", skip",
-                  pTq->pVnode->config.vgId, pPushEntry->dataRsp.reqOffset.version, ver);
+        SMqDataRsp* pRsp = pPushEntry->pDataRsp;
+        if (pRsp->reqOffset.version >= ver) {
+          tqDebug("vgId:%d, push entry req version %" PRId64 ", while push version %" PRId64 ", skip", vgId,
+                  pRsp->reqOffset.version, ver);
           continue;
         }
 
         STqExecHandle* pExec = &pHandle->execHandle;
         qTaskInfo_t    task = pExec->task;
 
-        SMqDataRsp* pRsp = &pPushEntry->dataRsp;
-
         // prepare scan mem data
         SPackedData submit = {
             .msgStr = data,
@@ -274,7 +277,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
           SSDataBlock* pDataBlock = NULL;
           uint64_t     ts = 0;
           if (qExecTask(task, &pDataBlock, &ts) < 0) {
-            tqDebug("vgId:%d, tq exec error since %s", pTq->pVnode->config.vgId, terrstr());
+            tqDebug("vgId:%d, tq exec error since %s", vgId, terrstr());
           }
 
           if (pDataBlock == NULL) {
@@ -285,11 +288,11 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
           pRsp->blockNum++;
         }
 
-        tqDebug("vgId:%d, tq handle push, subkey:%s, block num:%d", pTq->pVnode->config.vgId, pPushEntry->subKey,
-                pRsp->blockNum);
+        tqDebug("vgId:%d, tq handle push, subkey:%s, block num:%d", vgId, pPushEntry->subKey, pRsp->blockNum);
         if (pRsp->blockNum > 0) {
           // set offset
           tqOffsetResetToLog(&pRsp->rspOffset, ver);
+
           // remove from hash
           size_t kLen;
           void*  key = taosHashGetKey(pIter, &kLen);
@@ -311,6 +314,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
           tqError("vgId:%d, tq push hash remove key error, key: %s", pTq->pVnode->config.vgId, (char*)key);
         }
       }
+
       taosArrayDestroyP(cachedKeys, (FDelete)taosMemoryFree);
       taosArrayDestroy(cachedKeyLens);
       taosMemoryFree(data);
@@ -336,9 +340,9 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
       };
 
       tqDebug("tq copy write msg %p %d %" PRId64 " from %p", data, len, ver, pReq);
-
       tqProcessSubmitReq(pTq, submit);
     }
+
     if (msgType == TDMT_VND_DELETE) {
       tqProcessDelReq(pTq, POINTER_SHIFT(msg, sizeof(SMsgHead)), msgLen - sizeof(SMsgHead), ver);
     }
@@ -346,3 +350,61 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
 
   return 0;
 }
+
+int32_t tqRegisterPushEntry(STQ* pTq, void* pHandle, const SMqPollReq* pRequest, SRpcMsg* pRpcMsg,
+                            SMqDataRsp* pDataRsp, int32_t type) {
+  uint64_t   consumerId = pRequest->consumerId;
+  int32_t    vgId = TD_VID(pTq->pVnode);
+  STqHandle* pTqHandle = pHandle;
+
+  STqPushEntry* pPushEntry = taosMemoryCalloc(1, sizeof(STqPushEntry));
+  if (pPushEntry == NULL) {
+    tqDebug("tmq poll: consumer:0x%" PRIx64 ", vgId:%d failed to malloc, size:%d", consumerId, vgId,
+            (int32_t)sizeof(STqPushEntry));
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+
+  pPushEntry->info = pRpcMsg->info;
+  memcpy(pPushEntry->subKey, pTqHandle->subKey, TSDB_SUBSCRIBE_KEY_LEN);
+
+  if (type == TMQ_MSG_TYPE__TAOSX_RSP) {
+    pPushEntry->pDataRsp =  taosMemoryCalloc(1, sizeof(STaosxRsp));
+    memcpy(pPushEntry->pDataRsp, pDataRsp, sizeof(STaosxRsp));
+  } else if (type == TMQ_MSG_TYPE__POLL_RSP) {
+    pPushEntry->pDataRsp = taosMemoryCalloc(1, sizeof(SMqDataRsp));
+    memcpy(pPushEntry->pDataRsp, pDataRsp, sizeof(SMqDataRsp));
+  }
+
+  SMqRspHead* pHead = &pPushEntry->pDataRsp->head;
+  pHead->consumerId = consumerId;
+  pHead->epoch = pRequest->epoch;
+  pHead->mqMsgType = type;
+
+  taosHashPut(pTq->pPushMgr, pTqHandle->subKey, strlen(pTqHandle->subKey), &pPushEntry, sizeof(void*));
+
+  tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s offset:%" PRId64 ", vgId:%d save handle to push mgr, total:%d", consumerId,
+          pTqHandle->subKey, pDataRsp->reqOffset.version, vgId, taosHashGetSize(pTq->pPushMgr));
+  return 0;
+}
+
+int32_t tqRemovePushEntry(STQ* pTq, const char* pKey, int32_t keyLen, uint64_t consumerId, bool rspConsumer) {
+  int32_t       vgId = TD_VID(pTq->pVnode);
+  STqPushEntry** pEntry = taosHashGet(pTq->pPushMgr, pKey, keyLen);
+
+  if (pEntry != NULL) {
+    uint64_t cId = (*pEntry)->pDataRsp->head.consumerId;
+    ASSERT(consumerId == cId);
+
+    tqDebug("tmq poll: consumer:0x%" PRIx64 ", subkey %s vgId:%d remove from push mgr, remains:%d", consumerId,
+            (*pEntry)->subKey, vgId, taosHashGetSize(pTq->pPushMgr) - 1);
+
+    if (rspConsumer) { // rsp the old consumer with empty block.
+      tqPushDataRsp(pTq, *pEntry);
+    }
+
+    taosHashRemove(pTq->pPushMgr, pKey, keyLen);
+  }
+
+  return 0;
+}

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

@@ -183,14 +183,15 @@ end:
   return tbSuid == realTbSuid;
 }
 
-int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHead** ppCkHead) {
+int32_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHead** ppCkHead) {
   int32_t code = 0;
+
   taosThreadMutexLock(&pHandle->pWalReader->mutex);
   int64_t offset = *fetchOffset;
 
   while (1) {
     if (walFetchHead(pHandle->pWalReader, offset, *ppCkHead) < 0) {
-      tqDebug("tmq poll: consumer:%" PRId64 ", (epoch %d) vgId:%d offset %" PRId64 ", no more log to return",
+      tqDebug("tmq poll: consumer:0x%" PRIx64 ", (epoch %d) vgId:%d offset %" PRId64 ", no more log to return",
               pHandle->consumerId, pHandle->epoch, TD_VID(pTq->pVnode), offset);
       *fetchOffset = offset - 1;
       code = -1;
@@ -241,6 +242,7 @@ int64_t tqFetchLog(STQ* pTq, STqHandle* pHandle, int64_t* fetchOffset, SWalCkHea
       offset++;
     }
   }
+
 END:
   taosThreadMutexUnlock(&pHandle->pWalReader->mutex);
   return code;

source/libs/command/src/command.c

@@ -307,7 +307,7 @@ static void setCreateDBResultIntoDataBlock(SSDataBlock* pBlock, char* dbName, ch
 bool existLeaderRole(TAOS_ROW row, TAOS_FIELD* fields, int nFields) {
   // vgroup_id | db_name | tables | v1_dnode | v1_status | v2_dnode | v2_status | v3_dnode | v3_status | v4_dnode |
   // v4_status |  cacheload  | tsma |
-  if (nFields != 13) {
+  if (nFields != 14) {
     return false;
   }
 

source/libs/executor/src/executor.c

@@ -221,12 +221,12 @@ int32_t qSetSMAInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks,
 qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols, SSchemaWrapper** pSchema) {
   if (msg == NULL) {
     // create raw scan
-
     SExecTaskInfo* pTaskInfo = taosMemoryCalloc(1, sizeof(SExecTaskInfo));
     if (NULL == pTaskInfo) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       return NULL;
     }
+
     setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
 
     pTaskInfo->cost.created = taosGetTimestampUs();
@@ -715,7 +715,6 @@ void qStopTaskOperators(SExecTaskInfo* pTaskInfo) {
 
 int32_t qAsyncKillTask(qTaskInfo_t qinfo, int32_t rspCode) {
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qinfo;
-
   if (pTaskInfo == NULL) {
     return TSDB_CODE_QRY_INVALID_QHANDLE;
   }
@@ -723,7 +722,6 @@ int32_t qAsyncKillTask(qTaskInfo_t qinfo, int32_t rspCode) {
   qDebug("%s execTask async killed", GET_TASKID(pTaskInfo));
 
   setTaskKilled(pTaskInfo, rspCode);
-
   qStopTaskOperators(pTaskInfo);
 
   return TSDB_CODE_SUCCESS;

source/libs/executor/src/executorimpl.c

@@ -2774,11 +2774,17 @@ int32_t buildSessionResultDataBlock(SOperatorInfo* pOperator, SStreamState* pSta
 }
 
 void qStreamCloseTsdbReader(void* task) {
-  if (task == NULL) return;
+  if (task == NULL) {
+    return;
+  }
+
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)task;
   SOperatorInfo* pOp = pTaskInfo->pRoot;
-  qDebug("stream close tsdb reader, reset status uid %" PRId64 " ts %" PRId64, pTaskInfo->streamInfo.lastStatus.uid,
+
+  qDebug("stream close tsdb reader, reset status uid:%" PRId64 " ts:%" PRId64, pTaskInfo->streamInfo.lastStatus.uid,
          pTaskInfo->streamInfo.lastStatus.ts);
+
+  // todo refactor, other thread may already use this read to extract data.
   pTaskInfo->streamInfo.lastStatus = (STqOffsetVal){0};
   while (pOp->numOfDownstream == 1 && pOp->pDownstream[0]) {
     SOperatorInfo* pDownstreamOp = pOp->pDownstream[0];
@@ -2786,8 +2792,19 @@ void qStreamCloseTsdbReader(void* task) {
       SStreamScanInfo* pInfo = pDownstreamOp->info;
       if (pInfo->pTableScanOp) {
         STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
+
+        setOperatorCompleted(pInfo->pTableScanOp);
+        while(pTaskInfo->owner != 0) {
+          taosMsleep(100);
+          qDebug("wait for the reader stopping");
+        }
+
         tsdbReaderClose(pTSInfo->base.dataReader);
         pTSInfo->base.dataReader = NULL;
+
+        // restore the status, todo refactor.
+        pInfo->pTableScanOp->status = OP_OPENED;
+        pTaskInfo->status = TASK_NOT_COMPLETED;
         return;
       }
     }

source/libs/executor/src/scanoperator.c

@@ -751,7 +751,7 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) {
 
     while (1) {
       SSDataBlock* result = doGroupedTableScan(pOperator);
-      if (result) {
+      if (result || (pOperator->status == OP_EXEC_DONE)) {
         return result;
       }
 
@@ -985,6 +985,7 @@ void resetTableScanInfo(STableScanInfo* pTableScanInfo, STimeWindow* pWin) {
   pTableScanInfo->scanTimes = 0;
   pTableScanInfo->currentGroupId = -1;
   tsdbReaderClose(pTableScanInfo->base.dataReader);
+  qDebug("1");
   pTableScanInfo->base.dataReader = NULL;
 }
 
@@ -1143,6 +1144,7 @@ static SSDataBlock* doRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32
       pInfo->updateWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
       STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info;
       tsdbReaderClose(pTableScanInfo->base.dataReader);
+      qDebug("2");
       pTableScanInfo->base.dataReader = NULL;
       return NULL;
     }
@@ -1616,6 +1618,7 @@ static SSDataBlock* doQueueScan(SOperatorInfo* pOperator) {
       if (!pTaskInfo->streamInfo.returned) {
         STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
         tsdbReaderClose(pTSInfo->base.dataReader);
+        qDebug("3");
         pTSInfo->base.dataReader = NULL;
         tqOffsetResetToLog(&pTaskInfo->streamInfo.prepareStatus, pTaskInfo->streamInfo.snapshotVer);
         qDebug("queue scan tsdb over, switch to wal ver %" PRId64 "", pTaskInfo->streamInfo.snapshotVer + 1);
@@ -1767,6 +1770,8 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
 
     /*resetTableScanInfo(pTSInfo, pWin);*/
     tsdbReaderClose(pTSInfo->base.dataReader);
+    qDebug("4");
+
     pTSInfo->base.dataReader = NULL;
     pInfo->pTableScanOp->status = OP_OPENED;
 
@@ -1837,6 +1842,8 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
     pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__NONE;
     STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
     tsdbReaderClose(pTSInfo->base.dataReader);
+    qDebug("5");
+
     pTSInfo->base.dataReader = NULL;
 
     pTSInfo->base.cond.startVersion = -1;
@@ -2635,6 +2642,8 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
     return pBlock;
   }
 
+  qDebug("8");
+
   tsdbReaderClose(pInfo->base.dataReader);
   pInfo->base.dataReader = NULL;
   return NULL;

tests/system-test/6-cluster/5dnode1mnode.py

@@ -29,7 +29,7 @@ class TDTestCase:
         self.master_dnode = self.TDDnodes.dnodes[0]
         self.host=self.master_dnode.cfgDict["fqdn"]
         conn1 = taos.connect(self.master_dnode.cfgDict["fqdn"] , config=self.master_dnode.cfgDir)
-        tdSql.init(conn1.cursor())
+        tdSql.init(conn1.cursor(), True)
 
 
     def getBuildPath(self):

tests/system-test/7-tmq/subscribeDb1.py

@@ -123,8 +123,9 @@ class TDTestCase:
         pre_insert = "insert into "
         sql = pre_insert
 
-        t = time.time()
-        startTs = int(round(t * 1000))
+        # t = 1678609778776 #time.time()
+        t = 1600000000000
+        startTs = t #int(round(t * 1000))
         #tdLog.debug("doing insert data into stable:%s rows:%d ..."%(stbName, allRows))
         for i in range(ctbNum):
             sql += " %s_%d values "%(stbName,i)

utils/test/c/tmqSim.c

@@ -386,7 +386,7 @@ void addRowsToVgroupId(SThreadInfo* pInfo, int32_t vgroupId, int32_t rows) {
   pInfo->rowsOfPerVgroups[pInfo->numOfVgroups][1] += rows;
   pInfo->numOfVgroups++;
 
-  taosFprintfFile(g_fp, "consume id %d, add one new vogroup id: %d\n", pInfo->consumerId, vgroupId);
+  taosFprintfFile(g_fp, "consume id %d, add new vgroupId:%d\n", pInfo->consumerId, vgroupId);
   if (pInfo->numOfVgroups > MAX_VGROUP_CNT) {
     taosFprintfFile(g_fp, "====consume id %d, vgroup num %d over than 32. new vgroupId: %d\n", pInfo->consumerId,
                     pInfo->numOfVgroups, vgroupId);
@@ -578,18 +578,25 @@ static int32_t data_msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t msgIn
   char    buf[1024];
   int32_t totalRows = 0;
 
-  // printf("topic: %s\n", tmq_get_topic_name(msg));
   int32_t     vgroupId = tmq_get_vgroup_id(msg);
   const char* dbName = tmq_get_db_name(msg);
 
   taosFprintfFile(g_fp, "consumerId: %d, msg index:%d\n", pInfo->consumerId, msgIndex);
-  taosFprintfFile(g_fp, "dbName: %s, topic: %s, vgroupId: %d\n", dbName != NULL ? dbName : "invalid table",
-                  tmq_get_topic_name(msg), vgroupId);
+  int32_t index = 0;
+  for (index = 0; index < pInfo->numOfVgroups; index++) {
+    if (vgroupId == pInfo->rowsOfPerVgroups[index][0]) {
+      break;
+    }
+  }
+
+  taosFprintfFile(g_fp, "dbName: %s, topic: %s, vgroupId:%d, currentRows:%d\n", dbName != NULL ? dbName : "invalid table",
+                  tmq_get_topic_name(msg), vgroupId, pInfo->rowsOfPerVgroups[index][1]);
 
   while (1) {
     TAOS_ROW row = taos_fetch_row(msg);
-
-    if (row == NULL) break;
+    if (row == NULL) {
+      break;
+    }
 
     TAOS_FIELD* fields = taos_fetch_fields(msg);
     int32_t     numOfFields = taos_field_count(msg);
@@ -607,7 +614,6 @@ static int32_t data_msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t msgIn
 #endif
 
     dumpToFileForCheck(pInfo->pConsumeRowsFile, row, fields, length, numOfFields, precision);
-
     taos_print_row(buf, row, fields, numOfFields);
 
     if (0 != g_stConfInfo.showRowFlag) {
@@ -621,7 +627,6 @@ static int32_t data_msg_process(TAOS_RES* msg, SThreadInfo* pInfo, int32_t msgIn
   }
 
   addRowsToVgroupId(pInfo, vgroupId, totalRows);
-
   return totalRows;
 }
 
@@ -730,9 +735,7 @@ void build_consumer(SThreadInfo* pInfo) {
   }
 
   pInfo->tmq = tmq_consumer_new(conf, NULL, 0);
-
   tmq_conf_destroy(conf);
-
   return;
 }
 
@@ -817,7 +820,6 @@ void loop_consume(SThreadInfo* pInfo) {
       }
 
       taos_free_result(tmqMsg);
-
       totalMsgs++;
 
       int64_t currentPrintTime = taosGetTimestampMs();