Merge pull request #17800 from taosdata/feature/stream

refactor(stream): fill history

include/common/tmsg.h

@@ -57,7 +57,8 @@ extern int32_t tMsgDict[];
 #define TMSG_SEG_SEQ(TYPE)  ((TYPE)&0xff)
 #define TMSG_INFO(TYPE)                                                                                                \
   ((TYPE) < TDMT_DND_MAX_MSG || (TYPE) < TDMT_MND_MAX_MSG || (TYPE) < TDMT_VND_MAX_MSG || (TYPE) < TDMT_SCH_MAX_MSG || \
-   (TYPE) < TDMT_STREAM_MAX_MSG || (TYPE) < TDMT_MON_MAX_MSG || (TYPE) < TDMT_SYNC_MAX_MSG)                            \
+   (TYPE) < TDMT_VND_TMQ_MAX_MSG || (TYPE) < TDMT_STREAM_MAX_MSG || (TYPE) < TDMT_VND_STREAM_MAX_MSG ||                \
+   (TYPE) < TDMT_MON_MAX_MSG || (TYPE) < TDMT_SYNC_MAX_MSG)                                                            \
       ? tMsgInfo[tMsgDict[TMSG_SEG_CODE(TYPE)] + TMSG_SEG_SEQ(TYPE)]                                                   \
       : 0
 #define TMSG_INDEX(TYPE) (tMsgDict[TMSG_SEG_CODE(TYPE)] + TMSG_SEG_SEQ(TYPE))

include/common/tmsgdef.h

@@ -138,17 +138,15 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_GET_TABLE_INDEX, "get-table-index", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_BATCH_META, "batch-meta", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_TABLE_CFG, "table-cfg", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_CREATE_TOPIC, "create-topic", SMCreateTopicReq, SMCreateTopicRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_ALTER_TOPIC, "alter-topic", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_DROP_TOPIC, "drop-topic", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_SUBSCRIBE, "subscribe", SCMSubscribeReq, SCMSubscribeRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_ASK_EP, "ask-ep", SMqAskEpReq, SMqAskEpRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_LOST, "consumer-lost", SMqConsumerLostMsg, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_RECOVER, "consumer-recover", SMqConsumerRecoverMsg, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_HB, "consumer-hb", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_DO_REBALANCE, "do-rebalance", SMqDoRebalanceMsg, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_DROP_CGROUP, "drop-cgroup", SMqDropCGroupReq, SMqDropCGroupRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_COMMIT_OFFSET, "mnode-commit-offset", SMqCMCommitOffsetReq, SMqCMCommitOffsetRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_TMQ_CREATE_TOPIC, "create-topic", SMCreateTopicReq, SMCreateTopicRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_TMQ_DROP_TOPIC, "drop-topic", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_TMQ_SUBSCRIBE, "subscribe", SCMSubscribeReq, SCMSubscribeRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_TMQ_ASK_EP, "ask-ep", SMqAskEpReq, SMqAskEpRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_TMQ_CONSUMER_LOST, "consumer-lost", SMqConsumerLostMsg, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_TMQ_CONSUMER_RECOVER, "consumer-recover", SMqConsumerRecoverMsg, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_TMQ_HB, "consumer-hb", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_TMQ_DO_REBALANCE, "do-rebalance", SMqDoRebalanceMsg, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_TMQ_DROP_CGROUP, "drop-cgroup", SMqDropCGroupReq, SMqDropCGroupRsp)
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mq-tmr", SMTimerReq, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_TELEM_TIMER, "telem-tmr", SMTimerReq, SMTimerReq)
   TD_DEF_MSG_TYPE(TDMT_MND_TRANS_TIMER, "trans-tmr", NULL, NULL)
@@ -186,21 +184,6 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_CREATE_STB, "vnode-create-stb", SVCreateStbReq, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_ALTER_STB, "vnode-alter-stb", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_DROP_STB, "vnode-drop-stb", SVDropStbReq, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_MQ_VG_CHANGE, "vnode-mq-vg-change", SMqRebVgReq, SMqRebVgRsp)
-  TD_DEF_MSG_TYPE(TDMT_VND_MQ_VG_DELETE, "vnode-mq-vg-delete", SMqVDeleteReq, SMqVDeleteRsp)
-  TD_DEF_MSG_TYPE(TDMT_VND_MQ_COMMIT_OFFSET, "vnode-commit-offset", STqOffset, STqOffset)
-  TD_DEF_MSG_TYPE(TDMT_VND_ADD_CHECK_INFO, "vnode-add-check-info", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_DELETE_CHECK_INFO, "vnode-delete-check-info", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_CREATE_TOPIC, "vnode-create-topic", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_ALTER_TOPIC, "vnode-alter-topic", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_DROP_TOPIC, "vnode-drop-topic", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_SUBSCRIBE, "vnode-subscribe", SMVSubscribeReq, SMVSubscribeRsp)
-  TD_DEF_MSG_TYPE(TDMT_VND_CONSUME, "vnode-consume", SMqPollReq, SMqDataBlkRsp)
-  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_TRIGGER, "vnode-stream-trigger", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_DISPATCH_WRITE, "vnode-stream-task-dispatch-write", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_RECOVER_STEP1, "vnode-stream-recover1", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_RECOVER_STEP2, "vnode-stream-recover2", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_RECOVER_FINISH, "vnode-stream-finish", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_CREATE_SMA, "vnode-create-sma", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_CANCEL_SMA, "vnode-cancel-sma", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_DROP_SMA, "vnode-drop-sma", NULL, NULL)
@@ -232,15 +215,30 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_SCH_LINK_BROKEN, "link-broken", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_SCH_MAX_MSG, "sch-max", NULL, NULL)
 
+  TD_NEW_MSG_SEG(TDMT_VND_TMQ_MSG)
+  TD_DEF_MSG_TYPE(TDMT_VND_TMQ_SUBSCRIBE, "vnode-tmq-subscribe", SMqRebVgReq, SMqRebVgRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_TMQ_DELETE_SUB, "vnode-tmq-delete-sub", SMqVDeleteReq, SMqVDeleteRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_TMQ_COMMIT_OFFSET, "vnode-tmq-commit-offset", STqOffset, STqOffset)
+  TD_DEF_MSG_TYPE(TDMT_VND_TMQ_ADD_CHECKINFO, "vnode-tmq-add-checkinfo", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_TMQ_DEL_CHECKINFO, "vnode-del-checkinfo", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_TMQ_CONSUME, "vnode-tmq-consume", SMqPollReq, SMqDataBlkRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_TMQ_MAX_MSG, "vnd-tmq-max", NULL, NULL)
+
   TD_NEW_MSG_SEG(TDMT_STREAM_MSG)
   TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_DEPLOY, "stream-task-deploy", SStreamTaskDeployReq, SStreamTaskDeployRsp)
   TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_DROP, "stream-task-drop", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_RUN, "stream-task-run", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_DISPATCH, "stream-task-dispatch", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_STREAM_TASK_RECOVER, "stream-task-recover", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_RETRIEVE, "stream-retrieve", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_STREAM_RECOVER_FINISH, "vnode-stream-finish", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_STREAM_MAX_MSG, "stream-max", NULL, NULL)
 
+  TD_NEW_MSG_SEG(TDMT_VND_STREAM_MSG)
+  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_TRIGGER, "vnode-stream-trigger", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_RECOVER_STEP1, "vnode-stream-recover1", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_RECOVER_STEP2, "vnode-stream-recover2", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_MAX_MSG, "vnd-stream-max", NULL, NULL)
+
   TD_NEW_MSG_SEG(TDMT_MON_MSG)
   TD_DEF_MSG_TYPE(TDMT_MON_MAX_MSG, "monitor-max", NULL, NULL)
 

include/common/tname.h

@@ -73,11 +73,11 @@ bool tNameTbNameEqual(SName* left, SName* right);
 typedef struct {
   // input
   SArray*     tags;            // element is SSmlKv
-  const char* sTableName;     // super table name
-  uint8_t     sTableNameLen;  // the length of super table name
+  const char* stbFullName;     // super table name
+  uint8_t     stbFullNameLen;  // the length of super table name
 
   // output
-  char*    childTableName;  // must have size of TSDB_TABLE_NAME_LEN;
+  char*    ctbShortName;  // must have size of TSDB_TABLE_NAME_LEN;
   uint64_t uid;           // child table uid, may be useful
 } RandTableName;
 

include/libs/executor/executor.h

@@ -79,6 +79,7 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, SReadHandle* readers);
  */
 qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols, SSchemaWrapper** pSchema);
 
+int32_t qSetStreamOpOpen(qTaskInfo_t tinfo);
 /**
  * Set multiple input data blocks for the stream scan.
  * @param tinfo

include/libs/stream/tstream.h

@@ -441,6 +441,7 @@ typedef struct {
 } SStreamRetrieveRsp;
 
 typedef struct {
+  SMsgHead msgHead;
   int64_t  streamId;
   int32_t  taskId;
 } SStreamRecoverStep1Req, SStreamRecoverStep2Req;

source/client/src/clientTmq.c

@@ -508,7 +508,7 @@ static int32_t tmqSendCommitReq(tmq_t* tmq, SMqClientVg* pVg, SMqClientTopic* pT
   pMsgSendInfo->param = pParam;
   pMsgSendInfo->paramFreeFp = taosMemoryFree;
   pMsgSendInfo->fp = tmqCommitCb;
-  pMsgSendInfo->msgType = TDMT_VND_MQ_COMMIT_OFFSET;
+  pMsgSendInfo->msgType = TDMT_VND_TMQ_COMMIT_OFFSET;
   // send msg
 
   atomic_add_fetch_32(&pParamSet->waitingRspNum, 1);
@@ -750,7 +750,7 @@ void tmqSendHbReq(void* param, void* tmrId) {
   sendInfo->requestObjRefId = 0;
   sendInfo->param = NULL;
   sendInfo->fp = tmqHbCb;
-  sendInfo->msgType = TDMT_MND_MQ_HB;
+  sendInfo->msgType = TDMT_MND_TMQ_HB;
 
   SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);
 
@@ -1038,7 +1038,7 @@ int32_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
   sendInfo->requestObjRefId = 0;
   sendInfo->param = &param;
   sendInfo->fp = tmqSubscribeCb;
-  sendInfo->msgType = TDMT_MND_SUBSCRIBE;
+  sendInfo->msgType = TDMT_MND_TMQ_SUBSCRIBE;
 
   SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);
 
@@ -1420,7 +1420,7 @@ int32_t tmqAskEp(tmq_t* tmq, bool async) {
   sendInfo->requestObjRefId = 0;
   sendInfo->param = pParam;
   sendInfo->fp = tmqAskEpCb;
-  sendInfo->msgType = TDMT_MND_MQ_ASK_EP;
+  sendInfo->msgType = TDMT_MND_TMQ_ASK_EP;
 
   SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);
 
@@ -1573,7 +1573,7 @@ int32_t tmqPollImpl(tmq_t* tmq, int64_t timeout) {
       sendInfo->requestObjRefId = 0;
       sendInfo->param = pParam;
       sendInfo->fp = tmqPollCb;
-      sendInfo->msgType = TDMT_VND_CONSUME;
+      sendInfo->msgType = TDMT_VND_TMQ_CONSUME;
 
       int64_t transporterId = 0;
       /*printf("send poll\n");*/

