Merge remote-tracking branch 'origin/3.0' into fix/tsim

include/common/tcommon.h

@@ -42,12 +42,13 @@ enum {
 typedef enum EStreamType {
   STREAM_NORMAL = 1,
   STREAM_INVERT,
-  STREAM_REPROCESS,
+  STREAM_CLEAR,
   STREAM_INVALID,
   STREAM_GET_ALL,
   STREAM_DELETE,
   STREAM_RETRIEVE,
   STREAM_PUSH_DATA,
+  STREAM_PUSH_EMPTY,
 } EStreamType;
 
 typedef struct {

include/common/tdatablock.h

@@ -224,6 +224,7 @@ size_t blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize);
 int32_t blockDataTrimFirstNRows(SSDataBlock* pBlock, size_t n);
 
 int32_t      assignOneDataBlock(SSDataBlock* dst, const SSDataBlock* src);
+int32_t copyDataBlock(SSDataBlock* dst, const SSDataBlock* src);
 SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock, bool copyData);
 SSDataBlock* createDataBlock();
 int32_t      blockDataAppendColInfo(SSDataBlock* pBlock, SColumnInfoData* pColInfoData);
@@ -236,6 +237,8 @@ void        blockCompressEncode(const SSDataBlock* pBlock, char* data, int32_t*
 const char* blockCompressDecode(SSDataBlock* pBlock, int32_t numOfCols, int32_t numOfRows, const char* pData);
 
 void blockDebugShowData(const SArray* dataBlocks, const char* flag);
+// for debug
+char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** dumpBuf);
 
 int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks, STSchema* pTSchema, int32_t vgId,
                                     tb_uid_t suid);

include/libs/stream/tstream.h

@@ -319,7 +319,7 @@ static FORCE_INLINE int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem
       return -1;
     }
     taosWriteQitem(pTask->inputQueue->queue, pSubmitClone);
-  } else if (pItem->type == STREAM_INPUT__DATA_BLOCK) {
+  } else if (pItem->type == STREAM_INPUT__DATA_BLOCK || pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
     taosWriteQitem(pTask->inputQueue->queue, pItem);
   } else if (pItem->type == STREAM_INPUT__CHECKPOINT) {
     taosWriteQitem(pTask->inputQueue->queue, pItem);

include/libs/stream/tstreamUpdate.h

@@ -32,12 +32,15 @@ typedef struct SUpdateInfo {
   int64_t interval;
   int64_t watermark;
   TSKEY minTS;
+  SScalableBf* pCloseWinSBF;
 } SUpdateInfo;
 
 SUpdateInfo *updateInfoInitP(SInterval* pInterval, int64_t watermark);
 SUpdateInfo *updateInfoInit(int64_t interval, int32_t precision, int64_t watermark);
 bool updateInfoIsUpdated(SUpdateInfo *pInfo, tb_uid_t tableId, TSKEY ts);
 void updateInfoDestroy(SUpdateInfo *pInfo);
+void updateInfoAddCloseWindowSBF(SUpdateInfo *pInfo);
+void updateInfoDestoryColseWinSBF(SUpdateInfo *pInfo);
 
 #ifdef __cplusplus
 }

source/common/src/tdatablock.c

@@ -1164,7 +1164,7 @@ int32_t colInfoDataEnsureCapacity(SColumnInfoData* pColumn, uint32_t numOfRows)
 
 int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows) {
   int32_t code = 0;
-  ASSERT(numOfRows > 0);
+  //ASSERT(numOfRows > 0);
 
   if (numOfRows == 0) {
     return TSDB_CODE_SUCCESS;
@@ -1230,6 +1230,32 @@ int32_t assignOneDataBlock(SSDataBlock* dst, const SSDataBlock* src) {
   return 0;
 }
 
+int32_t copyDataBlock(SSDataBlock* dst, const SSDataBlock* src) {
+  ASSERT(src != NULL && dst != NULL);
+
+  blockDataCleanup(dst);
+  int32_t code = blockDataEnsureCapacity(dst, src->info.rows);
+  if (code != TSDB_CODE_SUCCESS) {
+    terrno = code;
+    return code;
+  }
+
+  size_t numOfCols = taosArrayGetSize(src->pDataBlock);
+  for (int32_t i = 0; i < numOfCols; ++i) {
+    SColumnInfoData* pDst = taosArrayGet(dst->pDataBlock, i);
+    SColumnInfoData* pSrc = taosArrayGet(src->pDataBlock, i);
+    if (pSrc->pData == NULL) {
+      continue;
+    }
+
+    colDataAssign(pDst, pSrc, src->info.rows, &src->info);
+  }
+
+  dst->info.rows = src->info.rows;
+  dst->info.window = src->info.window;
+  return TSDB_CODE_SUCCESS;
+}
+
 SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock, bool copyData) {
   if (pDataBlock == NULL) {
     return NULL;
@@ -1627,6 +1653,56 @@ void blockDebugShowData(const SArray* dataBlocks, const char* flag) {
   }
 }
 
+// for debug
+char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf) {
+  int32_t size = 2048;
+  *pDataBuf = taosMemoryCalloc(size, 1);
+  char* dumpBuf = *pDataBuf;
+  char pBuf[128] = {0};
+  int32_t colNum = taosArrayGetSize(pDataBlock->pDataBlock);
+  int32_t rows = pDataBlock->info.rows;
+  int32_t len = 0;
+  len += snprintf(dumpBuf + len, size - len, "\n%s |block type %d |child id %d|\n", flag, (int32_t)pDataBlock->info.type, pDataBlock->info.childId);
+  for (int32_t j = 0; j < rows; j++) {
+    len += snprintf(dumpBuf + len, size - len, "%s |", flag);
+    for (int32_t k = 0; k < colNum; k++) {
+      SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
+      void*            var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
+      if (colDataIsNull(pColInfoData, rows, j, NULL)) {
+        len += snprintf(dumpBuf + len, size - len, " %15s |", "NULL");
+        continue;
+      }
+      switch (pColInfoData->info.type) {
+        case TSDB_DATA_TYPE_TIMESTAMP:
+          formatTimestamp(pBuf, *(uint64_t*)var, TSDB_TIME_PRECISION_MILLI);
+          len += snprintf(dumpBuf + len, size - len, " %25s |", pBuf);
+          break;
+        case TSDB_DATA_TYPE_INT:
+          len += snprintf(dumpBuf + len, size - len, " %15d |", *(int32_t*)var);
+          break;
+        case TSDB_DATA_TYPE_UINT:
+          len += snprintf(dumpBuf + len, size - len, " %15u |", *(uint32_t*)var);
+          break;
+        case TSDB_DATA_TYPE_BIGINT:
+          len += snprintf(dumpBuf + len, size - len, " %15ld |", *(int64_t*)var);
+          break;
+        case TSDB_DATA_TYPE_UBIGINT:
+          len += snprintf(dumpBuf + len, size - len, " %15lu |", *(uint64_t*)var);
+          break;
+        case TSDB_DATA_TYPE_FLOAT:
+          len += snprintf(dumpBuf + len, size - len, " %15f |", *(float*)var);
+          break;
+        case TSDB_DATA_TYPE_DOUBLE:
+          len += snprintf(dumpBuf + len, size - len, " %15lf |", *(double*)var);
+          break;
+      }
+    }
+    len += snprintf(dumpBuf + len, size - len, "\n");
+  }
+  len += snprintf(dumpBuf + len, size - len, "%s |end\n", flag);
+  return dumpBuf;
+}
+
 /**
  * @brief TODO: Assume that the final generated result it less than 3M
  *

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

@@ -401,6 +401,10 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *
     }
   } else if (pCreate->subType == TOPIC_SUB_TYPE__TABLE) {
     SStbObj *pStb = mndAcquireStb(pMnode, pCreate->subStbName);
+    if (pStb == NULL) {
+      terrno = TSDB_CODE_MND_STB_NOT_EXIST;
+      return -1;
+    }
     topicObj.stbUid = pStb->uid;
   }
   /*} else if (pCreate->subType == TOPIC_SUB_TYPE__DB) {*/

source/dnode/snode/src/snode.c

@@ -257,7 +257,7 @@ int32_t sndProcessSMsg(SSnode *pSnode, SRpcMsg *pMsg) {
     case TDMT_STREAM_TASK_RECOVER_RSP:
       return sndProcessTaskRecoverRsp(pSnode, pMsg);
     case TDMT_STREAM_RETRIEVE_RSP:
-      return sndProcessTaskRecoverRsp(pSnode, pMsg);
+      return sndProcessTaskRetrieveRsp(pSnode, pMsg);
     default:
       ASSERT(0);
   }

source/dnode/vnode/inc/vnode.h

@@ -196,6 +196,7 @@ struct SVnodeCfg {
 typedef struct {
   TSKEY    lastKey;
   uint64_t uid;
+  uint64_t groupId;
 } STableKeyInfo;
 
 struct SMetaEntry {

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

@@ -109,11 +109,15 @@ int32_t tqReadHandleSetMsg(STqReadHandle* pReadHandle, SSubmitReq* pMsg, int64_t
 }
 
 bool tqNextDataBlock(STqReadHandle* pHandle) {
+  if (pHandle->pMsg == NULL) return false;
   while (1) {
     if (tGetSubmitMsgNext(&pHandle->msgIter, &pHandle->pBlock) < 0) {
       return false;
     }
-    if (pHandle->pBlock == NULL) return false;
+    if (pHandle->pBlock == NULL) {
+      pHandle->pMsg = NULL;
+      return false;
+    }
 
     if (pHandle->tbIdHash == NULL) {
       return true;

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

@@ -18,57 +18,87 @@
 SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, bool createTb, int64_t suid,
                             const char* stbFullName, int32_t vgId) {
   SSubmitReq* ret = NULL;
+  SArray*     schemaReqs = NULL;
+  SArray*     schemaReqSz = NULL;
   SArray*     tagArray = taosArrayInit(1, sizeof(STagVal));
   if (!tagArray) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return NULL;
   }
 
-  // cal size
-  int32_t cap = sizeof(SSubmitReq);
   int32_t sz = taosArrayGetSize(pBlocks);
-  for (int32_t i = 0; i < sz; i++) {
-    SSDataBlock* pDataBlock = taosArrayGet(pBlocks, i);
-    int32_t      rows = pDataBlock->info.rows;
-    // TODO min
-    int32_t rowSize = pDataBlock->info.rowSize;
-    int32_t maxLen = TD_ROW_MAX_BYTES_FROM_SCHEMA(pTSchema);
-    int32_t schemaLen = 0;
 
-    if (createTb) {
-      SVCreateTbReq createTbReq = {0};
-      char*         cname = buildCtbNameByGroupId(stbFullName, pDataBlock->info.groupId);
-      createTbReq.name = cname;
-      createTbReq.flags = 0;
-      createTbReq.type = TSDB_CHILD_TABLE;
-      createTbReq.ctb.suid = suid;
-
-      STagVal tagVal = {
-          .cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1,
-          .type = TSDB_DATA_TYPE_UBIGINT,
-          .i64 = (int64_t)pDataBlock->info.groupId,
+  if (createTb) {
+    schemaReqs = taosArrayInit(sz, sizeof(void*));
+    schemaReqSz = taosArrayInit(sz, sizeof(int32_t));
+    for (int32_t i = 0; i < sz; i++) {
+      SSDataBlock* pDataBlock = taosArrayGet(pBlocks, i);
+      STagVal      tagVal = {
+               .cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1,
+               .type = TSDB_DATA_TYPE_UBIGINT,
+               .i64 = (int64_t)pDataBlock->info.groupId,
       };
       STag* pTag = NULL;
       taosArrayClear(tagArray);
       taosArrayPush(tagArray, &tagVal);
       tTagNew(tagArray, 1, false, &pTag);
       if (pTag == NULL) {
-        tdDestroySVCreateTbReq(&createTbReq);
+        taosArrayDestroy(schemaReqs);
         taosArrayDestroy(tagArray);
         return NULL;
       }
+
+      SVCreateTbReq createTbReq = {0};
+      createTbReq.name = buildCtbNameByGroupId(stbFullName, pDataBlock->info.groupId);
+      createTbReq.flags = 0;
+      createTbReq.type = TSDB_CHILD_TABLE;
+      createTbReq.ctb.suid = suid;
       createTbReq.ctb.pTag = (uint8_t*)pTag;
 
       int32_t code;
+      int32_t schemaLen;
       tEncodeSize(tEncodeSVCreateTbReq, &createTbReq, schemaLen, code);
-
-      tdDestroySVCreateTbReq(&createTbReq);
       if (code < 0) {
+        tdDestroySVCreateTbReq(&createTbReq);
         taosArrayDestroy(tagArray);
+        taosMemoryFreeClear(ret);
+        return NULL;
+      }
+
+      void* schemaStr = taosMemoryMalloc(schemaLen);
+      if (schemaStr == NULL) {
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
+        return NULL;
+      }
+      taosArrayPush(schemaReqs, &schemaStr);
+      taosArrayPush(schemaReqSz, &schemaLen);
+
+      SEncoder encoder = {0};
+      tEncoderInit(&encoder, schemaStr, schemaLen);
+      code = tEncodeSVCreateTbReq(&encoder, &createTbReq);
+      if (code < 0) {
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
         return NULL;
       }
+      tEncoderClear(&encoder);
+      tdDestroySVCreateTbReq(&createTbReq);
     }
+  }
+  taosArrayDestroy(tagArray);
 
+  // cal size
+  int32_t cap = sizeof(SSubmitReq);
+  for (int32_t i = 0; i < sz; i++) {
+    SSDataBlock* pDataBlock = taosArrayGet(pBlocks, i);
+    int32_t      rows = pDataBlock->info.rows;
+    // TODO min
+    int32_t rowSize = pDataBlock->info.rowSize;
+    int32_t maxLen = TD_ROW_MAX_BYTES_FROM_SCHEMA(pTSchema);
+
+    int32_t schemaLen = 0;
+    if (createTb) {
+      schemaLen = *(int32_t*)taosArrayGet(schemaReqSz, i);
+    }
     cap += sizeof(SSubmitBlk) + schemaLen + rows * maxLen;
   }
 
@@ -99,55 +129,13 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
 
     int32_t schemaLen = 0;
     if (createTb) {
-      SVCreateTbReq createTbReq = {0};
-      char*         cname = buildCtbNameByGroupId(stbFullName, pDataBlock->info.groupId);
-      createTbReq.name = cname;
-      createTbReq.flags = 0;
-      createTbReq.type = TSDB_CHILD_TABLE;
-      createTbReq.ctb.suid = suid;
-
-      STagVal tagVal = {
-          .cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1,
-          .type = TSDB_DATA_TYPE_UBIGINT,
-          .i64 = (int64_t)pDataBlock->info.groupId,
-      };
-      taosArrayClear(tagArray);
-      taosArrayPush(tagArray, &tagVal);
-      STag* pTag = NULL;
-      tTagNew(tagArray, 1, false, &pTag);
-      if (pTag == NULL) {
-        tdDestroySVCreateTbReq(&createTbReq);
-        taosArrayDestroy(tagArray);
-        taosMemoryFreeClear(ret);
-        return NULL;
-      }
-      createTbReq.ctb.pTag = (uint8_t*)pTag;
-
-      int32_t code;
-      tEncodeSize(tEncodeSVCreateTbReq, &createTbReq, schemaLen, code);
-      if (code < 0) {
-        tdDestroySVCreateTbReq(&createTbReq);
-        taosArrayDestroy(tagArray);
-        taosMemoryFreeClear(ret);
-        return NULL;
-      }
-
-      SEncoder encoder = {0};
-      tEncoderInit(&encoder, blkSchema, schemaLen);
-      code = tEncodeSVCreateTbReq(&encoder, &createTbReq);
-      tEncoderClear(&encoder);
-      tdDestroySVCreateTbReq(&createTbReq);
-
-      if (code < 0) {
-        taosArrayDestroy(tagArray);
-        taosMemoryFreeClear(ret);
-        return NULL;
-      }
+      schemaLen = *(int32_t*)taosArrayGet(schemaReqSz, i);
+      void* schemaStr = taosArrayGetP(schemaReqs, i);
+      memcpy(blkSchema, schemaStr, schemaLen);
     }
     blkHead->schemaLen = htonl(schemaLen);
 
     STSRow* rowData = POINTER_SHIFT(blkSchema, schemaLen);
-
     for (int32_t j = 0; j < rows; j++) {
       SRowBuilder rb = {0};
       tdSRowInit(&rb, pTSchema->version);
@@ -175,7 +163,10 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
   }
 
   ret->length = htonl(ret->length);
-  taosArrayDestroy(tagArray);
+
+  if (schemaReqs) taosArrayDestroyP(schemaReqs, taosMemoryFree);
+  taosArrayDestroy(schemaReqSz);
+
   return ret;
 }
 

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

@@ -2852,7 +2852,7 @@ int32_t tsdbGetAllTableList(SMeta* pMeta, uint64_t uid, SArray* list) {
       break;
     }
 
-    STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, uid = id};
+    STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, uid = id, .groupId = 0};
     taosArrayPush(list, &info);
   }
 

source/libs/executor/inc/executorimpl.h

@@ -293,6 +293,7 @@ typedef enum EStreamScanMode {
   STREAM_SCAN_FROM_RES,
   STREAM_SCAN_FROM_UPDATERES,
   STREAM_SCAN_FROM_DATAREADER,
+  STREAM_SCAN_FROM_DATAREADER_RETRIEVE,
 } EStreamScanMode;
 
 typedef struct SCatchSupporter {
@@ -348,7 +349,9 @@ typedef struct SStreamBlockScanInfo {
   SArray*        childIds;
   SessionWindowSupporter sessionSup;
   bool            assignBlockUid; // assign block uid to groupId, temporarily used for generating rollup SMA.
-  int32_t         scanWinIndex;
+  int32_t         scanWinIndex;   // for state operator
+  int32_t         pullDataResIndex;
+  SSDataBlock*    pPullDataRes;             // pull data SSDataBlock
 } SStreamBlockScanInfo;
 
 typedef struct SSysTableScanInfo {
@@ -427,8 +430,13 @@ typedef struct SStreamFinalIntervalOperatorInfo {
   STimeWindowAggSupp twAggSup;
   SArray*            pChildren;
   SSDataBlock*       pUpdateRes;
+  bool               returnUpdate;
   SPhysiNode*        pPhyNode;           // create new child
   bool               isFinal;
+  SHashObj*          pPullDataMap;
+  SArray*            pPullWins;          // SPullWindowInfo
+  int32_t            pullIndex;
+  SSDataBlock*       pPullDataRes;
 } SStreamFinalIntervalOperatorInfo;
 
 typedef struct SAggOperatorInfo {
@@ -851,6 +859,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
 void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex);
 
 int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle, SNodeList* groupKey);
+SSDataBlock* createPullDataBlock();
 
 #ifdef __cplusplus
 }

source/libs/executor/src/executil.c

@@ -315,7 +315,7 @@ int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo
       }
 
       for (int i = 0; i < taosArrayGetSize(res); i++) {
-        STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, .uid = *(uint64_t*)taosArrayGet(res, i)};
+        STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, .uid = *(uint64_t*)taosArrayGet(res, i), .groupId = 0};
         taosArrayPush(pListInfo->pTableList, &info);
       }
       taosArrayDestroy(res);
@@ -336,7 +336,7 @@ int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo
       }
     }
   }else {  // Create one table group.
-    STableKeyInfo info = {.lastKey = 0, .uid = tableUid};
+    STableKeyInfo info = {.lastKey = 0, .uid = tableUid, .groupId = 0};
     taosArrayPush(pListInfo->pTableList, &info);
   }
   pListInfo->pGroupList = taosArrayInit(4, POINTER_BYTES);