source/common/src/tdatablock.c

@@ -2068,8 +2068,8 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataB
           default:
             if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
               if (colDataIsNull_s(pColInfoData, j)) {
-                tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type, TD_VTYPE_NULL, NULL, false, offset,
-                                    k);
+                tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type, TD_VTYPE_NULL, NULL, false,
+                                    offset, k);
               } else if (pCol->type == pColInfoData->info.type) {
                 tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type, TD_VTYPE_NORM, var, true, offset,
                                     k);
@@ -2142,10 +2142,26 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataB
   return TSDB_CODE_SUCCESS;
 }
 
-char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId) {
-  ASSERT(stbName[0] != 0);
+char* buildCtbNameByGroupId(const char* stbFullName, uint64_t groupId) {
+  ASSERT(stbFullName[0] != 0);
   SArray* tags = taosArrayInit(0, sizeof(void*));
+  if (tags == NULL) {
+    return NULL;
+  }
+
   SSmlKv* pTag = taosMemoryCalloc(1, sizeof(SSmlKv));
+  if (pTag == NULL) {
+    taosArrayDestroy(tags);
+    return NULL;
+  }
+
+  void* cname = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN + 1);
+  if (cname == NULL) {
+    taosArrayDestroy(tags);
+    taosMemoryFree(pTag);
+    return NULL;
+  }
+
   pTag->key = "group_id";
   pTag->keyLen = strlen(pTag->key);
   pTag->type = TSDB_DATA_TYPE_UBIGINT;
@@ -2153,13 +2169,11 @@ char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId) {
   pTag->length = sizeof(uint64_t);
   taosArrayPush(tags, &pTag);
 
-  void* cname = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN + 1);
-
   RandTableName rname = {
       .tags = tags,
-      .sTableName = stbName,
-      .sTableNameLen = strlen(stbName),
-      .childTableName = cname,
+      .stbFullName = stbFullName,
+      .stbFullNameLen = strlen(stbFullName),
+      .ctbShortName = cname,
   };
 
   buildChildTableName(&rname);
@@ -2167,8 +2181,8 @@ char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId) {
   taosMemoryFree(pTag);
   taosArrayDestroy(tags);
 
-  ASSERT(rname.childTableName && rname.childTableName[0]);
-  return rname.childTableName;
+  ASSERT(rname.ctbShortName && rname.ctbShortName[0]);
+  return rname.ctbShortName;
 }
 
 void blockEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_t numOfCols, int8_t needCompress) {

source/common/src/tname.c

@@ -315,7 +315,7 @@ static int compareKv(const void* p1, const void* p2) {
  */
 void buildChildTableName(RandTableName* rName) {
   SStringBuilder sb = {0};
-  taosStringBuilderAppendStringLen(&sb, rName->sTableName, rName->sTableNameLen);
+  taosStringBuilderAppendStringLen(&sb, rName->stbFullName, rName->stbFullNameLen);
   taosArraySort(rName->tags, compareKv);
   for (int j = 0; j < taosArrayGetSize(rName->tags); ++j) {
     taosStringBuilderAppendChar(&sb, ',');
@@ -336,11 +336,11 @@ void buildChildTableName(RandTableName* rName) {
   tMD5Final(&context);
 
   char temp[8] = {0};
-  rName->childTableName[0] = 't';
-  rName->childTableName[1] = '_';
+  rName->ctbShortName[0] = 't';
+  rName->ctbShortName[1] = '_';
   for (int i = 0; i < 16; i++) {
     sprintf(temp, "%02x", context.digest[i]);
-    strcat(rName->childTableName, temp);
+    strcat(rName->ctbShortName, temp);
   }
   taosStringBuilderDestroy(&sb);
   rName->uid = *(uint64_t*)(context.digest);

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

@@ -141,15 +141,13 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_STREAM, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_GET_INDEX, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_GET_TABLE_INDEX, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_MND_CREATE_TOPIC, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_MND_ALTER_TOPIC, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_MND_DROP_TOPIC, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_MND_SUBSCRIBE, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_MND_MQ_ASK_EP, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_MND_MQ_HB, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_MND_MQ_DROP_CGROUP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_MND_MQ_DROP_CGROUP_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_MND_MQ_COMMIT_OFFSET, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_TMQ_CREATE_TOPIC, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_TMQ_DROP_TOPIC, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_TMQ_SUBSCRIBE, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_TMQ_ASK_EP, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_TMQ_HB, mmPutMsgToReadQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_TMQ_DROP_CGROUP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_MND_TMQ_DROP_CGROUP_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_KILL_TRANS, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_KILL_QUERY, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_MND_KILL_CONN, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
@@ -171,10 +169,10 @@ SArray *mmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_STB_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_SMA_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_SMA_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_CHANGE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_DELETE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_ADD_CHECK_INFO_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_DELETE_CHECK_INFO_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_SUBSCRIBE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_DELETE_SUB_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_ADD_CHECKINFO_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_DEL_CHECKINFO_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SCH_DROP_TASK, mmPutMsgToFetchQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DEPLOY_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DROP_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

source/dnode/mgmt/mgmt_snode/src/smHandle.c

@@ -72,8 +72,6 @@ SArray *smGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RUN, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH_RSP, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RECOVER, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RECOVER_RSP, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE_RSP, smPutNodeMsgToStreamQueue, 1) == NULL) goto _OVER;
 

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

@@ -392,12 +392,12 @@ SArray *vmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_VND_CANCEL_SMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_SMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_SUBMIT_RSMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_CHANGE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_COMMIT_OFFSET, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_ADD_CHECK_INFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_DELETE_CHECK_INFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_VND_CONSUME, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_SUBSCRIBE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_DELETE_SUB, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_COMMIT_OFFSET, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_ADD_CHECKINFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_DEL_CHECKINFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_CONSUME, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_BATCH_DEL, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_COMMIT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
@@ -409,10 +409,9 @@ SArray *vmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RUN, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RECOVER, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
-  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RECOVER_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_STREAM_RECOVER_FINISH, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
 
   if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;

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

@@ -63,12 +63,12 @@ int32_t mndInitConsumer(SMnode *pMnode) {
       .deleteFp = (SdbDeleteFp)mndConsumerActionDelete,
   };
 
-  mndSetMsgHandle(pMnode, TDMT_MND_SUBSCRIBE, mndProcessSubscribeReq);
-  mndSetMsgHandle(pMnode, TDMT_MND_MQ_HB, mndProcessMqHbReq);
-  mndSetMsgHandle(pMnode, TDMT_MND_MQ_ASK_EP, mndProcessAskEpReq);
+  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_SUBSCRIBE, mndProcessSubscribeReq);
+  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_HB, mndProcessMqHbReq);
+  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_ASK_EP, mndProcessAskEpReq);
   mndSetMsgHandle(pMnode, TDMT_MND_MQ_TIMER, mndProcessMqTimerMsg);
-  mndSetMsgHandle(pMnode, TDMT_MND_MQ_CONSUMER_LOST, mndProcessConsumerLostMsg);
-  mndSetMsgHandle(pMnode, TDMT_MND_MQ_CONSUMER_RECOVER, mndProcessConsumerRecoverMsg);
+  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CONSUMER_LOST, mndProcessConsumerLostMsg);
+  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CONSUMER_RECOVER, mndProcessConsumerRecoverMsg);
 
   mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_CONSUMERS, mndRetrieveConsumer);
   mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_CONSUMERS, mndCancelGetNextConsumer);