source/libs/executor/src/executorimpl.c

@@ -4028,6 +4028,7 @@ int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle,
     int32_t   len = (int32_t)(pStart - (char*)keyBuf);
     uint64_t groupId = calcGroupId(keyBuf, len);
     taosHashPut(pTableListInfo->map, &(info->uid), sizeof(uint64_t), &groupId, sizeof(uint64_t));
+    info->groupId = groupId;
     groupNum++;
 
     nodesClearList(groupNew);
@@ -4126,7 +4127,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
           return NULL;
         }
       } else {  // Create one table group.
-        STableKeyInfo info = {.lastKey = 0, .uid = pBlockNode->uid};
+        STableKeyInfo info = {.lastKey = 0, .uid = pBlockNode->uid, .groupId = 0};
         taosArrayPush(pTableListInfo->pTableList, &info);
       }
 

source/libs/executor/src/scanoperator.c

@@ -776,15 +776,14 @@ static bool isStateWindow(SStreamBlockScanInfo* pInfo) {
   return pInfo->sessionSup.parentType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE;
 }
 
-static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
-  SSDataBlock* pSDB = pInfo->pUpdateRes;
+static bool prepareDataScan(SStreamBlockScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) {
   STimeWindow  win = {
        .skey = INT64_MIN,
        .ekey = INT64_MAX,
   };
   bool needRead = false;
-  if (!isStateWindow(pInfo) && pInfo->updateResIndex < pSDB->info.rows) {
-    SColumnInfoData* pColDataInfo = taosArrayGet(pSDB->pDataBlock, pInfo->primaryTsIndex);
+  if (!isStateWindow(pInfo) && (*pRowIndex) < pSDB->info.rows) {
+    SColumnInfoData* pColDataInfo = taosArrayGet(pSDB->pDataBlock, tsColIndex);
     TSKEY*           tsCols = (TSKEY*)pColDataInfo->pData;
     SResultRowInfo   dumyInfo;
     dumyInfo.cur.pageId = -1;
@@ -793,14 +792,14 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
       int64_t              gap = pInfo->sessionSup.gap;
       int32_t              winIndex = 0;
       SResultWindowInfo*   pCurWin =
-          getSessionTimeWindow(pAggSup, tsCols[pInfo->updateResIndex], INT64_MIN, pSDB->info.groupId, gap, &winIndex);
+          getSessionTimeWindow(pAggSup, tsCols[(*pRowIndex)], INT64_MIN, pSDB->info.groupId, gap, &winIndex);
       win = pCurWin->win;
-      pInfo->updateResIndex +=
-          updateSessionWindowInfo(pCurWin, tsCols, NULL, pSDB->info.rows, pInfo->updateResIndex, gap, NULL);
+      (*pRowIndex) +=
+          updateSessionWindowInfo(pCurWin, tsCols, NULL, pSDB->info.rows, (*pRowIndex), gap, NULL);
     } else {
-      win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[pInfo->updateResIndex], &pInfo->interval,
+      win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[(*pRowIndex)], &pInfo->interval,
                                 pInfo->interval.precision, NULL);
-      pInfo->updateResIndex += getNumOfRowsInTimeWindow(&pSDB->info, tsCols, pInfo->updateResIndex, win.ekey,
+      (*pRowIndex) += getNumOfRowsInTimeWindow(&pSDB->info, tsCols, (*pRowIndex), win.ekey,
                                                         binarySearchForKey, NULL, TSDB_ORDER_ASC);
     }
     needRead = true;
@@ -823,6 +822,9 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
   pTableScanInfo->cond.twindows[0] = win;
   pTableScanInfo->curTWinIdx = 0;
 //  tsdbResetReadHandle(pTableScanInfo->dataReader, &pTableScanInfo->cond, 0);
+  // if (!pTableScanInfo->dataReader) {
+  //   return false;
+  // }
   pTableScanInfo->scanTimes = 0;
   pTableScanInfo->currentGroupId = -1;
   return true;
@@ -862,12 +864,12 @@ static uint64_t getGroupId(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_
   */
 }
 
-static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo) {
+static SSDataBlock* doDataScan(SStreamBlockScanInfo* pInfo, SSDataBlock* pSDB, int32_t tsColIndex, int32_t* pRowIndex) {
   while (1) {
     SSDataBlock* pResult = NULL;
     pResult = doTableScan(pInfo->pSnapshotReadOp);
     if (pResult == NULL) {
-      if (prepareDataScan(pInfo)) {
+      if (prepareDataScan(pInfo, pSDB, tsColIndex, pRowIndex)) {
         // scan next window data
         pResult = doTableScan(pInfo->pSnapshotReadOp);
       }
@@ -916,7 +918,7 @@ static void setUpdateData(SStreamBlockScanInfo* pInfo, SSDataBlock* pBlock, SSDa
     pUpdateBlock->info.rows = i;
     pInfo->tsArrayIndex += i;
     pUpdateBlock->info.groupId = pInfo->groupId;
-    pUpdateBlock->info.type = STREAM_REPROCESS;
+    pUpdateBlock->info.type = STREAM_CLEAR;
     blockDataUpdateTsWindow(pUpdateBlock, 0);
   }
   // all rows have same group id
@@ -970,6 +972,14 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
     int32_t      current = pInfo->validBlockIndex++;
     SSDataBlock* pBlock = taosArrayGetP(pInfo->pBlockLists, current);
     blockDataUpdateTsWindow(pBlock, 0);
+    if (pBlock->info.type == STREAM_RETRIEVE) {
+      pInfo->blockType = STREAM_DATA_TYPE_SUBMIT_BLOCK;
+      pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RETRIEVE;
+      copyDataBlock(pInfo->pPullDataRes, pBlock);
+      pInfo->pullDataResIndex = 0;
+      prepareDataScan(pInfo, pInfo->pPullDataRes, 0, &pInfo->pullDataResIndex);
+      updateInfoAddCloseWindowSBF(pInfo->pUpdateInfo);
+    }
     return pBlock;
   } else if (pInfo->blockType == STREAM_DATA_TYPE_SUBMIT_BLOCK) {
     if (pInfo->scanMode == STREAM_SCAN_FROM_RES) {
@@ -979,28 +989,39 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
     } else if (pInfo->scanMode == STREAM_SCAN_FROM_UPDATERES) {
       pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
       if (!isStateWindow(pInfo)) {
-        prepareDataScan(pInfo);
+        prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
       }
       return pInfo->pUpdateRes;
+    } else if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER_RETRIEVE) {
+      SSDataBlock* pSDB = doDataScan(pInfo, pInfo->pPullDataRes, 0, &pInfo->pullDataResIndex);
+      if (pSDB != NULL) {
+        getUpdateDataBlock(pInfo, true, pSDB, NULL);
+        pSDB->info.type = STREAM_PUSH_DATA;
+        return pSDB;
+      }
+      pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
     } else {
       if (isStateWindow(pInfo) && taosArrayGetSize(pInfo->sessionSup.pStreamAggSup->pScanWindow) > 0) {
         pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER;
         pInfo->updateResIndex = pInfo->pUpdateRes->info.rows;
-        prepareDataScan(pInfo);
+        prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
       }
       if (pInfo->scanMode == STREAM_SCAN_FROM_DATAREADER) {
-        SSDataBlock* pSDB = doDataScan(pInfo);
+        SSDataBlock* pSDB = doDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
         if (pSDB == NULL) {
           setUpdateData(pInfo, pInfo->pRes, pInfo->pUpdateRes);
           if (pInfo->pUpdateRes->info.rows > 0) {
             if (!isStateWindow(pInfo)) {
-              prepareDataScan(pInfo);
+              // Todo(liuyao) mybe can delete this.
+              bool test = prepareDataScan(pInfo, pInfo->pUpdateRes, pInfo->primaryTsIndex, &pInfo->updateResIndex);
+              ASSERT(test == false);
             }
             return pInfo->pUpdateRes;
           } else {
             pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
           }
         } else {
+          pSDB->info.type = STREAM_NORMAL;
           getUpdateDataBlock(pInfo, true, pSDB, NULL);
           return pSDB;
         }
@@ -1070,10 +1091,12 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
       taosArrayDestroy(block.pDataBlock);
       if (pInfo->pRes->pDataBlock == NULL) {
         // TODO add log
+        updateInfoDestoryColseWinSBF(pInfo->pUpdateInfo);
         pOperator->status = OP_EXEC_DONE;
         pTaskInfo->code = terrno;
         return NULL;
       }
+
       // currently only the tbname pseudo column
       if (pInfo->numOfPseudoExpr > 0) {
         addTagPseudoColumnData(&pInfo->readHandle, pInfo->pPseudoExpr, pInfo->numOfPseudoExpr, pInfo->pRes);
@@ -1091,12 +1114,13 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
     pOperator->resultInfo.totalRows += pBlockInfo->rows;
 
     if (pBlockInfo->rows == 0) {
+      updateInfoDestoryColseWinSBF(pInfo->pUpdateInfo);
       pOperator->status = OP_EXEC_DONE;
     } else if (pInfo->pUpdateInfo) {
       pInfo->tsArrayIndex = 0;
       getUpdateDataBlock(pInfo, true, pInfo->pRes, pInfo->pUpdateRes);
       if (pInfo->pUpdateRes->info.rows > 0) {
-        if (pInfo->pUpdateRes->info.type == STREAM_REPROCESS) {
+        if (pInfo->pUpdateRes->info.type == STREAM_CLEAR) {
           pInfo->updateResIndex = 0;
           pInfo->scanMode = STREAM_SCAN_FROM_UPDATERES;
         } else if (pInfo->pUpdateRes->info.type == STREAM_INVERT) {
@@ -1209,6 +1233,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle,
   pInfo->scanMode = STREAM_SCAN_FROM_READERHANDLE;
   pInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = NULL, .gap = -1};
   pInfo->groupId = 0;
+  pInfo->pPullDataRes = createPullDataBlock();
 
   pOperator->name = "StreamBlockScanOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN;
@@ -1965,7 +1990,10 @@ _error:
 
 typedef struct STableMergeScanInfo {
   STableListInfo* tableListInfo;
-  int32_t         currentGroupId;
+  int32_t         tableStartIndex;
+  int32_t         tableEndIndex;
+  bool            hasGroupId;
+  uint64_t        groupId;
 
   SArray*     dataReaders;  // array of tsdbReaderT*
   SReadHandle readHandle;
@@ -2006,7 +2034,7 @@ typedef struct STableMergeScanInfo {
   int32_t             scanFlag;  // table scan flag to denote if it is a repeat/reverse/main scan
   int32_t             dataBlockLoadFlag;
   SInterval interval;  // if the upstream is an interval operator, the interval info is also kept here to get the time
-                       // window to check if current data block needs to be loaded.
+                                 // window to check if current data block needs to be loaded.
 
   SSampleExecInfo sample;  // sample execution info
 } STableMergeScanInfo;
@@ -2030,6 +2058,22 @@ int32_t createScanTableListInfo(STableScanPhysiNode* pTableScanNode, SReadHandle
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t createMultipleDataReaders(SQueryTableDataCond* pQueryCond, SReadHandle* pHandle, STableListInfo* pTableListInfo,
+                                  int32_t tableStartIdx, int32_t tableEndIdx, SArray* arrayReader, uint64_t queryId,
+                                  uint64_t taskId) {
+  for (int32_t i = tableStartIdx; i <= tableEndIdx; ++i) {
+    SArray* subTableList = taosArrayInit(1, sizeof(STableKeyInfo));
+    taosArrayPush(subTableList, taosArrayGet(pTableListInfo->pTableList, i));
+
+    tsdbReaderT* pReader = tsdbReaderOpen(pHandle->vnode, pQueryCond, subTableList, queryId, taskId);
+    taosArrayPush(arrayReader, &pReader);
+
+    taosArrayDestroy(subTableList);
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
 // todo refactor
 static int32_t loadDataBlockFromOneTable(SOperatorInfo* pOperator, STableMergeScanInfo* pTableScanInfo,
                                          int32_t readerIdx, SSDataBlock* pBlock, uint32_t* status) {
@@ -2216,34 +2260,32 @@ SArray* generateSortByTsInfo(int32_t order) {
   return pList;
 }
 
-static int32_t createMultipleDataReaders(SQueryTableDataCond* pQueryCond, SReadHandle* pHandle, SArray* tableList, SArray* arrayReader, uint64_t queryId,
-                                  uint64_t taskId) {
-  for (int32_t i = 0; i < taosArrayGetSize(tableList); ++i) {
-    SArray* tmp = taosArrayInit(1, sizeof(STableKeyInfo));
-    taosArrayPush(tmp, taosArrayGet(tableList, i));
-
-    tsdbReaderT* pReader = tsdbReaderOpen(pHandle->vnode, pQueryCond, tmp, queryId, taskId);
-    taosArrayPush(arrayReader, &pReader);
-
-    taosArrayDestroy(tmp);
-  }
-
-  return TSDB_CODE_SUCCESS;
-}
-
 int32_t startGroupTableMergeScan(SOperatorInfo* pOperator) {
   STableMergeScanInfo* pInfo = pOperator->info;
   SExecTaskInfo*       pTaskInfo = pOperator->pTaskInfo;
 
-  SArray* tableList = taosArrayGetP(pInfo->tableListInfo->pGroupList, pInfo->currentGroupId);
+  {
+    size_t  tableListSize = taosArrayGetSize(pInfo->tableListInfo->pTableList);
+    int32_t i = pInfo->tableStartIndex + 1;
+    for (; i < tableListSize; ++i) {
+      STableKeyInfo* tableKeyInfo = taosArrayGet(pInfo->tableListInfo->pTableList, i);
+      if (tableKeyInfo->groupId != pInfo->groupId) {
+        break;
+      }
+    }
+    pInfo->tableEndIndex = i - 1;
+  }
+
+  int32_t tableStartIdx = pInfo->tableStartIndex;
+  int32_t tableEndIdx = pInfo->tableEndIndex;
 
-  createMultipleDataReaders(&pInfo->cond, &pInfo->readHandle, tableList,
+  STableListInfo* tableListInfo = pInfo->tableListInfo;
+  createMultipleDataReaders(&pInfo->cond, &pInfo->readHandle, tableListInfo, tableStartIdx, tableEndIdx,
                             pInfo->dataReaders, pInfo->queryId, pInfo->taskId);
 
   // todo the total available buffer should be determined by total capacity of buffer of this task.
   // the additional one is reserved for merge result
-  int32_t tableLen = taosArrayGetSize(tableList);
-  pInfo->sortBufSize = pInfo->bufPageSize * ((tableLen==0?1:tableLen) + 1);
+  pInfo->sortBufSize = pInfo->bufPageSize * (tableEndIdx - tableStartIdx + 1 + 1);
   int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;
   pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize, numOfBufPage,
                                              pInfo->pSortInputBlock, pTaskInfo->id.str);
@@ -2330,43 +2372,38 @@ SSDataBlock* doTableMergeScan(SOperatorInfo* pOperator) {
   if (code != TSDB_CODE_SUCCESS) {
     longjmp(pTaskInfo->env, code);
   }
+  size_t tableListSize = taosArrayGetSize(pInfo->tableListInfo->pTableList);
+  if (!pInfo->hasGroupId) {
+    pInfo->hasGroupId = true;
 
-  if (pInfo->currentGroupId == -1) {
-    pInfo->currentGroupId++;
-    if (pInfo->currentGroupId >= taosArrayGetSize(pInfo->tableListInfo->pGroupList)) {
+    if (tableListSize == 0) {
       doSetOperatorCompleted(pOperator);
       return NULL;
     }
+    pInfo->tableStartIndex = 0;
+    pInfo->groupId = ((STableKeyInfo*)taosArrayGet(pInfo->tableListInfo->pTableList, pInfo->tableStartIndex))->groupId;
     startGroupTableMergeScan(pOperator);
   }
-  SSDataBlock* pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pOperator->resultInfo.capacity, pOperator);
-  if (pBlock != NULL) {
-    uint64_t* groupId = taosHashGet(pInfo->tableListInfo->map, &(pBlock->info.uid), sizeof(uint64_t));
-    if(groupId) pBlock->info.groupId = *groupId;
-
-    pOperator->resultInfo.totalRows += pBlock->info.rows;
-    return pBlock;
-  }
-
-  stopGroupTableMergeScan(pOperator);
-  pInfo->currentGroupId++;
-  if (pInfo->currentGroupId >= taosArrayGetSize(pInfo->tableListInfo->pGroupList)) {
-    doSetOperatorCompleted(pOperator);
-    return NULL;
-  }
-  startGroupTableMergeScan(pOperator);
-
-  pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pOperator->resultInfo.capacity, pOperator);
-  if (pBlock != NULL) {
-    uint64_t* groupId = taosHashGet(pInfo->tableListInfo->map, &(pBlock->info.uid), sizeof(uint64_t));
-    if(groupId) pBlock->info.groupId = *groupId;
-
-    pOperator->resultInfo.totalRows += pBlock->info.rows;
-    return pBlock;
+  SSDataBlock* pBlock = NULL;
+  while (pInfo->tableStartIndex < tableListSize) {
+    pBlock = getSortedTableMergeScanBlockData(pInfo->pSortHandle, pOperator->resultInfo.capacity, pOperator);
+    if (pBlock != NULL) {
+      pBlock->info.groupId = pInfo->groupId;
+      pOperator->resultInfo.totalRows += pBlock->info.rows;
+      return pBlock;
+    } else {
+      stopGroupTableMergeScan(pOperator);
+      if (pInfo->tableEndIndex >= tableListSize - 1) {
+        doSetOperatorCompleted(pOperator);
+        break;
+      }
+      pInfo->tableStartIndex = pInfo->tableEndIndex + 1;
+      pInfo->groupId =
+          ((STableKeyInfo*)taosArrayGet(pInfo->tableListInfo->pTableList, pInfo->tableStartIndex))->groupId;
+      startGroupTableMergeScan(pOperator);
+    }
   }
 
-  doSetOperatorCompleted(pOperator);
-
   return pBlock;
 }
 
@@ -2403,6 +2440,12 @@ int32_t getTableMergeScanExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExpla
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t compareTableKeyInfoByGid(const void* p1, const void* p2) {
+  const STableKeyInfo* info1 = p1;
+  const STableKeyInfo* info2 = p2;
+  return info1->groupId - info2->groupId;
+}
+
 SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, STableListInfo* pTableListInfo,
                                                 SReadHandle* readHandle, SExecTaskInfo* pTaskInfo, uint64_t queryId,
                                                 uint64_t taskId) {
@@ -2411,6 +2454,9 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
   if (pInfo == NULL || pOperator == NULL) {
     goto _error;
   }
+  if (pTableScanNode->pPartitionTags) {
+    taosArraySort(pTableListInfo->pTableList, compareTableKeyInfoByGid);
+  }
 
   SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;
 
@@ -2443,7 +2489,6 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
   pInfo->dataReaders = taosArrayInit(64, POINTER_BYTES);
   pInfo->queryId = queryId;
   pInfo->taskId = taskId;
-  pInfo->currentGroupId = -1;
 
   pInfo->sortSourceParams = taosArrayInit(64, sizeof(STableMergeScanSortSourceParam));
 

source/libs/stream/src/streamDispatch.c

@@ -115,6 +115,7 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock)
       .srcNodeId = pTask->nodeId,
       .srcTaskId = pTask->taskId,
       .pRetrieve = pRetrieve,
+      .retrieveLen = dataStrLen,
   };
 
   int32_t sz = taosArrayGetSize(pTask->childEpInfo);
@@ -146,7 +147,7 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock)
         .code = 0,
         .msgType = TDMT_STREAM_RETRIEVE,
         .pCont = buf,
-        .contLen = len,
+        .contLen = sizeof(SMsgHead) + len,
     };
 
     if (tmsgSendReq(&pEpInfo->epSet, &rpcMsg) < 0) {

source/libs/stream/src/streamExec.c

@@ -45,11 +45,16 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, void* data, SArray* pRes)
       ASSERT(false);
     }
     if (output == NULL) {
-      if (pItem->type == STREAM_INPUT__DATA_RETRIEVE && !hasData) {
-        SSDataBlock block = {0};
-        block.info.type = STREAM_PUSH_DATA;
-        block.info.childId = pTask->selfChildId;
-        taosArrayPush(pRes, &block);
+      if (pItem->type == STREAM_INPUT__DATA_RETRIEVE) {
+        //SSDataBlock block = {0};
+        //block.info.type = STREAM_PUSH_EMPTY;
+        //block.info.childId = pTask->selfChildId;
+        SStreamDataBlock* pRetrieveBlock = (SStreamDataBlock*)data;
+        ASSERT(taosArrayGetSize(pRetrieveBlock->blocks) == 1);
+        SSDataBlock* pBlock = createOneDataBlock(taosArrayGet(pRetrieveBlock->blocks, 0), true);
+        pBlock->info.type = STREAM_PUSH_EMPTY;
+        pBlock->info.childId = pTask->selfChildId;
+        taosArrayPush(pRes, pBlock);
       }
       break;
     }
@@ -109,7 +114,7 @@ static SArray* streamExecForQall(SStreamTask* pTask, SArray* pRes) {
     if (type == STREAM_INPUT__TRIGGER) {
       blockDataDestroy(((SStreamTrigger*)data)->pBlock);
       taosFreeQitem(data);
-    } else if (type == STREAM_INPUT__DATA_BLOCK) {
+    } else if (type == STREAM_INPUT__DATA_BLOCK || type == STREAM_INPUT__DATA_RETRIEVE) {
       taosArrayDestroyEx(((SStreamDataBlock*)data)->blocks, (FDelete)tDeleteSSDataBlock);
       taosFreeQitem(data);
     } else if (type == STREAM_INPUT__DATA_SUBMIT) {

source/libs/stream/src/streamUpdate.c

@@ -119,6 +119,7 @@ SUpdateInfo *updateInfoInit(int64_t interval, int32_t precision, int64_t waterma
     taosArrayPush(pInfo->pTsBuckets, &dumy);
   }
   pInfo->numBuckets = DEFAULT_BUCKET_SIZE;
+  pInfo->pCloseWinSBF = NULL;
   return pInfo;
 }
 
@@ -154,6 +155,9 @@ bool updateInfoIsUpdated(SUpdateInfo *pInfo, tb_uid_t tableId, TSKEY ts) {
   TSKEY maxTs = *(TSKEY *)taosArrayGet(pInfo->pTsBuckets, index);
   if (ts < maxTs - pInfo->watermark) {
     // this window has been closed.
+    if (pInfo->pCloseWinSBF) {
+      return tScalableBfPut(pInfo->pCloseWinSBF, &ts, sizeof(TSKEY));
+    }
     return true;
   }
 
@@ -193,3 +197,19 @@ void updateInfoDestroy(SUpdateInfo *pInfo) {
   taosArrayDestroy(pInfo->pTsSBFs);
   taosMemoryFree(pInfo);
 }
+
+void updateInfoAddCloseWindowSBF(SUpdateInfo *pInfo) {
+  if (pInfo->pCloseWinSBF) {
+    return;
+  }
+  int64_t rows = adjustExpEntries(pInfo->interval * ROWS_PER_MILLISECOND);
+  pInfo->pCloseWinSBF = tScalableBfInit(rows, DEFAULT_FALSE_POSITIVE);
+}
+
+void updateInfoDestoryColseWinSBF(SUpdateInfo *pInfo) {
+  if (!pInfo || !pInfo->pCloseWinSBF) {
+    return;
+  }
+  tScalableBfDestroy(pInfo->pCloseWinSBF);
+  pInfo->pCloseWinSBF = NULL;
+}

source/libs/sync/src/syncMain.c

@@ -914,6 +914,9 @@ void syncNodeStart(SSyncNode* pSyncNode) {
     syncNodeBecomeLeader(pSyncNode, "one replica start");
 
     // Raft 3.6.2 Committing entries from previous terms
+    syncNodeAppendNoop(pSyncNode);
+    syncMaybeAdvanceCommitIndex(pSyncNode);
+
     return;
   }
 
@@ -1662,6 +1665,12 @@ void syncNodeDoConfigChange(SSyncNode* pSyncNode, SSyncCfg* pNewConfig, SyncInde
     // change isStandBy to normal (election timeout)
     if (pSyncNode->state == TAOS_SYNC_STATE_LEADER) {
       syncNodeBecomeLeader(pSyncNode, tmpbuf);
+
+      // Raft 3.6.2 Committing entries from previous terms
+      syncNodeReplicate(pSyncNode);
+      syncNodeAppendNoop(pSyncNode);
+      syncMaybeAdvanceCommitIndex(pSyncNode);
+
     } else {
       syncNodeBecomeFollower(pSyncNode, tmpbuf);
     }
@@ -1807,16 +1816,9 @@ void syncNodeBecomeLeader(SSyncNode* pSyncNode, const char* debugStr) {
   // stop elect timer
   syncNodeStopElectTimer(pSyncNode);
 
-  // start replicate right now!
-  syncNodeReplicate(pSyncNode);
-
   // start heartbeat timer
   syncNodeStartHeartbeatTimer(pSyncNode);
 
-  // append noop
-  syncNodeAppendNoop(pSyncNode);
-  syncMaybeAdvanceCommitIndex(pSyncNode);  // maybe only one replica
-
   // trace log
   do {
     int32_t debugStrLen = strlen(debugStr);
@@ -1841,9 +1843,9 @@ void syncNodeCandidate2Leader(SSyncNode* pSyncNode) {
   syncNodeLog2("==state change syncNodeCandidate2Leader==", pSyncNode);
 
   // Raft 3.6.2 Committing entries from previous terms
-
-  // do not use this
-  // syncNodeEqNoop(pSyncNode);
+  syncNodeReplicate(pSyncNode);
+  syncNodeAppendNoop(pSyncNode);
+  syncMaybeAdvanceCommitIndex(pSyncNode);
 }
 
 void syncNodeFollower2Candidate(SSyncNode* pSyncNode) {

source/os/src/osFile.c

@@ -671,7 +671,7 @@ void taosFprintfFile(TdFilePtr pFile, const char *format, ...) {
   fflush(pFile->fp);
 }
 
-bool taosValidFile(TdFilePtr pFile) { return pFile != NULL; }
+bool taosValidFile(TdFilePtr pFile) { return pFile != NULL && pFile->fd > 0; }
 
 int32_t taosUmaskFile(int32_t maskVal) {
 #ifdef WINDOWS

tests/pytest/util/dnodes.py

@@ -508,7 +508,6 @@ class TDDnode:
 
     def stoptaosd(self):
         if (not self.remoteIP == ""):
-            print("123")
             self.remoteExec(self.cfgDict, "tdDnodes.dnodes[%d].running=1\ntdDnodes.dnodes[%d].stop()"%(self.index-1,self.index-1))
             tdLog.info("stop dnode%d"%self.index)
             return
@@ -518,18 +517,21 @@ class TDDnode:
             toBeKilled = "valgrind.bin"
 
         if self.running != 0:
-            psCmd = "ps -ef|grep -w %s| grep dnode%d|grep -v grep | awk '{print $2}'" % (toBeKilled,self.index)
-            processID = subprocess.check_output(
-                psCmd, shell=True).decode("utf-8")
-
-            while(processID):
-                killCmd = "kill -INT %s > /dev/null 2>&1" % processID
-                os.system(killCmd)
-                time.sleep(1)
+            if platform.system().lower() == 'windows':
+                os.system("wmic process where \"name='taosd.exe' and CommandLine like '%%dnode%d%%'\" get processId | xargs echo | awk '{print $2}' | xargs taskkill -f -pid"%self.index)
+            else:
+                psCmd = "ps -ef|grep -w %s| grep dnode%d|grep -v grep | awk '{print $2}'" % (toBeKilled,self.index)
                 processID = subprocess.check_output(
                     psCmd, shell=True).decode("utf-8")
-            if self.valgrind:
-                time.sleep(2)
+
+                while(processID):
+                    killCmd = "kill -INT %s > /dev/null 2>&1" % processID
+                    os.system(killCmd)
+                    time.sleep(1)
+                    processID = subprocess.check_output(
+                        psCmd, shell=True).decode("utf-8")
+                if self.valgrind:
+                    time.sleep(2)
 
             self.running = 0
             tdLog.debug("dnode:%d is stopped by kill -INT" % (self.index))

tests/script/test-all.bat

+@echo off
+SETLOCAL EnableDelayedExpansion
+for /F "tokens=1,2 delims=#" %%a in ('"prompt #$H#$E# & echo on & for %%b in (1) do     rem"') do (  set "DEL=%%a")
+set /a a=0
+echo Windows Taosd Full Test
+set /a exitNum=0
+rm -rf failed.txt
+set caseFile="jenkins\\basic.txt"
+if not "%2" == "" (
+    set caseFile="%2"
+)
+for /F "usebackq tokens=*" %%i in (!caseFile!) do (
+    set line=%%i
+    if "!line:~,9!" == "./test.sh" (
+        set /a a+=1
+        echo !a! Processing %%i
+        call :GetTimeSeconds !time!
+        set time1=!_timeTemp!
+        echo Start at !time!
+        call !line:./test.sh=wtest.bat! > result_!a!.txt 2>error_!a!.txt
+        if errorlevel 1 ( call :colorEcho 0c "failed" &echo. && set /a exitNum=8 && echo %%i >>failed.txt ) else ( call :colorEcho 0a "Success" &echo. )
+    )
+)
+exit !exitNum!
+
+:colorEcho
+set timeNow=%time%
+call :GetTimeSeconds %timeNow%
+set time2=%_timeTemp%
+set /a interTime=%time2% - %time1%
+echo End at %timeNow% , cast %interTime%s
+echo off
+<nul set /p ".=%DEL%" > "%~2"
+findstr /v /a:%1 /R "^$" "%~2" nul
+del "%~2" > nul 2>&1i
+goto :eof
+
+:GetTimeSeconds
+set tt=%1
+set tt=%tt:.= %
+set tt=%tt::= %
+set tt=%tt: 0= %
+set /a index=1
+for %%a in (%tt%) do (
+    if !index! EQU 1 (
+        set /a hh=%%a
+    )^
+    else if  !index! EQU 2 (
+        set /a mm=%%a
+ 
+    )^
+    else if  !index! EQU 3 (
+        set /a ss=%%a
+    )
+   set /a index=index+1
+)
+set /a _timeTemp=(%hh%*60+%mm%)*60+%ss%
+goto :eof

tests/script/tsim/stream/distributeInterval0.sim

@@ -80,17 +80,17 @@ endi
 
 if $data03 != 4 then
   print ======$data03
-  return -1
+  goto loop1
 endi
 
 if $data04 != 52 then
   print ======$data04
-  return -1
+  goto loop1
 endi
 
 if $data05 != 13 then
   print ======$data05
-  return -1
+  goto loop1
 endi
 
 # row 1
@@ -179,7 +179,7 @@ sql use test1;
 sql create stable st(ts timestamp, a int, b int , c int) tags(ta int,tb int,tc int);
 sql create table ts1 using st tags(1,1,1);
 sql create table ts2 using st tags(2,2,2);
-sql create stream stream_t2 trigger at_once  into streamtST1 as select  _wstartts, count(*) c1, count(a) c2 , sum(a) c3 ,  max(b)  c5, min(c) c6 from st interval(10s) ;
+sql create stream stream_t2 trigger at_once  watermark 20s into streamtST1 as select  _wstartts, count(*) c1, count(a) c2 , sum(a) c3 ,  max(b)  c5, min(c) c6 from st interval(10s) ;
 
 sql insert into ts1 values(1648791211000,1,2,3);
 sql insert into ts1 values(1648791222001,2,2,3);

tests/script/tsim/stream/partitionby.sim

@@ -74,7 +74,7 @@ sql create stable st(ts timestamp,a int,b int,c int,id int) tags(ta int,tb int,t
 sql create table ts1 using st tags(1,1,1);
 sql create table ts2 using st tags(2,2,2);
 
-sql create stream stream_t2 trigger at_once  into streamtST as select  _wstartts, count(*) c1, count(a) c2 , sum(a) c3 ,  max(b)  c5, min(c) c6, max(id) c7 from st partition by ta interval(10s) ;
+sql create stream stream_t2 trigger at_once  watermark 20s into streamtST as select  _wstartts, count(*) c1, count(a) c2 , sum(a) c3 ,  max(b)  c5, min(c) c6, max(id) c7 from st partition by ta interval(10s) ;
 sql insert into ts1 values(1648791211000,1,2,3,1);
 sql insert into ts1 values(1648791222001,2,2,3,2);
 sql insert into ts2 values(1648791211000,1,2,3,3);
@@ -83,16 +83,16 @@ sql insert into ts2 values(1648791222001,2,2,3,4);
 sql insert into ts2 values(1648791222002,2,2,3,5);
 sql insert into ts2 values(1648791222002,2,2,3,6);
 
-sql insert into ts1 values(1648791211000,1,2,3,1);
-sql insert into ts1 values(1648791222001,2,2,3,2);
-sql insert into ts2 values(1648791211000,1,2,3,3);
-sql insert into ts2 values(1648791222001,2,2,3,4);
+sql insert into ts1 values(1648791211000,1,2,3,7);
+sql insert into ts1 values(1648791222001,2,2,3,8);
+sql insert into ts2 values(1648791211000,1,2,3,9);
+sql insert into ts2 values(1648791222001,2,2,3,10);
 
 $loop_count = 0
 
 loop2:
 sleep 300
-sql select * from streamtST;
+sql select * from streamtST order by c7 asc;
 
 $loop_count = $loop_count + 1
 if $loop_count == 10 then
@@ -104,8 +104,18 @@ print =====data01=$data01
 goto loop2
 endi
 
-if $data02 != 1 then
-print =====data02=$data02
+if $data11 != 1 then
+print =====data11=$data11
+goto loop2
+endi
+
+if $data21 != 1 then
+print =====data21=$data21
+goto loop2
+endi
+
+if $data31 != 2 then
+print =====data31=$data31
 goto loop2
 endi
 
@@ -114,8 +124,18 @@ print =====data03=$data03
 goto loop2
 endi
 
-if $data04 != 2 then
-print =====data04=$data04
+if $data13 != 2 then
+print =====data13=$data13
+goto loop2
+endi
+
+if $data23 != 1 then
+print =====data23=$data23
+goto loop2
+endi
+
+if $data33 != 4 then
+print =====data33=$data33
 goto loop2
 endi
 

tests/script/tsim/stream/schedSnode.sim

@@ -79,17 +79,17 @@ endi
 
 if $data03 != 4 then
   print ======$data03
-  return -1
+  goto loop1
 endi
 
 if $data04 != 52 then
   print ======$data04
-  return -1
+  goto loop1
 endi
 
 if $data05 != 13 then
   print ======$data05
-  return -1
+  goto loop1
 endi
 
 # row 1

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

@@ -20,7 +20,7 @@ class MyDnodes(TDDnodes):
         self.simDeployed = False
         
 class TDTestCase:
-
+    noConn = True
     def init(self,conn ,logSql):
         tdLog.debug(f"start to excute {__file__}")
         self.TDDnodes = None
@@ -40,7 +40,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]
@@ -81,7 +81,7 @@ class TDTestCase:
             dnode_id = dnode.cfgDict["fqdn"] +  ":" +dnode.cfgDict["serverPort"]
             dnode_first_host = dnode.cfgDict["firstEp"].split(":")[0]
             dnode_first_port = dnode.cfgDict["firstEp"].split(":")[-1]
-            cmd = f" taos -h {dnode_first_host} -P {dnode_first_port} -s ' create dnode \"{dnode_id} \" ' ;"
+            cmd = f"{self.getBuildPath()}/build/bin/taos -h {dnode_first_host} -P {dnode_first_port} -s \"create dnode \\\"{dnode_id}\\\"\""
             print(cmd)
             os.system(cmd)
         

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

@@ -20,6 +20,7 @@ class MyDnodes(TDDnodes):
         self.simDeployed = False
         
 class TDTestCase:
+    noConn = True
 
     def init(self,conn ,logSql):
         tdLog.debug(f"start to excute {__file__}")
@@ -40,7 +41,7 @@ class TDTestCase:
             projPath = selfPath[:selfPath.find("tests")]
 
         for root, dirs, files in os.walk(projPath):
-            if ("taosd" in files):
+            if ("taosd" in files or "taosd.exe" in files):
                 rootRealPath = os.path.dirname(os.path.realpath(root))
                 if ("packaging" not in rootRealPath):
                     buildPath = root[:len(root) - len("/build/bin")]
@@ -85,7 +86,7 @@ class TDTestCase:
             dnode_id = dnode.cfgDict["fqdn"] +  ":" +dnode.cfgDict["serverPort"]
             dnode_first_host = dnode.cfgDict["firstEp"].split(":")[0]
             dnode_first_port = dnode.cfgDict["firstEp"].split(":")[-1]
-            cmd = f" taos -h {dnode_first_host} -P {dnode_first_port} -s ' create dnode \"{dnode_id} \" ' ;"
+            cmd = f"{self.getBuildPath()}/build/bin/taos -h {dnode_first_host} -P {dnode_first_port} -s \"create dnode \\\"{dnode_id}\\\"\""
             print(cmd)
             os.system(cmd)
         

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

@@ -86,7 +86,13 @@ class TMQCom:
             shellCmd += '--tool=memcheck --leak-check=full --show-reachable=no --track-origins=yes --show-leak-kinds=all --num-callers=20 -v --workaround-gcc296-bugs=yes '
         
         if (platform.system().lower() == 'windows'):
-            shellCmd = 'mintty -h never -w hide ' + buildPath + '\\build\\bin\\tmq_sim.exe -c ' + cfgPath
+            processorName = buildPath + '\\build\\bin\\tmq_sim.exe'
+            if alias != 0:
+                processorNameNew = buildPath + '\\build\\bin\\tmq_sim_new.exe'
+                shellCmd = 'cp %s %s'%(processorName, processorNameNew)
+                os.system(shellCmd)
+                processorName = processorNameNew
+            shellCmd = 'mintty -h never ' + processorName + ' -c ' + cfgPath
             shellCmd += " -y %d -d %s -g %d -r %d -w %s "%(pollDelay, dbName, showMsg, showRow, cdbName) 
             shellCmd += "> nul 2>&1 &"   
         else:

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

@@ -288,7 +288,10 @@ class TDTestCase:
             tdLog.exit("tmq consume rows error!")
 
         tdSql.query("drop topic %s"%topicFromStb1)
-        os.system('pkill tmq_sim')
+        if (platform.system().lower() == 'windows'):
+            os.system("TASKKILL /F /IM tmq_sim.exe")
+        else:
+            os.system('pkill tmq_sim')
 
         tdLog.printNoPrefix("======== test case 1 end ...... ")  
 

tests/system-test/test-all.bat

@@ -2,7 +2,7 @@
 SETLOCAL EnableDelayedExpansion
 for /F "tokens=1,2 delims=#" %%a in ('"prompt #$H#$E# & echo on & for %%b in (1) do     rem"') do (  set "DEL=%%a")
 set /a a=0
-if %1 == full (
+if "%1" == "full" (
     echo Windows Taosd Full Test
     set /a exitNum=0
     del /Q /F failed.txt

tests/system-test/test.py

@@ -23,6 +23,7 @@ import platform
 import socket
 import threading
 from distutils.log import warn as printf
+from tkinter import N
 from fabric2 import Connection
 sys.path.append("../pytest")
 from util.log import *
@@ -187,9 +188,9 @@ if __name__ == "__main__":
 
     tdLog.info("Procedures for tdengine deployed in %s" % (host))
     if platform.system().lower() == 'windows':
+        fileName = fileName.replace("/", os.sep)
         if (masterIp == "" and not fileName[0:12] == "0-others\\udf"):
             threading.Thread(target=checkRunTimeError,daemon=True).start()
-        tdCases.logSql(logSql)
         tdLog.info("Procedures for testing self-deployment")
         tdDnodes.init(deployPath, masterIp)
         tdDnodes.setTestCluster(testCluster)
@@ -208,18 +209,46 @@ if __name__ == "__main__":
             uModule = importlib.import_module(moduleName)
             try:
                 ucase = uModule.TDTestCase()
-                if ((json.dumps(updateCfgDict) == '{}') and (ucase.updatecfgDict is not None)):
+                if ((json.dumps(updateCfgDict) == '{}') and hasattr(ucase, 'updatecfgDict')):
                     updateCfgDict = ucase.updatecfgDict
                     updateCfgDictStr = "-d %s"%base64.b64encode(json.dumps(updateCfgDict).encode()).decode()
             except Exception as r:
                 print(r)
         else:
             pass
-        tdDnodes.deploy(1,updateCfgDict)
-        tdDnodes.start(1)
-        conn = taos.connect(
-            host="%s"%(host),
-            config=tdDnodes.sim.getCfgDir())
+        if dnodeNums == 1 :
+            tdDnodes.deploy(1,updateCfgDict)
+            tdDnodes.start(1)
+            tdCases.logSql(logSql)
+        else :
+            tdLog.debug("create an cluster  with %s nodes and make %s dnode as independent mnode"%(dnodeNums,mnodeNums))
+            dnodeslist = cluster.configure_cluster(dnodeNums=dnodeNums,mnodeNums=mnodeNums)
+            tdDnodes = ClusterDnodes(dnodeslist)
+            tdDnodes.init(deployPath, masterIp)
+            tdDnodes.setTestCluster(testCluster)
+            tdDnodes.setValgrind(valgrind)
+            tdDnodes.stopAll()
+            for dnode in tdDnodes.dnodes:
+                tdDnodes.deploy(dnode.index,{})
+            for dnode in tdDnodes.dnodes:
+                tdDnodes.starttaosd(dnode.index)
+            tdCases.logSql(logSql)
+            conn = taos.connect(
+                host,
+                config=tdDnodes.getSimCfgPath())
+            print(tdDnodes.getSimCfgPath(),host)
+            cluster.create_dnode(conn)
+            try:
+                if cluster.check_dnode(conn) :
+                    print("check dnode ready")
+            except Exception as r:
+                print(r)
+        if ucase is not None and hasattr(ucase, 'noConn') and ucase.noConn == True:
+            conn = None
+        else:
+            conn = taos.connect(
+                host="%s"%(host),
+                config=tdDnodes.sim.getCfgDir())
         if is_test_framework:
             tdCases.runOneWindows(conn, fileName)
         else:
@@ -307,4 +336,5 @@ if __name__ == "__main__":
                     tdCases.runOneLinux(conn, sp[0] + "_" + "restart.py")
                 else:
                     tdLog.info("not need to query")
-    conn.close()
+    if conn is not None:
+        conn.close()

tools/shell/src/shellEngine.c

@@ -156,6 +156,7 @@ void shellRunSingleCommandImp(char *command) {
     }
 
     fname = sptr + 2;
+    while (*fname == ' ') fname++;
     *sptr = '\0';
   }