@@ -207,7 +207,7 @@ static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg) {
 
       pLostMsg->consumerId = pConsumer->consumerId;
       SRpcMsg pRpcMsg = {
-          .msgType = TDMT_MND_MQ_CONSUMER_LOST,
+          .msgType = TDMT_MND_TMQ_CONSUMER_LOST,
           .pCont = pLostMsg,
           .contLen = sizeof(SMqConsumerLostMsg),
       };
@@ -256,7 +256,7 @@ static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg) {
   if (taosHashGetSize(pRebMsg->rebSubHash) != 0) {
     mInfo("mq rebalance will be triggered");
     SRpcMsg rpcMsg = {
-        .msgType = TDMT_MND_MQ_DO_REBALANCE,
+        .msgType = TDMT_MND_TMQ_DO_REBALANCE,
         .pCont = pRebMsg,
         .contLen = sizeof(SMqDoRebalanceMsg),
     };
@@ -292,7 +292,7 @@ static int32_t mndProcessMqHbReq(SRpcMsg *pMsg) {
 
     pRecoverMsg->consumerId = consumerId;
     SRpcMsg pRpcMsg = {
-        .msgType = TDMT_MND_MQ_CONSUMER_RECOVER,
+        .msgType = TDMT_MND_TMQ_CONSUMER_RECOVER,
         .pCont = pRecoverMsg,
         .contLen = sizeof(SMqConsumerRecoverMsg),
     };
@@ -331,7 +331,7 @@ static int32_t mndProcessAskEpReq(SRpcMsg *pMsg) {
 
     pRecoverMsg->consumerId = consumerId;
     SRpcMsg pRpcMsg = {
-        .msgType = TDMT_MND_MQ_CONSUMER_RECOVER,
+        .msgType = TDMT_MND_TMQ_CONSUMER_RECOVER,
         .pCont = pRecoverMsg,
         .contLen = sizeof(SMqConsumerRecoverMsg),
     };

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

@@ -487,6 +487,7 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
         qDestroyQueryPlan(pPlan);
         return -1;
       }
+      pTask->fillHistory = pStream->fillHistory;
       mndAddTaskToTaskSet(taskOneLevel, pTask);
 
       // source

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

@@ -557,78 +557,6 @@ static int32_t mndPersistTaskDropReq(STrans *pTrans, SStreamTask *pTask) {
   return 0;
 }
 
-#if 0
-static int32_t mndPersistTaskRecoverReq(STrans *pTrans, SStreamTask *pTask) {
-  SMStreamTaskRecoverReq *pReq = taosMemoryCalloc(1, sizeof(SMStreamTaskRecoverReq));
-  if (pReq == NULL) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return -1;
-  }
-  pReq->streamId = pTask->streamId;
-  pReq->taskId = pTask->taskId;
-  int32_t len;
-  int32_t code;
-  tEncodeSize(tEncodeSMStreamTaskRecoverReq, pReq, len, code);
-  if (code != 0) {
-    return -1;
-  }
-  void *buf = taosMemoryCalloc(1, sizeof(SMsgHead) + len);
-  if (buf == NULL) {
-    return -1;
-  }
-  void    *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
-  SEncoder encoder;
-  tEncoderInit(&encoder, abuf, len);
-  tEncodeSMStreamTaskRecoverReq(&encoder, pReq);
-  ((SMsgHead *)buf)->vgId = pTask->nodeId;
-
-  STransAction action = {0};
-  memcpy(&action.epSet, &pTask->epSet, sizeof(SEpSet));
-  action.pCont = buf;
-  action.contLen = sizeof(SMsgHead) + len;
-  action.msgType = TDMT_STREAM_TASK_RECOVER;
-  if (mndTransAppendRedoAction(pTrans, &action) != 0) {
-    taosMemoryFree(buf);
-    return -1;
-  }
-  return 0;
-}
-
-int32_t mndRecoverStreamTasks(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream) {
-  if (pStream->isDistributed) {
-    int32_t lv = taosArrayGetSize(pStream->tasks);
-    for (int32_t i = 0; i < lv; i++) {
-      SArray      *pTasks = taosArrayGetP(pStream->tasks, i);
-      int32_t      sz = taosArrayGetSize(pTasks);
-      SStreamTask *pTask = taosArrayGetP(pTasks, 0);
-      if (pTask->taskLevel == TASK_LEVEL__AGG) {
-        ASSERT(sz == 1);
-        if (mndPersistTaskRecoverReq(pTrans, pTask) < 0) {
-          return -1;
-        }
-      } else {
-        continue;
-      }
-    }
-  } else {
-    int32_t lv = taosArrayGetSize(pStream->tasks);
-    for (int32_t i = 0; i < lv; i++) {
-      SArray *pTasks = taosArrayGetP(pStream->tasks, i);
-      int32_t sz = taosArrayGetSize(pTasks);
-      for (int32_t j = 0; j < sz; j++) {
-        SStreamTask *pTask = taosArrayGetP(pTasks, j);
-        if (pTask->taskLevel != TASK_LEVEL__SOURCE) break;
-        ASSERT(pTask->taskLevel != TASK_LEVEL__SINK);
-        if (mndPersistTaskRecoverReq(pTrans, pTask) < 0) {
-          return -1;
-        }
-      }
-    }
-  }
-  return 0;
-}
-#endif
-
 int32_t mndDropStreamTasks(SMnode *pMnode, STrans *pTrans, SStreamObj *pStream) {
   int32_t lv = taosArrayGetSize(pStream->tasks);
   for (int32_t i = 0; i < lv; i++) {

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

@@ -73,11 +73,11 @@ int32_t mndInitSubscribe(SMnode *pMnode) {
       .deleteFp = (SdbDeleteFp)mndSubActionDelete,
   };
 
-  mndSetMsgHandle(pMnode, TDMT_VND_MQ_VG_CHANGE_RSP, mndTransProcessRsp);
-  mndSetMsgHandle(pMnode, TDMT_VND_MQ_VG_DELETE_RSP, mndTransProcessRsp);
-  mndSetMsgHandle(pMnode, TDMT_MND_MQ_DO_REBALANCE, mndProcessRebalanceReq);
-  mndSetMsgHandle(pMnode, TDMT_MND_MQ_DROP_CGROUP, mndProcessDropCgroupReq);
-  mndSetMsgHandle(pMnode, TDMT_MND_MQ_DROP_CGROUP_RSP, mndTransProcessRsp);
+  mndSetMsgHandle(pMnode, TDMT_VND_TMQ_SUBSCRIBE_RSP, mndTransProcessRsp);
+  mndSetMsgHandle(pMnode, TDMT_VND_TMQ_DELETE_SUB_RSP, mndTransProcessRsp);
+  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DO_REBALANCE, mndProcessRebalanceReq);
+  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_CGROUP, mndProcessDropCgroupReq);
+  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_CGROUP_RSP, mndTransProcessRsp);
 
   mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_SUBSCRIPTIONS, mndRetrieveSubscribe);
   mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndCancelGetNextSubscribe);
@@ -164,7 +164,7 @@ static int32_t mndPersistSubChangeVgReq(SMnode *pMnode, STrans *pTrans, const SM
   action.epSet = mndGetVgroupEpset(pMnode, pVgObj);
   action.pCont = buf;
   action.contLen = tlen;
-  action.msgType = TDMT_VND_MQ_VG_CHANGE;
+  action.msgType = TDMT_VND_TMQ_SUBSCRIBE;
 
   mndReleaseVgroup(pMnode, pVgObj);
   if (mndTransAppendRedoAction(pTrans, &action) != 0) {
@@ -920,7 +920,7 @@ int32_t mndDropSubByTopic(SMnode *pMnode, STrans *pTrans, const char *topicName)
       action.epSet = pVgEp->epSet;
       action.pCont = pReq;
       action.contLen = sizeof(SMqVDeleteReq);
-      action.msgType = TDMT_VND_MQ_VG_DELETE;
+      action.msgType = TDMT_VND_TMQ_DELETE_SUB;
       if (mndTransAppendRedoAction(pTrans, &action) != 0) {
         taosMemoryFree(pReq);
         return -1;

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

@@ -53,11 +53,10 @@ int32_t mndInitTopic(SMnode *pMnode) {
       .deleteFp = (SdbDeleteFp)mndTopicActionDelete,
   };
 
-  mndSetMsgHandle(pMnode, TDMT_MND_CREATE_TOPIC, mndProcessCreateTopicReq);
-  mndSetMsgHandle(pMnode, TDMT_MND_DROP_TOPIC, mndProcessDropTopicReq);
-  mndSetMsgHandle(pMnode, TDMT_VND_DROP_TOPIC_RSP, mndTransProcessRsp);
-  mndSetMsgHandle(pMnode, TDMT_VND_ADD_CHECK_INFO_RSP, mndTransProcessRsp);
-  mndSetMsgHandle(pMnode, TDMT_VND_DELETE_CHECK_INFO_RSP, mndTransProcessRsp);
+  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CREATE_TOPIC, mndProcessCreateTopicReq);
+  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_DROP_TOPIC, mndProcessDropTopicReq);
+  mndSetMsgHandle(pMnode, TDMT_VND_TMQ_ADD_CHECKINFO_RSP, mndTransProcessRsp);
+  mndSetMsgHandle(pMnode, TDMT_VND_TMQ_DEL_CHECKINFO_RSP, mndTransProcessRsp);
 
   mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndRetrieveTopic);
   mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndCancelGetNextTopic);
@@ -506,7 +505,7 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
       action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
       action.pCont = buf;
       action.contLen = sizeof(SMsgHead) + len;
-      action.msgType = TDMT_VND_ADD_CHECK_INFO;
+      action.msgType = TDMT_VND_TMQ_ADD_CHECKINFO;
       if (mndTransAppendRedoAction(pTrans, &action) != 0) {
         taosMemoryFree(buf);
         sdbRelease(pSdb, pVgroup);
@@ -715,7 +714,7 @@ static int32_t mndProcessDropTopicReq(SRpcMsg *pReq) {
       action.epSet = mndGetVgroupEpset(pMnode, pVgroup);
       action.pCont = buf;
       action.contLen = sizeof(SMsgHead) + TSDB_TOPIC_FNAME_LEN;
-      action.msgType = TDMT_VND_DELETE_CHECK_INFO;
+      action.msgType = TDMT_VND_TMQ_DEL_CHECKINFO;
       if (mndTransAppendRedoAction(pTrans, &action) != 0) {
         taosMemoryFree(buf);
         sdbRelease(pSdb, pVgroup);

source/dnode/qnode/src/qnode.c

@@ -90,12 +90,12 @@ int32_t qndProcessQueryMsg(SQnode *pQnode, int64_t ts, SRpcMsg *pMsg) {
       code = qWorkerProcessFetchMsg(pQnode, pQnode->pQuery, pMsg, ts);
       break;
     case TDMT_SCH_CANCEL_TASK:
-      //code = qWorkerProcessCancelMsg(pQnode, pQnode->pQuery, pMsg, ts);
+      // code = qWorkerProcessCancelMsg(pQnode, pQnode->pQuery, pMsg, ts);
       break;
     case TDMT_SCH_DROP_TASK:
       code = qWorkerProcessDropMsg(pQnode, pQnode->pQuery, pMsg, ts);
       break;
-    case TDMT_VND_CONSUME:
+    case TDMT_VND_TMQ_CONSUME:
       // code =  tqProcessConsumeReq(pQnode->pTq, pMsg);
       // break;
     case TDMT_SCH_QUERY_HEARTBEAT:

source/dnode/snode/src/snode.c

@@ -261,18 +261,52 @@ int32_t sndProcessWriteMsg(SSnode *pSnode, SRpcMsg *pMsg, SRpcMsg *pRsp) {
   return 0;
 }
 
+int32_t sndProcessTaskRecoverFinishReq(SSnode *pSnode, SRpcMsg *pMsg) {
+  char   *msg = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
+  int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);
+
+  // deserialize
+  SStreamRecoverFinishReq req;
+
+  SDecoder decoder;
+  tDecoderInit(&decoder, msg, msgLen);
+  tDecodeSStreamRecoverFinishReq(&decoder, &req);
+  tDecoderClear(&decoder);
+
+  // find task
+  SStreamTask *pTask = streamMetaGetTask(pSnode->pMeta, req.taskId);
+  if (pTask == NULL) {
+    return -1;
+  }
+  // do process request
+  if (streamProcessRecoverFinishReq(pTask, req.childId) < 0) {
+    return -1;
+  }
+
+  return 0;
+}
+
+int32_t sndProcessTaskRecoverFinishRsp(SSnode *pSnode, SRpcMsg *pMsg) {
+  //
+  return 0;
+}
+
 int32_t sndProcessStreamMsg(SSnode *pSnode, SRpcMsg *pMsg) {
   switch (pMsg->msgType) {
     case TDMT_STREAM_TASK_RUN:
       return sndProcessTaskRunReq(pSnode, pMsg);
     case TDMT_STREAM_TASK_DISPATCH:
       return sndProcessTaskDispatchReq(pSnode, pMsg, true);
-    case TDMT_STREAM_RETRIEVE:
-      return sndProcessTaskRetrieveReq(pSnode, pMsg);
     case TDMT_STREAM_TASK_DISPATCH_RSP:
       return sndProcessTaskDispatchRsp(pSnode, pMsg);
+    case TDMT_STREAM_RETRIEVE:
+      return sndProcessTaskRetrieveReq(pSnode, pMsg);
     case TDMT_STREAM_RETRIEVE_RSP:
       return sndProcessTaskRetrieveRsp(pSnode, pMsg);
+    case TDMT_STREAM_RECOVER_FINISH:
+      return sndProcessTaskRecoverFinishReq(pSnode, pMsg);
+    case TDMT_STREAM_RECOVER_FINISH_RSP:
+      return sndProcessTaskRecoverFinishRsp(pSnode, pMsg);
     default:
       ASSERT(0);
   }

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

@@ -176,8 +176,8 @@ int32_t tqCheckColModifiable(STQ* pTq, int64_t tbUid, int32_t colId);
 // tq-mq
 int32_t tqProcessAddCheckInfoReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
 int32_t tqProcessDelCheckInfoReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
-int32_t tqProcessVgChangeReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
-int32_t tqProcessVgDeleteReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
+int32_t tqProcessSubscribeReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
+int32_t tqProcessDeleteSubReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
 int32_t tqProcessOffsetCommitReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
 int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg);
 // tq-stream
@@ -187,11 +187,15 @@ int32_t tqProcessSubmitReq(STQ* pTq, SSubmitReq* data, int64_t ver);
 int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver);
 int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg, bool exec);
-int32_t tqProcessTaskRecoverReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg);
-int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg);
+// int32_t tqProcessTaskRecoverReq(STQ* pTq, SRpcMsg* pMsg);
+// int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRetrieveRsp(STQ* pTq, SRpcMsg* pMsg);
+int32_t tqProcessTaskRecover1Req(STQ* pTq, SRpcMsg* pMsg);
+int32_t tqProcessTaskRecover2Req(STQ* pTq, int64_t version, char* msg, int32_t msgLen);
+int32_t tqProcessTaskRecoverFinishReq(STQ* pTq, SRpcMsg* pMsg);
+int32_t tqProcessTaskRecoverFinishRsp(STQ* pTq, SRpcMsg* pMsg);
 
 SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchema* pSchema,
                             SSchemaWrapper* pTagSchemaWrapper, bool createTb, int64_t suid, const char* stbFullName,

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

@@ -709,7 +709,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg) {
   return 0;
 }
 
-int32_t tqProcessVgDeleteReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) {
+int32_t tqProcessDeleteSubReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) {
   SMqVDeleteReq* pReq = (SMqVDeleteReq*)msg;
 
   taosWLockLatch(&pTq->pushLock);
@@ -767,7 +767,7 @@ int32_t tqProcessDelCheckInfoReq(STQ* pTq, int64_t version, char* msg, int32_t m
   return 0;
 }
 
-int32_t tqProcessVgChangeReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) {
+int32_t tqProcessSubscribeReq(STQ* pTq, int64_t version, char* msg, int32_t msgLen) {
   SMqRebVgReq req = {0};
   tDecodeSMqRebVgReq(msg, &req);
   // todo lock
@@ -982,25 +982,33 @@ int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t version, char* msg, int32_t msg
 
   // 3.go through recover steps to fill history
   if (pTask->fillHistory) {
-    streamSetParamForRecover(pTask);
     if (pTask->taskLevel == TASK_LEVEL__SOURCE) {
+      streamSetParamForRecover(pTask);
       streamSourceRecoverPrepareStep1(pTask, version);
 
       SStreamRecoverStep1Req req;
       streamBuildSourceRecover1Req(pTask, &req);
-
-      void*   serialziedReq = (void*)&req;
       int32_t len = sizeof(SStreamRecoverStep1Req);
 
+      void* serializedReq = rpcMallocCont(len);
+      if (serializedReq == NULL) {
+        return -1;
+      }
+
+      memcpy(serializedReq, &req, len);
+
       SRpcMsg rpcMsg = {
           .contLen = len,
-          .pCont = serialziedReq,
+          .pCont = serializedReq,
           .msgType = TDMT_VND_STREAM_RECOVER_STEP1,
       };
 
-      tmsgPutToQueue(&pTq->pVnode->msgCb, STREAM_QUEUE, &rpcMsg);
+      if (tmsgPutToQueue(&pTq->pVnode->msgCb, STREAM_QUEUE, &rpcMsg) < 0) {
+        /*ASSERT(0);*/
+      }
 
     } else if (pTask->taskLevel == TASK_LEVEL__AGG) {
+      streamSetParamForRecover(pTask);
       streamAggRecoverPrepare(pTask);
     } else if (pTask->taskLevel == TASK_LEVEL__SINK) {
       // do nothing
@@ -1010,8 +1018,11 @@ int32_t tqProcessTaskDeployReq(STQ* pTq, int64_t version, char* msg, int32_t msg
   return 0;
 }
 
-int32_t tqProcessTaskRecover1Req(STQ* pTq, char* msg, int32_t msgLen) {
+int32_t tqProcessTaskRecover1Req(STQ* pTq, SRpcMsg* pMsg) {
   int32_t code;
+  char*   msg = pMsg->pCont;
+  int32_t msgLen = pMsg->contLen;
+
   SStreamRecoverStep1Req* pReq = (SStreamRecoverStep1Req*)msg;
   SStreamTask*            pTask = streamMetaGetTask(pTq->pStreamMeta, pReq->taskId);
   if (pTask == NULL) {
@@ -1035,16 +1046,24 @@ int32_t tqProcessTaskRecover1Req(STQ* pTq, char* msg, int32_t msgLen) {
     return -1;
   }
 
+  ASSERT(pReq->taskId == pTask->taskId);
+
   // serialize msg
-  int32_t len = sizeof(SStreamRecoverStep2Req);
-  void*   serializedReq = (void*)&req;
+  int32_t len = sizeof(SStreamRecoverStep1Req);
+
+  void* serializedReq = rpcMallocCont(len);
+  if (serializedReq == NULL) {
+    return -1;
+  }
+
+  memcpy(serializedReq, &req, len);
 
   // dispatch msg
   SRpcMsg rpcMsg = {
       .code = 0,
       .contLen = len,
       .msgType = TDMT_VND_STREAM_RECOVER_STEP2,
-      .pCont = (void*)serializedReq,
+      .pCont = serializedReq,
   };
 
   tmsgPutToQueue(&pTq->pVnode->msgCb, WRITE_QUEUE, &rpcMsg);
@@ -1087,15 +1106,15 @@ int32_t tqProcessTaskRecover2Req(STQ* pTq, int64_t version, char* msg, int32_t m
   return 0;
 }
 
-int32_t tqProcessTaskRecoverFinishReq(STQ* pTq, char* msg, int32_t msgLen) {
-  int32_t code;
+int32_t tqProcessTaskRecoverFinishReq(STQ* pTq, SRpcMsg* pMsg) {
+  char*   msg = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
+  int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);
 
   // deserialize
-  int32_t                 len;
   SStreamRecoverFinishReq req;
 
   SDecoder decoder;
-  tDecoderInit(&decoder, msg, sizeof(SStreamRecoverFinishReq));
+  tDecoderInit(&decoder, msg, msgLen);
   tDecodeSStreamRecoverFinishReq(&decoder, &req);
   tDecoderClear(&decoder);
 
@@ -1112,6 +1131,11 @@ int32_t tqProcessTaskRecoverFinishReq(STQ* pTq, char* msg, int32_t msgLen) {
   return 0;
 }
 
+int32_t tqProcessTaskRecoverFinishRsp(STQ* pTq, SRpcMsg* pMsg) {
+  //
+  return 0;
+}
+
 int32_t tqProcessDelReq(STQ* pTq, void* pReq, int32_t len, int64_t ver) {
   bool        failed = false;
   SDecoder*   pCoder = &(SDecoder){0};
@@ -1235,7 +1259,7 @@ int32_t tqProcessSubmitReq(STQ* pTq, SSubmitReq* pReq, int64_t ver) {
   pSubmit = streamDataSubmitNew(pReq);
   if (pSubmit == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
-    qError("failed to create data submit for stream since out of memory");
+    tqError("failed to create data submit for stream since out of memory");
     failed = true;
   }
 
@@ -1244,18 +1268,21 @@ int32_t tqProcessSubmitReq(STQ* pTq, SSubmitReq* pReq, int64_t ver) {
     if (pIter == NULL) break;
     SStreamTask* pTask = *(SStreamTask**)pIter;
     if (pTask->taskLevel != TASK_LEVEL__SOURCE) continue;
-    if (pTask->taskStatus == TASK_STATUS__RECOVER_PREPARE || pTask->taskStatus == TASK_STATUS__RECOVER1) continue;
+    if (pTask->taskStatus == TASK_STATUS__RECOVER_PREPARE || pTask->taskStatus == TASK_STATUS__RECOVER1) {
+      tqDebug("skip push task %d, task status %d", pTask->taskId, pTask->taskStatus);
+      continue;
+    }
 
-    qDebug("data submit enqueue stream task: %d, ver: %" PRId64, pTask->taskId, ver);
+    tqDebug("data submit enqueue stream task: %d, ver: %" PRId64, pTask->taskId, ver);
 
     if (!failed) {
       if (streamTaskInput(pTask, (SStreamQueueItem*)pSubmit) < 0) {
-        qError("stream task input failed, task id %d", pTask->taskId);
+        tqError("stream task input failed, task id %d", pTask->taskId);
         continue;
       }
 
       if (streamSchedExec(pTask) < 0) {
-        qError("stream task launch failed, task id %d", pTask->taskId);
+        tqError("stream task launch failed, task id %d", pTask->taskId);
         continue;
       }
     } else {
@@ -1306,33 +1333,6 @@ int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg, bool exec) {
   }
 }
 
-#if 0
-int32_t tqProcessTaskRecoverReq(STQ* pTq, SRpcMsg* pMsg) {
-  SStreamTaskRecoverReq* pReq = pMsg->pCont;
-  int32_t                taskId = pReq->taskId;
-  SStreamTask*           pTask = streamMetaGetTask(pTq->pStreamMeta, taskId);
-  if (pTask) {
-    streamProcessRecoverReq(pTask, pReq, pMsg);
-    return 0;
-  } else {
-    return -1;
-  }
-}
-
-int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg) {
-  SStreamTaskRecoverRsp* pRsp = pMsg->pCont;
-  int32_t                taskId = pRsp->rspTaskId;
-
-  SStreamTask* pTask = streamMetaGetTask(pTq->pStreamMeta, taskId);
-  if (pTask) {
-    streamProcessRecoverRsp(pTask, pRsp);
-    return 0;
-  } else {
-    return -1;
-  }
-}
-#endif
-
 int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg) {
   SStreamDispatchRsp* pRsp = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
   int32_t             taskId = pRsp->taskId;

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

@@ -228,30 +228,30 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp
       if (vnodeProcessBatchDeleteReq(pVnode, version, pReq, len, pRsp) < 0) goto _err;
       break;
     /* TQ */
-    case TDMT_VND_MQ_VG_CHANGE:
-      if (tqProcessVgChangeReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
+    case TDMT_VND_TMQ_SUBSCRIBE:
+      if (tqProcessSubscribeReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
                                 pMsg->contLen - sizeof(SMsgHead)) < 0) {
         goto _err;
       }
       break;
-    case TDMT_VND_MQ_VG_DELETE:
-      if (tqProcessVgDeleteReq(pVnode->pTq, version, pMsg->pCont, pMsg->contLen) < 0) {
+    case TDMT_VND_TMQ_DELETE_SUB:
+      if (tqProcessDeleteSubReq(pVnode->pTq, version, pMsg->pCont, pMsg->contLen) < 0) {
         goto _err;
       }
       break;
-    case TDMT_VND_MQ_COMMIT_OFFSET:
+    case TDMT_VND_TMQ_COMMIT_OFFSET:
       if (tqProcessOffsetCommitReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
                                    pMsg->contLen - sizeof(SMsgHead)) < 0) {
         goto _err;
       }
       break;
-    case TDMT_VND_ADD_CHECK_INFO:
+    case TDMT_VND_TMQ_ADD_CHECKINFO:
       if (tqProcessAddCheckInfoReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
                                    pMsg->contLen - sizeof(SMsgHead)) < 0) {
         goto _err;
       }
       break;
-    case TDMT_VND_DELETE_CHECK_INFO:
+    case TDMT_VND_TMQ_DEL_CHECKINFO:
       if (tqProcessDelCheckInfoReq(pVnode->pTq, version, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
                                    pMsg->contLen - sizeof(SMsgHead)) < 0) {
         goto _err;
@@ -268,6 +268,11 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp
         goto _err;
       }
     } break;
+    case TDMT_VND_STREAM_RECOVER_STEP2: {
+      if (tqProcessTaskRecover2Req(pVnode->pTq, version, pMsg->pCont, pMsg->contLen) < 0) {
+        goto _err;
+      }
+    } break;
     case TDMT_VND_ALTER_CONFIRM:
       vnodeProcessAlterConfirmReq(pVnode, version, pReq, len, pRsp);
       break;
@@ -356,14 +361,11 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
     return 0;
   }
 
-  if (pMsg->msgType == TDMT_VND_CONSUME && !pVnode->restored) {
+  if (pMsg->msgType == TDMT_VND_TMQ_CONSUME && !pVnode->restored) {
     vnodeRedirectRpcMsg(pVnode, pMsg);
     return 0;
   }
 
-  char   *msgstr = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
-  int32_t msgLen = pMsg->contLen - sizeof(SMsgHead);
-
   switch (pMsg->msgType) {
     case TDMT_SCH_FETCH:
     case TDMT_SCH_MERGE_FETCH:
@@ -382,7 +384,7 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
       return vnodeGetTableCfg(pVnode, pMsg, true);
     case TDMT_VND_BATCH_META:
       return vnodeGetBatchMeta(pVnode, pMsg);
-    case TDMT_VND_CONSUME:
+    case TDMT_VND_TMQ_CONSUME:
       return tqProcessPollReq(pVnode->pTq, pMsg);
     case TDMT_STREAM_TASK_RUN:
       return tqProcessTaskRunReq(pVnode->pTq, pMsg);
@@ -390,16 +392,18 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
     case TDMT_STREAM_TASK_DISPATCH:
       return tqProcessTaskDispatchReq(pVnode->pTq, pMsg, true);
 #endif
-      /*case TDMT_STREAM_TASK_RECOVER:*/
-      /*return tqProcessTaskRecoverReq(pVnode->pTq, pMsg);*/
-    case TDMT_STREAM_RETRIEVE:
-      return tqProcessTaskRetrieveReq(pVnode->pTq, pMsg);
     case TDMT_STREAM_TASK_DISPATCH_RSP:
       return tqProcessTaskDispatchRsp(pVnode->pTq, pMsg);
-      /*case TDMT_STREAM_TASK_RECOVER_RSP:*/
-      /*return tqProcessTaskRecoverRsp(pVnode->pTq, pMsg);*/
+    case TDMT_STREAM_RETRIEVE:
+      return tqProcessTaskRetrieveReq(pVnode->pTq, pMsg);
     case TDMT_STREAM_RETRIEVE_RSP:
       return tqProcessTaskRetrieveRsp(pVnode->pTq, pMsg);
+    case TDMT_VND_STREAM_RECOVER_STEP1:
+      return tqProcessTaskRecover1Req(pVnode->pTq, pMsg);
+    case TDMT_STREAM_RECOVER_FINISH:
+      return tqProcessTaskRecoverFinishReq(pVnode->pTq, pMsg);
+    case TDMT_STREAM_RECOVER_FINISH_RSP:
+      return tqProcessTaskRecoverFinishRsp(pVnode->pTq, pMsg);
     default:
       vError("unknown msg type:%d in fetch queue", pMsg->msgType);
       return TSDB_CODE_VND_APP_ERROR;

source/libs/executor/inc/executorimpl.h

@@ -150,18 +150,13 @@ typedef struct {
 
   SSchemaWrapper*     schema;
   char                tbName[TSDB_TABLE_NAME_LEN];
-  SSDataBlock*        pullOverBlk;  // for streaming
-  SWalFilterCond      cond;
-  int64_t             lastScanUid;
   int8_t              recoverStep;
   SQueryTableDataCond tableCond;
-  int64_t             recoverStartVer;
-  int64_t             recoverEndVer;
   int64_t             fillHistoryVer1;
   int64_t             fillHistoryVer2;
 
-  int8_t        triggerSaved;
-  int64_t       deleteMarkSaved;
+  // int8_t        triggerSaved;
+  // int64_t       deleteMarkSaved;
   SStreamState* pState;
 } SStreamTaskInfo;
 
@@ -461,8 +456,10 @@ typedef struct SPartitionDataInfo {
 
 typedef struct STimeWindowAggSupp {
   int8_t          calTrigger;
-  int64_t         waterMark;
+  int8_t          calTriggerSaved;
   int64_t         deleteMark;
+  int64_t         deleteMarkSaved;
+  int64_t         waterMark;
   TSKEY           maxTs;
   TSKEY           minTs;
   SColumnInfoData timeWindowData;  // query time window info for scalar function execution.

source/libs/executor/src/executor.c

@@ -70,6 +70,26 @@ static int32_t doSetSMABlock(SOperatorInfo* pOperator, void* input, size_t numOf
   }
 }
 
+static int32_t doSetStreamOpOpen(SOperatorInfo* pOperator, char* id) {
+  {
+    ASSERT(pOperator != NULL);
+    if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
+      if (pOperator->numOfDownstream == 0) {
+        qError("failed to find stream scan operator to set the input data block, %s" PRIx64, id);
+        return TSDB_CODE_QRY_APP_ERROR;
+      }
+
+      if (pOperator->numOfDownstream > 1) {  // not handle this in join query
+        qError("join not supported for stream block scan, %s" PRIx64, id);
+        return TSDB_CODE_QRY_APP_ERROR;
+      }
+      pOperator->status = OP_NOT_OPENED;
+      return doSetStreamOpOpen(pOperator->pDownstream[0], id);
+    }
+  }
+  return 0;
+}
+
 static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t numOfBlocks, int32_t type, char* id) {
   ASSERT(pOperator != NULL);
   if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
@@ -117,7 +137,22 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
   }
 }
 
-static FORCE_INLINE void streamInputBlockDataDestory(void* pBlock) { blockDataDestroy((SSDataBlock*)pBlock); }
+int32_t qSetStreamOpOpen(qTaskInfo_t tinfo) {
+  if (tinfo == NULL) {
+    return TSDB_CODE_QRY_APP_ERROR;
+  }
+
+  SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
+
+  int32_t code = doSetStreamOpOpen(pTaskInfo->pRoot, GET_TASKID(pTaskInfo));
+  if (code != TSDB_CODE_SUCCESS) {
+    qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo));
+  } else {
+    qDebug("%s set the stream block successfully", GET_TASKID(pTaskInfo));
+  }
+
+  return code;
+}
 
 int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numOfBlocks, int32_t type) {
   if (tinfo == NULL) {
@@ -706,8 +741,7 @@ int32_t qStreamInput(qTaskInfo_t tinfo, void* pItem) {
 int32_t qStreamSourceRecoverStep1(qTaskInfo_t tinfo, int64_t ver) {
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
   ASSERT(pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM);
-  pTaskInfo->streamInfo.recoverStartVer = 0;
-  pTaskInfo->streamInfo.recoverEndVer = ver;
+  pTaskInfo->streamInfo.fillHistoryVer1 = ver;
   pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__PREPARE1;
   return 0;
 }
@@ -715,8 +749,7 @@ int32_t qStreamSourceRecoverStep1(qTaskInfo_t tinfo, int64_t ver) {
 int32_t qStreamSourceRecoverStep2(qTaskInfo_t tinfo, int64_t ver) {
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
   ASSERT(pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM);
-  pTaskInfo->streamInfo.recoverStartVer = pTaskInfo->streamInfo.recoverEndVer;
-  pTaskInfo->streamInfo.recoverEndVer = ver;
+  pTaskInfo->streamInfo.fillHistoryVer2 = ver;
   pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__PREPARE2;
   return 0;
 }
@@ -737,22 +770,44 @@ int32_t qStreamSetParamForRecover(qTaskInfo_t tinfo) {
         pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL ||
         pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL) {
       SStreamIntervalOperatorInfo* pInfo = pOperator->info;
-      pTaskInfo->streamInfo.triggerSaved = pInfo->twAggSup.calTrigger;
-      pTaskInfo->streamInfo.deleteMarkSaved = pInfo->twAggSup.deleteMark;
+      ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE ||
+             pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE);
+      ASSERT(pInfo->twAggSup.calTriggerSaved == 0);
+      ASSERT(pInfo->twAggSup.deleteMarkSaved == 0);
+
+      qInfo("save stream param for interval: %d,  %" PRId64, pInfo->twAggSup.calTrigger, pInfo->twAggSup.deleteMark);
+
+      pInfo->twAggSup.calTriggerSaved = pInfo->twAggSup.calTrigger;
+      pInfo->twAggSup.deleteMarkSaved = pInfo->twAggSup.deleteMark;
       pInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
       pInfo->twAggSup.deleteMark = INT64_MAX;
+
     } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION ||
                pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION ||
                pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) {
       SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
-      pTaskInfo->streamInfo.triggerSaved = pInfo->twAggSup.calTrigger;
-      pTaskInfo->streamInfo.deleteMarkSaved = pInfo->twAggSup.deleteMark;
+      ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE ||
+             pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE);
+      ASSERT(pInfo->twAggSup.calTriggerSaved == 0);
+      ASSERT(pInfo->twAggSup.deleteMarkSaved == 0);
+
+      qInfo("save stream param for session: %d,  %" PRId64, pInfo->twAggSup.calTrigger, pInfo->twAggSup.deleteMark);
+
+      pInfo->twAggSup.calTriggerSaved = pInfo->twAggSup.calTrigger;
+      pInfo->twAggSup.deleteMarkSaved = pInfo->twAggSup.deleteMark;
       pInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
       pInfo->twAggSup.deleteMark = INT64_MAX;
     } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE) {
       SStreamStateAggOperatorInfo* pInfo = pOperator->info;
-      pTaskInfo->streamInfo.triggerSaved = pInfo->twAggSup.calTrigger;
-      pTaskInfo->streamInfo.deleteMarkSaved = pInfo->twAggSup.deleteMark;
+      ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE ||
+             pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE);
+      ASSERT(pInfo->twAggSup.calTriggerSaved == 0);
+      ASSERT(pInfo->twAggSup.deleteMarkSaved == 0);
+
+      qInfo("save stream param for state: %d,  %" PRId64, pInfo->twAggSup.calTrigger, pInfo->twAggSup.deleteMark);
+
+      pInfo->twAggSup.calTriggerSaved = pInfo->twAggSup.calTrigger;
+      pInfo->twAggSup.deleteMarkSaved = pInfo->twAggSup.deleteMark;
       pInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
       pInfo->twAggSup.deleteMark = INT64_MAX;
     }
@@ -782,21 +837,36 @@ int32_t qStreamRestoreParam(qTaskInfo_t tinfo) {
         pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL ||
         pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL) {
       SStreamIntervalOperatorInfo* pInfo = pOperator->info;
-
-      pInfo->twAggSup.calTrigger = pTaskInfo->streamInfo.triggerSaved;
-      pInfo->twAggSup.deleteMark = pTaskInfo->streamInfo.deleteMarkSaved;
+      ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE);
+      ASSERT(pInfo->twAggSup.deleteMark == INT64_MAX);
+
+      pInfo->twAggSup.calTrigger = pInfo->twAggSup.calTriggerSaved;
+      pInfo->twAggSup.deleteMark = pInfo->twAggSup.deleteMarkSaved;
+      ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE ||
+             pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE);
+      qInfo("restore stream param for interval: %d,  %" PRId64, pInfo->twAggSup.calTrigger, pInfo->twAggSup.deleteMark);
     } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION ||
                pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION ||
                pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) {
       SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
-
-      pInfo->twAggSup.calTrigger = pTaskInfo->streamInfo.triggerSaved;
-      pInfo->twAggSup.deleteMark = pTaskInfo->streamInfo.deleteMarkSaved;
+      ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE);
+      ASSERT(pInfo->twAggSup.deleteMark == INT64_MAX);
+
+      pInfo->twAggSup.calTrigger = pInfo->twAggSup.calTriggerSaved;
+      pInfo->twAggSup.deleteMark = pInfo->twAggSup.deleteMarkSaved;
+      ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE ||
+             pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE);
+      qInfo("restore stream param for session: %d,  %" PRId64, pInfo->twAggSup.calTrigger, pInfo->twAggSup.deleteMark);
     } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE) {
       SStreamStateAggOperatorInfo* pInfo = pOperator->info;
-
-      pInfo->twAggSup.calTrigger = pTaskInfo->streamInfo.triggerSaved;
-      pInfo->twAggSup.deleteMark = pTaskInfo->streamInfo.deleteMarkSaved;
+      ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE);
+      ASSERT(pInfo->twAggSup.deleteMark == INT64_MAX);
+
+      pInfo->twAggSup.calTrigger = pInfo->twAggSup.calTriggerSaved;
+      pInfo->twAggSup.deleteMark = pInfo->twAggSup.deleteMarkSaved;
+      ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE ||
+             pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE);
+      qInfo("restore stream param for state: %d,  %" PRId64, pInfo->twAggSup.calTrigger, pInfo->twAggSup.deleteMark);
     }
 
     // iterate operator tree

source/libs/executor/src/scanoperator.c

@@ -1890,6 +1890,11 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
       pTSInfo->cond.startVersion = pTaskInfo->streamInfo.fillHistoryVer1 + 1;
       pTSInfo->cond.endVersion = pTaskInfo->streamInfo.fillHistoryVer2;
     }
+
+    /*resetTableScanInfo(pTSInfo, pWin);*/
+    tsdbReaderClose(pTSInfo->dataReader);
+    pTSInfo->dataReader = NULL;
+
     pTSInfo->scanTimes = 0;
     pTSInfo->currentGroupId = -1;
     pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__SCAN;
@@ -1903,6 +1908,10 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) {
       return pBlock;
     }
     pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__NONE;
+    STableScanInfo* pTSInfo = pInfo->pTableScanOp->info;
+    pTSInfo->cond.startVersion = 0;
+    pTSInfo->cond.endVersion = -1;
+
     return NULL;
   }
 #endif

source/libs/executor/src/timewindowoperator.c

@@ -3303,6 +3303,8 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
       // for test 315360000000
       .deleteMark = 1000LL * 60LL * 60LL * 24LL * 365LL * 10LL,
       // .deleteMark = INT64_MAX,
+      .deleteMarkSaved = 0,
+      .calTriggerSaved = 0,
   };
   ASSERT(pInfo->twAggSup.calTrigger != STREAM_TRIGGER_MAX_DELAY);
   pInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;

source/libs/parser/src/parTranslater.c

@@ -5281,7 +5281,7 @@ static int32_t translateCreateTopic(STranslateContext* pCxt, SCreateTopicStmt* p
     code = buildCreateTopicReq(pCxt, pStmt, &createReq);
   }
   if (TSDB_CODE_SUCCESS == code) {
-    code = buildCmdMsg(pCxt, TDMT_MND_CREATE_TOPIC, (FSerializeFunc)tSerializeSCMCreateTopicReq, &createReq);
+    code = buildCmdMsg(pCxt, TDMT_MND_TMQ_CREATE_TOPIC, (FSerializeFunc)tSerializeSCMCreateTopicReq, &createReq);
   }
   tFreeSCMCreateTopicReq(&createReq);
   return code;
@@ -5295,7 +5295,7 @@ static int32_t translateDropTopic(STranslateContext* pCxt, SDropTopicStmt* pStmt
   tNameGetFullDbName(&name, dropReq.name);
   dropReq.igNotExists = pStmt->ignoreNotExists;
 
-  return buildCmdMsg(pCxt, TDMT_MND_DROP_TOPIC, (FSerializeFunc)tSerializeSMDropTopicReq, &dropReq);
+  return buildCmdMsg(pCxt, TDMT_MND_TMQ_DROP_TOPIC, (FSerializeFunc)tSerializeSMDropTopicReq, &dropReq);
 }
 
 static int32_t translateDropCGroup(STranslateContext* pCxt, SDropCGroupStmt* pStmt) {
@@ -5307,7 +5307,7 @@ static int32_t translateDropCGroup(STranslateContext* pCxt, SDropCGroupStmt* pSt
   dropReq.igNotExists = pStmt->ignoreNotExists;
   strcpy(dropReq.cgroup, pStmt->cgroup);
 
-  return buildCmdMsg(pCxt, TDMT_MND_MQ_DROP_CGROUP, (FSerializeFunc)tSerializeSMDropCgroupReq, &dropReq);
+  return buildCmdMsg(pCxt, TDMT_MND_TMQ_DROP_CGROUP, (FSerializeFunc)tSerializeSMDropCgroupReq, &dropReq);
 }
 
 static int32_t translateAlterLocal(STranslateContext* pCxt, SAlterLocalStmt* pStmt) {

source/libs/stream/inc/streamInc.h

@@ -43,6 +43,9 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock)
 
 int32_t tEncodeStreamRetrieveReq(SEncoder* pEncoder, const SStreamRetrieveReq* pReq);
 
+int32_t streamDispatchOneRecoverFinishReq(SStreamTask* pTask, const SStreamRecoverFinishReq* pReq, int32_t vgId,
+                                          SEpSet* pEpSet);
+
 SStreamQueueItem* streamMergeQueueItem(SStreamQueueItem* dst, SStreamQueueItem* elem);
 void              streamFreeQitem(SStreamQueueItem* data);
 

source/libs/stream/src/streamDispatch.c

@@ -239,7 +239,7 @@ int32_t streamDispatchOneRecoverFinishReq(SStreamTask* pTask, const SStreamRecov
 
   msg.contLen = tlen + sizeof(SMsgHead);
   msg.pCont = buf;
-  msg.msgType = TDMT_VND_STREAM_RECOVER_FINISH;
+  msg.msgType = TDMT_STREAM_RECOVER_FINISH;
 
   tmsgSendReq(pEpSet, &msg);
 
@@ -250,7 +250,7 @@ FAIL:
   return code;
 }
 
-int32_t streamDispatchOneReq(SStreamTask* pTask, const SStreamDispatchReq* pReq, int32_t vgId, SEpSet* pEpSet) {
+int32_t streamDispatchOneDataReq(SStreamTask* pTask, const SStreamDispatchReq* pReq, int32_t vgId, SEpSet* pEpSet) {
   void*   buf = NULL;
   int32_t code = -1;
   SRpcMsg msg = {0};
@@ -292,13 +292,19 @@ FAIL:
 
 int32_t streamSearchAndAddBlock(SStreamTask* pTask, SStreamDispatchReq* pReqs, SSDataBlock* pDataBlock, int32_t vgSz,
                                 int64_t groupId) {
-  char* ctbName;
+  char* ctbName = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN);
+  if (ctbName == NULL) {
+    return -1;
+  }
+
   if (pDataBlock->info.parTbName[0]) {
-    ctbName = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN);
     snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->shuffleDispatcher.dbInfo.db, pDataBlock->info.parTbName);
   } else {
-    ctbName = buildCtbNameByGroupId(pTask->shuffleDispatcher.stbFullName, groupId);
+    char* ctbShortName = buildCtbNameByGroupId(pTask->shuffleDispatcher.stbFullName, groupId);
+    snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->shuffleDispatcher.dbInfo.db, ctbShortName);
+    taosMemoryFree(ctbShortName);
   }
+
   SArray* vgInfo = pTask->shuffleDispatcher.dbInfo.pVgroupInfos;
 
   /*uint32_t hashValue = MurmurHash3_32(ctbName, strlen(ctbName));*/
@@ -365,7 +371,7 @@ int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pDat
     qDebug("dispatch from task %d (child id %d) to down stream task %d in vnode %d", pTask->taskId, pTask->selfChildId,
            downstreamTaskId, vgId);
 
-    if (streamDispatchOneReq(pTask, &req, vgId, pEpSet) < 0) {
+    if (streamDispatchOneDataReq(pTask, &req, vgId, pEpSet) < 0) {
       goto FAIL_FIXED_DISPATCH;
     }
     code = 0;
@@ -427,7 +433,7 @@ int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pDat
       if (pReqs[i].blockNum > 0) {
         // send
         SVgroupInfo* pVgInfo = taosArrayGet(vgInfo, i);
-        if (streamDispatchOneReq(pTask, &pReqs[i], pVgInfo->vgId, &pVgInfo->epSet) < 0) {
+        if (streamDispatchOneDataReq(pTask, &pReqs[i], pVgInfo->vgId, &pVgInfo->epSet) < 0) {
           goto FAIL_SHUFFLE_DISPATCH;
         }
       }

source/libs/stream/src/streamExec.c

@@ -90,6 +90,8 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
 
   void* exec = pTask->exec.executor;
 
+  qSetStreamOpOpen(exec);
+
   while (1) {
     SArray* pRes = taosArrayInit(0, sizeof(SSDataBlock));
     if (pRes == NULL) {
@@ -127,7 +129,10 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
     qRes->type = STREAM_INPUT__DATA_BLOCK;
     qRes->blocks = pRes;
     streamTaskOutput(pTask, qRes);
-    // TODO stream sched dispatch
+
+    if (pTask->outputType == TASK_OUTPUT__FIXED_DISPATCH || pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) {
+      streamDispatch(pTask);
+    }
   }
   return 0;
 }

source/libs/stream/src/streamRecover.c

@@ -36,6 +36,7 @@ int32_t streamSourceRecoverPrepareStep1(SStreamTask* pTask, int64_t ver) {
 }
 
 int32_t streamBuildSourceRecover1Req(SStreamTask* pTask, SStreamRecoverStep1Req* pReq) {
+  pReq->msgHead.vgId = pTask->nodeId;
   pReq->streamId = pTask->streamId;
   pReq->taskId = pTask->taskId;
   return 0;
@@ -44,10 +45,10 @@ int32_t streamBuildSourceRecover1Req(SStreamTask* pTask, SStreamRecoverStep1Req*
 int32_t streamSourceRecoverScanStep1(SStreamTask* pTask) {
   //
   return streamScanExec(pTask, 100);
-  // TODO next: dispatch msg to launch scan step2
 }
 
 int32_t streamBuildSourceRecover2Req(SStreamTask* pTask, SStreamRecoverStep2Req* pReq) {
+  pReq->msgHead.vgId = pTask->nodeId;
   pReq->streamId = pTask->streamId;
   pReq->taskId = pTask->taskId;
   return 0;
@@ -64,11 +65,20 @@ int32_t streamSourceRecoverScanStep2(SStreamTask* pTask, int64_t ver) {
 int32_t streamDispatchRecoverFinishReq(SStreamTask* pTask) {
   SStreamRecoverFinishReq req = {
       .streamId = pTask->streamId,
-      .taskId = pTask->taskId,
       .childId = pTask->selfChildId,
   };
+  // serialize
   if (pTask->outputType == TASK_OUTPUT__FIXED_DISPATCH) {
+    req.taskId = pTask->fixedEpDispatcher.taskId;
+    streamDispatchOneRecoverFinishReq(pTask, &req, pTask->fixedEpDispatcher.nodeId, &pTask->fixedEpDispatcher.epSet);
   } else if (pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) {
+    SArray* vgInfo = pTask->shuffleDispatcher.dbInfo.pVgroupInfos;
+    int32_t vgSz = taosArrayGetSize(vgInfo);
+    for (int32_t i = 0; i < vgSz; i++) {
+      SVgroupInfo* pVgInfo = taosArrayGet(vgInfo, i);
+      req.taskId = pVgInfo->taskId;
+      streamDispatchOneRecoverFinishReq(pTask, &req, pVgInfo->vgId, &pVgInfo->epSet);
+    }
   }
   return 0;
 }
@@ -76,9 +86,9 @@ int32_t streamDispatchRecoverFinishReq(SStreamTask* pTask) {
 // agg
 int32_t streamAggRecoverPrepare(SStreamTask* pTask) {
   void* exec = pTask->exec.executor;
-  if (qStreamSetParamForRecover(exec) < 0) {
-    return -1;
-  }
+  /*if (qStreamSetParamForRecover(exec) < 0) {*/
+  /*return -1;*/
+  /*}*/
   pTask->recoverWaitingChild = taosArrayGetSize(pTask->childEpInfo);
   return 0;
 }
@@ -96,11 +106,13 @@ int32_t streamAggChildrenRecoverFinish(SStreamTask* pTask) {
 }
 
 int32_t streamProcessRecoverFinishReq(SStreamTask* pTask, int32_t childId) {
+  if (pTask->taskLevel == TASK_LEVEL__AGG) {
     int32_t left = atomic_sub_fetch_32(&pTask->recoverWaitingChild, 1);
     ASSERT(left >= 0);
     if (left == 0) {
       streamAggChildrenRecoverFinish(pTask);
     }
+  }
   return 0;
 }
 

tests/script/jenkins/basic.txt

@@ -234,6 +234,7 @@
 ./test.sh -f tsim/stream/basic1.sim
 ./test.sh -f tsim/stream/basic2.sim
 ./test.sh -f tsim/stream/drop_stream.sim
+./test.sh -f tsim/stream/fillHistoryBasic1.sim
 ./test.sh -f tsim/stream/distributeInterval0.sim
 ./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
 ./test.sh -f tsim/stream/distributeSession0.sim

tests/script/tsim/stream/fillHistoryBasic1.sim

+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start
+sleep 50
+sql connect
+
+print =============== create database
+sql create database test vgroups 1;
+sql select * from information_schema.ins_databases
+if $rows != 3 then
+  return -1
+endi
+
+print $data00 $data01 $data02
+
+sql use test;
+
+sql create table t1(ts timestamp, a int, b int , c int, d double);
+
+sql create stream stream1 trigger at_once fill_history 1 into streamt as select  _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b)  c4, min(c) c5 from t1 interval(10s);
+
+sql insert into t1 values(1648791213000,1,2,3,1.0);
+sql insert into t1 values(1648791223001,2,2,3,1.1);
+sql insert into t1 values(1648791233002,3,2,3,2.1);
+sql insert into t1 values(1648791243003,4,2,3,3.1);
+sql insert into t1 values(1648791213004,4,2,3,4.1);
+
+
+sleep 1000
+sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
+
+if $rows != 4 then
+  print ======$rows
+  return -1
+endi
+
+# row 0
+if $data01 != 2 then
+  print ======$data01
+  return -1
+endi
+
+if $data02 != 2 then
+  print ======$data02
+  return -1
+endi
+
+if $data03 != 5 then
+  print ======$data03
+  return -1
+endi
+
+if $data04 != 2 then
+  print ======$data04
+  return -1
+endi
+
+if $data05 != 3 then
+  print ======$data05
+  return -1
+endi
+
+# row 1
+if $data11 != 1 then
+  print ======$data11
+  return -1
+endi
+
+if $data12 != 1 then
+  print ======$data12
+  return -1
+endi
+
+if $data13 != 2 then
+  print ======$data13
+  return -1
+endi
+
+if $data14 != 2 then
+  print ======$data14
+  return -1
+endi
+
+if $data15 != 3 then
+  print ======$data15
+  return -1
+endi
+
+# row 2
+if $data21 != 1 then
+  print ======$data21
+  return -1
+endi
+
+if $data22 != 1 then
+  print ======$data22
+  return -1
+endi
+
+if $data23 != 3 then
+  print ======$data23
+  return -1
+endi
+
+if $data24 != 2 then
+  print ======$data24
+  return -1
+endi
+
+if $data25 != 3 then
+  print ======$data25
+  return -1
+endi
+
+# row 3
+if $data31 != 1 then
+  print ======$data31
+  return -1
+endi
+
+if $data32 != 1 then
+  print ======$data32
+  return -1
+endi
+
+if $data33 != 4 then
+  print ======$data33
+  return -1
+endi
+
+if $data34 != 2 then
+  print ======$data34
+  return -1
+endi
+
+if $data35 != 3 then
+  print ======$data35
+  return -1
+endi
+
+sql insert into t1 values(1648791223001,12,14,13,11.1);
+sleep 500
+sql select * from streamt;
+
+print    count(*) , count(d) , sum(a) , max(b) , min(c)
+print 0: $data00 , $data01 , $data02 , $data03 , $data04 , $data05
+print 1: $data10 , $data11 , $data12 , $data13 , $data14 , $data15
+
+if $rows != 4 then
+  print ======$rows
+  return -1
+endi
+
+# row 0
+if $data01 != 2 then
+  print ======$data01
+  return -1
+endi
+
+if $data02 != 2 then
+  print ======$data02
+  return -1
+endi
+
+if $data03 != 5 then
+  print ======$data03
+  return -1
+endi
+
+if $data04 != 2 then
+  print ======$data04
+  return -1
+endi
+
+if $data05 != 3 then
+  print ======$data05
+  return -1
+endi
+
+# row 1
+if $data11 != 1 then
+  print ======$data11
+  return -1
+endi
+
+if $data12 != 1 then
+  print ======$data12
+  return -1
+endi
+
+if $data13 != 12 then
+  print ======$data13
+  return -1
+endi
+
+if $data14 != 14 then
+  print ======$data14
+  return -1
+endi
+
+if $data15 != 13 then
+  print ======$data15
+  return -1
+endi
+
+# row 2
+if $data21 != 1 then
+  print ======$data21
+  return -1
+endi
+
+if $data22 != 1 then
+  print ======$data22
+  return -1
+endi
+
+if $data23 != 3 then
+  print ======$data23
+  return -1
+endi
+
+if $data24 != 2 then
+  print ======$data24
+  return -1
+endi
+
+if $data25 != 3 then
+  print ======$data25
+  return -1
+endi
+
+# row 3
+if $data31 != 1 then
+  print ======$data31
+  return -1
+endi
+
+if $data32 != 1 then
+  print ======$data32
+  return -1
+endi
+
+if $data33 != 4 then
+  print ======$data33
+  return -1
+endi
+
+if $data34 != 2 then
+  print ======$data34
+  return -1
+endi
+
+if $data35 != 3 then
+  print ======$data35
+  return -1
+endi
+
+sql insert into t1 values(1648791223002,12,14,13,11.1);
+sleep 100
+sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
+
+# row 1
+if $data11 != 2 then
+  print ======$data11
+  return -1
+endi
+
+if $data12 != 2 then
+  print ======$data12
+  return -1
+endi
+
+if $data13 != 24 then
+  print ======$data13
+  return -1
+endi
+
+if $data14 != 14 then
+  print ======$data14
+  return -1
+endi
+
+if $data15 != 13 then
+  print ======$data15
+  return -1
+endi
+
+sql insert into t1 values(1648791223003,12,14,13,11.1);
+sleep 100
+sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
+
+# row 1
+if $data11 != 3 then
+  print ======$data11
+  return -1
+endi
+
+if $data12 != 3 then
+  print ======$data12
+  return -1
+endi
+
+if $data13 != 36 then
+  print ======$data13
+  return -1
+endi
+
+if $data14 != 14 then
+  print ======$data14
+  return -1
+endi
+
+if $data15 != 13 then
+  print ======$data15
+  return -1
+endi
+
+sql insert into t1 values(1648791223001,1,1,1,1.1);
+sql insert into t1 values(1648791223002,2,2,2,2.1);
+sql insert into t1 values(1648791223003,3,3,3,3.1);
+sleep 100
+sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
+
+# row 1
+if $data11 != 3 then
+  print ======$data11
+  return -1
+endi
+
+if $data12 != 3 then
+  print ======$data12
+  return -1
+endi
+
+if $data13 != 6 then
+  print ======$data13
+  return -1
+endi
+
+if $data14 != 3 then
+  print ======$data14
+  return -1
+endi
+
+if $data15 != 1 then
+  print ======$data15
+  return -1
+endi
+
+sql insert into t1 values(1648791233003,3,2,3,2.1);
+sql insert into t1 values(1648791233002,5,6,7,8.1);
+sql insert into t1 values(1648791233002,3,2,3,2.1);
+sleep 100
+sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
+
+# row 2
+if $data21 != 2 then
+  print ======$data21
+  return -1
+endi
+
+if $data22 != 2 then
+  print ======$data22
+  return -1
+endi
+
+if $data23 != 6 then
+  print ======$data23
+  return -1
+endi
+
+if $data24 != 2 then
+  print ======$data24
+  return -1
+endi
+
+if $data25 != 3 then
+  print ======$data25
+  return -1
+endi
+
+sql insert into t1 values(1648791213004,4,2,3,4.1) (1648791213006,5,4,7,9.1) (1648791213004,40,20,30,40.1) (1648791213005,4,2,3,4.1);
+sleep 100
+sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
+
+# row 0
+if $data01 != 4 then
+  print ======$data01
+  return -1
+endi
+
+if $data02 != 4 then
+  print ======$data02
+  return -1
+endi
+
+if $data03 != 50 then
+  print ======$data03 != 50
+  return -1
+endi
+
+if $data04 != 20 then
+  print ======$data04 != 20
+  return -1
+endi
+
+if $data05 != 3 then
+  print ======$data05
+  return -1
+endi
+
+sql insert into t1 values(1648791223004,4,2,3,4.1) (1648791233006,5,4,7,9.1) (1648791223004,40,20,30,40.1) (1648791233005,4,2,3,4.1);
+sleep 100
+sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
+
+# row 1
+if $data11 != 4 then
+  print ======$data11
+  return -1
+endi
+
+if $data12 != 4 then
+  print ======$data12
+  return -1
+endi
+
+if $data13 != 46 then
+  print ======$data13 != 46
+  return -1
+endi
+
+if $data14 != 20 then
+  print ======$data14 != 20
+  return -1
+endi
+
+if $data15 != 1 then
+  print ======$data15
+  return -1
+endi
+
+# row 2
+if $data21 != 4 then
+  print ======$data21
+  return -1
+endi
+
+if $data22 != 4 then
+  print ======$data22
+  return -1
+endi
+
+if $data23 != 15 then
+  print ======$data23
+    return -1
+endi
+
+if $data24 != 4 then
+  print ======$data24
+  return -1
+endi
+
+if $data25 != 3 then
+  print ======$data25
+  return -1
+endi
+
+
+
+
+
+sql create database test2 vgroups 1;
+sql select * from information_schema.ins_databases;
+
+sql use test2;
+
+sql create table t1(ts timestamp, a int, b int , c int, d double);
+
+sql insert into t1 values(1648791213000,1,2,3,1.0);
+sql insert into t1 values(1648791223001,2,2,3,1.1);
+sql insert into t1 values(1648791233002,3,2,3,2.1);
+sql insert into t1 values(1648791243003,4,2,3,3.1);
+sql insert into t1 values(1648791213004,4,2,3,4.1);
+
+sql create stream stream2 trigger at_once fill_history 1 into streamt as select  _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b)  c4, min(c) c5 from t1 interval(10s);
+
+sleep 1000
+sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
+
+if $rows != 4 then
+  print ======$rows
+  return -1
+endi
+
+# row 0
+if $data01 != 2 then
+  print ======$data01
+  return -1
+endi
+
+if $data02 != 2 then
+  print ======$data02
+  return -1
+endi
+
+if $data03 != 5 then
+  print ======$data03
+  return -1
+endi
+
+if $data04 != 2 then
+  print ======$data04
+  return -1
+endi
+
+if $data05 != 3 then
+  print ======$data05
+  return -1
+endi
+
+# row 1
+if $data11 != 1 then
+  print ======$data11
+  return -1
+endi
+
+if $data12 != 1 then
+  print ======$data12
+  return -1
+endi
+
+if $data13 != 2 then
+  print ======$data13
+  return -1
+endi
+
+if $data14 != 2 then
+  print ======$data14
+  return -1
+endi
+
+if $data15 != 3 then
+  print ======$data15
+  return -1
+endi
+
+# row 2
+if $data21 != 1 then
+  print ======$data21
+  return -1
+endi
+
+if $data22 != 1 then
+  print ======$data22
+  return -1
+endi
+
+if $data23 != 3 then
+  print ======$data23
+  return -1
+endi
+
+if $data24 != 2 then
+  print ======$data24
+  return -1
+endi
+
+if $data25 != 3 then
+  print ======$data25
+  return -1
+endi
+
+# row 3
+if $data31 != 1 then
+  print ======$data31
+  return -1
+endi
+
+if $data32 != 1 then
+  print ======$data32
+  return -1
+endi
+
+if $data33 != 4 then
+  print ======$data33
+  return -1
+endi
+
+if $data34 != 2 then
+  print ======$data34
+  return -1
+endi
+
+if $data35 != 3 then
+  print ======$data35
+  return -1
+endi
+
+sql insert into t1 values(1648791223001,12,14,13,11.1);
+sleep 500
+sql select * from streamt;
+
+print    count(*) , count(d) , sum(a) , max(b) , min(c)
+print 0: $data00 , $data01 , $data02 , $data03 , $data04 , $data05
+print 1: $data10 , $data11 , $data12 , $data13 , $data14 , $data15
+
+if $rows != 4 then
+  print ======$rows
+  return -1
+endi
+
+# row 0
+if $data01 != 2 then
+  print ======$data01
+  return -1
+endi
+
+if $data02 != 2 then
+  print ======$data02
+  return -1
+endi
+
+if $data03 != 5 then
+  print ======$data03
+  return -1
+endi
+
+if $data04 != 2 then
+  print ======$data04
+  return -1
+endi
+
+if $data05 != 3 then
+  print ======$data05
+  return -1
+endi
+
+# row 1
+if $data11 != 1 then
+  print ======$data11
+  return -1
+endi
+
+if $data12 != 1 then
+  print ======$data12
+  return -1
+endi
+
+if $data13 != 12 then
+  print ======$data13
+  return -1
+endi
+
+if $data14 != 14 then
+  print ======$data14
+  return -1
+endi
+
+if $data15 != 13 then
+  print ======$data15
+  return -1
+endi
+
+# row 2
+if $data21 != 1 then
+  print ======$data21
+  return -1
+endi
+
+if $data22 != 1 then
+  print ======$data22
+  return -1
+endi
+
+if $data23 != 3 then
+  print ======$data23
+  return -1
+endi
+
+if $data24 != 2 then
+  print ======$data24
+  return -1
+endi
+
+if $data25 != 3 then
+  print ======$data25
+  return -1
+endi
+
+# row 3
+if $data31 != 1 then
+  print ======$data31
+  return -1
+endi
+
+if $data32 != 1 then
+  print ======$data32
+  return -1
+endi
+
+if $data33 != 4 then
+  print ======$data33
+  return -1
+endi
+
+if $data34 != 2 then
+  print ======$data34
+  return -1
+endi
+
+if $data35 != 3 then
+  print ======$data35
+  return -1
+endi
+
+sql insert into t1 values(1648791223002,12,14,13,11.1);
+sleep 100
+sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
+
+# row 1
+if $data11 != 2 then
+  print ======$data11
+  return -1
+endi
+
+if $data12 != 2 then
+  print ======$data12
+  return -1
+endi
+
+if $data13 != 24 then
+  print ======$data13
+  return -1
+endi
+
+if $data14 != 14 then
+  print ======$data14
+  return -1
+endi
+
+if $data15 != 13 then
+  print ======$data15
+  return -1
+endi
+
+sql insert into t1 values(1648791223003,12,14,13,11.1);
+sleep 100
+sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
+
+# row 1
+if $data11 != 3 then
+  print ======$data11
+  return -1
+endi
+
+if $data12 != 3 then
+  print ======$data12
+  return -1
+endi
+
+if $data13 != 36 then
+  print ======$data13
+  return -1
+endi
+
+if $data14 != 14 then
+  print ======$data14
+  return -1
+endi
+
+if $data15 != 13 then
+  print ======$data15
+  return -1
+endi
+
+sql insert into t1 values(1648791223001,1,1,1,1.1);
+sql insert into t1 values(1648791223002,2,2,2,2.1);
+sql insert into t1 values(1648791223003,3,3,3,3.1);
+sleep 100
+sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
+
+# row 1
+if $data11 != 3 then
+  print ======$data11
+  return -1
+endi
+
+if $data12 != 3 then
+  print ======$data12
+  return -1
+endi
+
+if $data13 != 6 then
+  print ======$data13
+  return -1
+endi
+
+if $data14 != 3 then
+  print ======$data14
+  return -1
+endi
+
+if $data15 != 1 then
+  print ======$data15
+  return -1
+endi
+
+sql insert into t1 values(1648791233003,3,2,3,2.1);
+sql insert into t1 values(1648791233002,5,6,7,8.1);
+sql insert into t1 values(1648791233002,3,2,3,2.1);
+sleep 100
+sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
+
+# row 2
+if $data21 != 2 then
+  print ======$data21
+  return -1
+endi
+
+if $data22 != 2 then
+  print ======$data22
+  return -1
+endi
+
+if $data23 != 6 then
+  print ======$data23
+  return -1
+endi
+
+if $data24 != 2 then
+  print ======$data24
+  return -1
+endi
+
+if $data25 != 3 then
+  print ======$data25
+  return -1
+endi
+
+sql insert into t1 values(1648791213004,4,2,3,4.1) (1648791213006,5,4,7,9.1) (1648791213004,40,20,30,40.1) (1648791213005,4,2,3,4.1);
+sleep 100
+sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
+
+# row 0
+if $data01 != 4 then
+  print ======$data01
+  return -1
+endi
+
+if $data02 != 4 then
+  print ======$data02
+  return -1
+endi
+
+if $data03 != 50 then
+  print ======$data03 != 50
+  return -1
+endi
+
+if $data04 != 20 then
+  print ======$data04 != 20
+  return -1
+endi
+
+if $data05 != 3 then
+  print ======$data05
+  return -1
+endi
+
+sql insert into t1 values(1648791223004,4,2,3,4.1) (1648791233006,5,4,7,9.1) (1648791223004,40,20,30,40.1) (1648791233005,4,2,3,4.1);
+sleep 100
+sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
+
+# row 1
+if $data11 != 4 then
+  print ======$data11
+  return -1
+endi
+
+if $data12 != 4 then
+  print ======$data12
+  return -1
+endi
+
+if $data13 != 46 then
+  print ======$data13 != 46
+  return -1
+endi
+
+if $data14 != 20 then
+  print ======$data14 != 20
+  return -1
+endi
+
+if $data15 != 1 then
+  print ======$data15
+  return -1
+endi
+
+# row 2
+if $data21 != 4 then
+  print ======$data21
+  return -1
+endi
+
+if $data22 != 4 then
+  print ======$data22
+  return -1
+endi
+
+if $data23 != 15 then
+  print ======$data23
+    return -1
+endi
+
+if $data24 != 4 then
+  print ======$data24
+  return -1
+endi
+
+if $data25 != 3 then
+  print ======$data25
+  return -1
+endi
+
+