Merge remote-tracking branch 'origin/3.0' into enh/TD-20288

cmake/taostools_CMakeLists.txt.in

@@ -2,7 +2,7 @@
 # taos-tools
 ExternalProject_Add(taos-tools
         GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
-        GIT_TAG 0fb640b
+        GIT_TAG a921bd4
         SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
         BINARY_DIR ""
         #BUILD_IN_SOURCE TRUE

include/common/tdataformat.h

@@ -130,7 +130,7 @@ void    tColDataInit(SColData *pColData, int16_t cid, int8_t type, int8_t smaOn)
 void    tColDataClear(SColData *pColData);
 int32_t tColDataAppendValue(SColData *pColData, SColVal *pColVal);
 void    tColDataGetValue(SColData *pColData, int32_t iVal, SColVal *pColVal);
-uint8_t tColDataGetBitValue(SColData *pColData, int32_t iVal);
+uint8_t tColDataGetBitValue(const SColData *pColData, int32_t iVal);
 int32_t tColDataCopy(SColData *pColDataSrc, SColData *pColDataDest);
 extern void (*tColDataCalcSMA[])(SColData *pColData, int64_t *sum, int64_t *max, int64_t *min, int16_t *numOfNull);
 

include/common/tmsg.h

@@ -1806,7 +1806,7 @@ int32_t tDeserializeSCMCreateTopicRsp(void* buf, int32_t bufLen, SCMCreateTopicR
 
 typedef struct {
   int64_t consumerId;
-} SMqConsumerLostMsg, SMqConsumerRecoverMsg;
+} SMqConsumerLostMsg, SMqConsumerRecoverMsg, SMqConsumerClearMsg;
 
 typedef struct {
   int64_t consumerId;

include/common/tmsgdef.h

@@ -149,7 +149,7 @@ enum {
   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_UNUSED2, "unused2", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_MND_TMQ_TIMER, "mq-tmr", SMTimerReq, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_TMQ_TIMER, "tmq-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)
   TD_DEF_MSG_TYPE(TDMT_MND_TTL_TIMER, "ttl-tmr", NULL, NULL)
@@ -171,6 +171,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_SHOW_VARIABLES, "show-variables", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_SERVER_VERSION, "server-version", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_UPTIME_TIMER, "uptime-timer", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_TMQ_LOST_CONSUMER_CLEAR, "lost-consumer-clear", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_MAX_MSG, "mnd-max", NULL, NULL)
 
   TD_NEW_MSG_SEG(TDMT_VND_MSG)

include/libs/stream/streamState.h

@@ -60,19 +60,19 @@ int32_t streamStateDel(SStreamState* pState, const SWinKey* key);
 int32_t streamStateClear(SStreamState* pState);
 void    streamStateSetNumber(SStreamState* pState, int32_t number);
 
-int32_t streamStateSessionAddIfNotExist(SStreamState* pState, SSessionKey* key, void** pVal, int32_t* pVLen);
+int32_t streamStateSessionAddIfNotExist(SStreamState* pState, SSessionKey* key, TSKEY gap, void** pVal, int32_t* pVLen);
 int32_t streamStateSessionPut(SStreamState* pState, const SSessionKey* key, const void* value, int32_t vLen);
 int32_t streamStateSessionGet(SStreamState* pState, SSessionKey* key, void** pVal, int32_t* pVLen);
 int32_t streamStateSessionDel(SStreamState* pState, const SSessionKey* key);
 int32_t streamStateSessionClear(SStreamState* pState);
-int32_t streamStateSessionGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, const void** pVal, int32_t* pVLen);
+int32_t streamStateSessionGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, void** pVal, int32_t* pVLen);
 int32_t streamStateStateAddIfNotExist(SStreamState* pState, SSessionKey* key, char* pKeyData, int32_t keyDataLen,
                                       state_key_cmpr_fn fn, void** pVal, int32_t* pVLen);
-int32_t streamStateSessionGetKey(SStreamState* pState, const SSessionKey* key, SSessionKey* curKey);
+int32_t streamStateSessionGetKeyByRange(SStreamState* pState, const SSessionKey* range, SSessionKey* curKey);
 
 SStreamStateCur* streamStateSessionSeekKeyNext(SStreamState* pState, const SSessionKey* key);
 SStreamStateCur* streamStateSessionSeekKeyCurrentPrev(SStreamState* pState, const SSessionKey* key);
-SStreamStateCur* streamStateSessionGetCur(SStreamState* pState, const SSessionKey* key);
+SStreamStateCur* streamStateSessionSeekKeyCurrentNext(SStreamState* pState, const SSessionKey* key);
 
 int32_t streamStateFillPut(SStreamState* pState, const SWinKey* key, const void* value, int32_t vLen);
 int32_t streamStateFillGet(SStreamState* pState, const SWinKey* key, void** pVal, int32_t* pVLen);
@@ -99,7 +99,9 @@ int32_t streamStateSeekLast(SStreamState* pState, SStreamStateCur* pCur);
 int32_t streamStateCurNext(SStreamState* pState, SStreamStateCur* pCur);
 int32_t streamStateCurPrev(SStreamState* pState, SStreamStateCur* pCur);
 
-// char* streamStateSessionDump(SStreamState* pState);
+#if 0
+char* streamStateSessionDump(SStreamState* pState);
+#endif
 
 #ifdef __cplusplus
 }

include/libs/sync/sync.h

@@ -139,6 +139,7 @@ typedef struct SSyncFSM {
   void (*FpReConfigCb)(const struct SSyncFSM* pFsm, const SRpcMsg* pMsg, const SReConfigCbMeta* pMeta);
   void (*FpLeaderTransferCb)(const struct SSyncFSM* pFsm, const SRpcMsg* pMsg, const SFsmCbMeta* pMeta);
   bool (*FpApplyQueueEmptyCb)(const struct SSyncFSM* pFsm);
+  int32_t (*FpApplyQueueItems)(const struct SSyncFSM* pFsm);
 
   void (*FpBecomeLeaderCb)(const struct SSyncFSM* pFsm);
   void (*FpBecomeFollowerCb)(const struct SSyncFSM* pFsm);

source/client/src/clientMain.c

@@ -990,7 +990,7 @@ void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
   // all data has returned to App already, no need to try again
   if (pResultInfo->completed) {
     // it is a local executed query, no need to do async fetch
-    if (QUERY_EXEC_MODE_LOCAL == pRequest->body.execMode) {
+    if (QUERY_EXEC_MODE_SCHEDULE != pRequest->body.execMode) {
       if (pResultInfo->localResultFetched) {
         pResultInfo->numOfRows = 0;
         pResultInfo->current = 0;

source/common/src/tdataformat.c

@@ -1557,7 +1557,7 @@ void tColDataGetValue(SColData *pColData, int32_t iVal, SColVal *pColVal) {
   tColDataGetValueImpl[pColData->flag](pColData, iVal, pColVal);
 }
 
-uint8_t tColDataGetBitValue(SColData *pColData, int32_t iVal) {
+uint8_t tColDataGetBitValue(const SColData *pColData, int32_t iVal) {
   uint8_t v;
   switch (pColData->flag) {
     case HAS_NONE:

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

@@ -191,6 +191,16 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
     dInfo("vgId:%d, replica:%d id:%d fqdn:%s port:%u", req.vgId, i, req.replicas[i].id, req.replicas[i].fqdn,
           req.replicas[i].port);
   }
+
+  SReplica *pReplica = &req.replicas[req.selfIndex];
+  if (pReplica->id != pMgmt->pData->dnodeId || pReplica->port != tsServerPort ||
+      strcmp(pReplica->fqdn, tsLocalFqdn) != 0) {
+    terrno = TSDB_CODE_INVALID_MSG;
+    dError("vgId:%d, dnodeId:%d ep:%s:%u not matched with local dnode", req.vgId, pReplica->id, pReplica->fqdn,
+           pReplica->port);
+    return -1;
+  }
+
   vmGenerateVnodeCfg(&req, &vnodeCfg);
 
   if (vmTsmaAdjustDays(&vnodeCfg, &req) < 0) {
@@ -285,6 +295,15 @@ int32_t vmProcessAlterVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
     return -1;
   }
 
+  SReplica *pReplica = &alterReq.replicas[alterReq.selfIndex];
+  if (pReplica->id != pMgmt->pData->dnodeId || pReplica->port != tsServerPort ||
+      strcmp(pReplica->fqdn, tsLocalFqdn) != 0) {
+    terrno = TSDB_CODE_INVALID_MSG;
+    dError("vgId:%d, dnodeId:%d ep:%s:%u not matched with local dnode", alterReq.vgId, pReplica->id, pReplica->fqdn,
+           pReplica->port);
+    return -1;
+  }
+
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId);
   if (pVnode == NULL) {
     dError("vgId:%d, failed to alter replica since %s", vgId, terrstr());
@@ -341,6 +360,12 @@ int32_t vmProcessDropVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   int32_t vgId = dropReq.vgId;
   dDebug("vgId:%d, start to drop vnode", vgId);
 
+  if (dropReq.dnodeId != pMgmt->pData->dnodeId) {
+    terrno = TSDB_CODE_INVALID_MSG;
+    dError("vgId:%d, dnodeId:%d not matched with local dnode", dropReq.vgId, dropReq.dnodeId);
+    return -1;
+  }
+
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId);
   if (pVnode == NULL) {
     dDebug("vgId:%d, failed to drop since %s", vgId, terrstr());

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

@@ -44,6 +44,7 @@ SSdbRaw *mndConsumerActionEncode(SMqConsumerObj *pConsumer);
 SSdbRow *mndConsumerActionDecode(SSdbRaw *pRaw);
 
 int32_t mndSetConsumerCommitLogs(SMnode *pMnode, STrans *pTrans, SMqConsumerObj *pConsumer);
+int32_t mndSetConsumerDropLogs(SMnode *pMnode, STrans *pTrans, SMqConsumerObj *pConsumer);
 
 bool mndRebTryStart();
 void mndRebEnd();

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

@@ -32,7 +32,8 @@
 #define MND_CONSUMER_VER_NUMBER   1
 #define MND_CONSUMER_RESERVE_SIZE 64
 
-#define MND_CONSUMER_LOST_HB_CNT 3
+#define MND_CONSUMER_LOST_HB_CNT          3
+#define MND_CONSUMER_LOST_CLEAR_THRESHOLD 43200
 
 static int8_t mqRebInExecCnt = 0;
 
@@ -50,6 +51,7 @@ static int32_t mndProcessAskEpReq(SRpcMsg *pMsg);
 static int32_t mndProcessMqHbReq(SRpcMsg *pMsg);
 static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg);
 static int32_t mndProcessConsumerLostMsg(SRpcMsg *pMsg);
+static int32_t mndProcessConsumerClearMsg(SRpcMsg *pMsg);
 static int32_t mndProcessConsumerRecoverMsg(SRpcMsg *pMsg);
 
 int32_t mndInitConsumer(SMnode *pMnode) {
@@ -69,6 +71,7 @@ int32_t mndInitConsumer(SMnode *pMnode) {
   mndSetMsgHandle(pMnode, TDMT_MND_TMQ_TIMER, mndProcessMqTimerMsg);
   mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CONSUMER_LOST, mndProcessConsumerLostMsg);
   mndSetMsgHandle(pMnode, TDMT_MND_TMQ_CONSUMER_RECOVER, mndProcessConsumerRecoverMsg);
+  mndSetMsgHandle(pMnode, TDMT_MND_TMQ_LOST_CONSUMER_CLEAR, mndProcessConsumerClearMsg);
 
   mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_CONSUMERS, mndRetrieveConsumer);
   mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_CONSUMERS, mndCancelGetNextConsumer);
@@ -162,6 +165,43 @@ FAIL:
   return -1;
 }
 
+static int32_t mndProcessConsumerClearMsg(SRpcMsg *pMsg) {
+  SMnode              *pMnode = pMsg->info.node;
+  SMqConsumerClearMsg *pClearMsg = pMsg->pCont;
+  SMqConsumerObj      *pConsumer = mndAcquireConsumer(pMnode, pClearMsg->consumerId);
+  if (pConsumer == NULL) {
+    return 0;
+  }
+
+  mInfo("receive consumer clear msg, consumer id %" PRId64 ", status %s", pClearMsg->consumerId,
+        mndConsumerStatusName(pConsumer->status));
+
+  if (pConsumer->status != MQ_CONSUMER_STATUS__LOST_REBD) {
+    mndReleaseConsumer(pMnode, pConsumer);
+    return -1;
+  }
+
+  SMqConsumerObj *pConsumerNew = tNewSMqConsumerObj(pConsumer->consumerId, pConsumer->cgroup);
+  pConsumerNew->updateType = CONSUMER_UPDATE__LOST;
+
+  mndReleaseConsumer(pMnode, pConsumer);
+
+  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pMsg, "clear-csm");
+  if (pTrans == NULL) goto FAIL;
+  if (mndSetConsumerDropLogs(pMnode, pTrans, pConsumerNew) != 0) goto FAIL;
+  if (mndTransPrepare(pMnode, pTrans) != 0) goto FAIL;
+
+  tDeleteSMqConsumerObj(pConsumerNew);
+  taosMemoryFree(pConsumerNew);
+  mndTransDrop(pTrans);
+  return 0;
+FAIL:
+  tDeleteSMqConsumerObj(pConsumerNew);
+  taosMemoryFree(pConsumerNew);
+  mndTransDrop(pTrans);
+  return -1;
+}
+
 static SMqRebInfo *mndGetOrCreateRebSub(SHashObj *pHash, const char *key) {
   SMqRebInfo *pRebInfo = taosHashGet(pHash, key, strlen(key) + 1);
   if (pRebInfo == NULL) {
@@ -206,15 +246,28 @@ static int32_t mndProcessMqTimerMsg(SRpcMsg *pMsg) {
       SMqConsumerLostMsg *pLostMsg = rpcMallocCont(sizeof(SMqConsumerLostMsg));
 
       pLostMsg->consumerId = pConsumer->consumerId;
-      SRpcMsg pRpcMsg = {
+      SRpcMsg rpcMsg = {
           .msgType = TDMT_MND_TMQ_CONSUMER_LOST,
           .pCont = pLostMsg,
           .contLen = sizeof(SMqConsumerLostMsg),
       };
-      tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &pRpcMsg);
+      tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
     }
-    if (status == MQ_CONSUMER_STATUS__LOST_REBD || status == MQ_CONSUMER_STATUS__READY) {
+
+    if (status == MQ_CONSUMER_STATUS__READY) {
       // do nothing
+    } else if (status == MQ_CONSUMER_STATUS__LOST_REBD) {
+      if (hbStatus > MND_CONSUMER_LOST_CLEAR_THRESHOLD) {
+        SMqConsumerClearMsg *pClearMsg = rpcMallocCont(sizeof(SMqConsumerClearMsg));
+
+        pClearMsg->consumerId = pConsumer->consumerId;
+        SRpcMsg rpcMsg = {
+            .msgType = TDMT_MND_TMQ_LOST_CONSUMER_CLEAR,
+            .pCont = pClearMsg,
+            .contLen = sizeof(SMqConsumerClearMsg),
+        };
+        tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
+      }
     } else if (status == MQ_CONSUMER_STATUS__LOST) {
       taosRLockLatch(&pConsumer->lock);
       int32_t topicNum = taosArrayGetSize(pConsumer->currentTopics);
@@ -444,6 +497,14 @@ FAIL:
   return -1;
 }
 
+int32_t mndSetConsumerDropLogs(SMnode *pMnode, STrans *pTrans, SMqConsumerObj *pConsumer) {
+  SSdbRaw *pCommitRaw = mndConsumerActionEncode(pConsumer);
+  if (pCommitRaw == NULL) return -1;
+  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
+  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED) != 0) return -1;
+  return 0;
+}
+
 int32_t mndSetConsumerCommitLogs(SMnode *pMnode, STrans *pTrans, SMqConsumerObj *pConsumer) {
   SSdbRaw *pCommitRaw = mndConsumerActionEncode(pConsumer);
   if (pCommitRaw == NULL) return -1;

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

@@ -205,8 +205,23 @@ static void mndBecomeLeader(const SSyncFSM *pFsm) {
 static bool mndApplyQueueEmpty(const SSyncFSM *pFsm) {
   SMnode *pMnode = pFsm->data;
 
-  int32_t itemSize = tmsgGetQueueSize(&pMnode->msgCb, 1, APPLY_QUEUE);
-  return (itemSize == 0);
+  if (pMnode != NULL && pMnode->msgCb.qsizeFp != NULL) {
+    int32_t itemSize = tmsgGetQueueSize(&pMnode->msgCb, 1, APPLY_QUEUE);
+    return (itemSize == 0);
+  } else {
+    return true;
+  }
+}
+
+static int32_t mndApplyQueueItems(const SSyncFSM *pFsm) {
+  SMnode *pMnode = pFsm->data;
+
+  if (pMnode != NULL && pMnode->msgCb.qsizeFp != NULL) {
+    int32_t itemSize = tmsgGetQueueSize(&pMnode->msgCb, 1, APPLY_QUEUE);
+    return itemSize;
+  } else {
+    return -1;
+  }
 }
 
 SSyncFSM *mndSyncMakeFsm(SMnode *pMnode) {
@@ -218,6 +233,7 @@ SSyncFSM *mndSyncMakeFsm(SMnode *pMnode) {
   pFsm->FpRestoreFinishCb = mndRestoreFinish;
   pFsm->FpLeaderTransferCb = NULL;
   pFsm->FpApplyQueueEmptyCb = mndApplyQueueEmpty;
+  pFsm->FpApplyQueueItems = mndApplyQueueItems;
   pFsm->FpReConfigCb = NULL;
   pFsm->FpBecomeLeaderCb = mndBecomeLeader;
   pFsm->FpBecomeFollowerCb = mndBecomeFollower;
@@ -291,7 +307,7 @@ int32_t mndSyncPropose(SMnode *pMnode, SSdbRaw *pRaw, int32_t transId) {
 
   SRpcMsg req = {.msgType = TDMT_MND_APPLY_MSG, .contLen = sdbGetRawTotalSize(pRaw)};
   if (req.contLen <= 0) return -1;
-  
+
   req.pCont = rpcMallocCont(req.contLen);
   if (req.pCont == NULL) return -1;
   memcpy(req.pCont, pRaw, req.contLen);

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

@@ -893,7 +893,7 @@ static int doBinarySearchKey(TSKEY* keyList, int num, int pos, TSKEY key, int or
   }
 }
 
-int32_t getEndPosInDataBlock(STsdbReader* pReader, SBlockData* pBlockData, SDataBlk* pBlock, int32_t pos) {
+static int32_t getEndPosInDataBlock(STsdbReader* pReader, SBlockData* pBlockData, SDataBlk* pBlock, int32_t pos) {
   // NOTE: reverse the order to find the end position in data block
   int32_t endPos = -1;
   bool    asc = ASCENDING_TRAVERSE(pReader->order);
@@ -910,6 +910,117 @@ int32_t getEndPosInDataBlock(STsdbReader* pReader, SBlockData* pBlockData, SData
   return endPos;
 }
 
+static void copyPrimaryTsCol(const SBlockData* pBlockData, SFileBlockDumpInfo* pDumpInfo, SColumnInfoData* pColData,
+                             int32_t dumpedRows, bool asc) {
+  if (asc) {
+    memcpy(pColData->pData, &pBlockData->aTSKEY[pDumpInfo->rowIndex], dumpedRows * sizeof(int64_t));
+  } else {
+    int32_t startIndex = pDumpInfo->rowIndex - dumpedRows + 1;
+    memcpy(pColData->pData, &pBlockData->aTSKEY[startIndex], dumpedRows * sizeof(int64_t));
+
+    // todo: opt perf by extract the loop
+    // reverse the array list
+    int32_t  mid = dumpedRows >> 1u;
+    int64_t* pts = (int64_t*)pColData->pData;
+    for (int32_t j = 0; j < mid; ++j) {
+      int64_t t = pts[j];
+      pts[j] = pts[dumpedRows - j - 1];
+      pts[dumpedRows - j - 1] = t;
+    }
+  }
+}
+
+// a faster version of copy procedure.
+static void copyNumericCols(const SColData* pData, SFileBlockDumpInfo* pDumpInfo, SColumnInfoData* pColData,
+                            int32_t dumpedRows, bool asc)  {
+  uint8_t* p = NULL;
+  if (asc) {
+    p = pData->pData + tDataTypes[pData->type].bytes * pDumpInfo->rowIndex;
+  } else {
+    int32_t startIndex = pDumpInfo->rowIndex - dumpedRows + 1;
+    p = pData->pData + tDataTypes[pData->type].bytes * startIndex;
+  }
+
+  int32_t step = asc? 1:-1;
+
+  // make sure it is aligned to 8bit
+  ASSERT((((uint64_t)pColData->pData) & (0x8 - 1)) == 0);
+
+  // 1. copy data in a batch model
+  memcpy(pColData->pData, p, dumpedRows * tDataTypes[pData->type].bytes);
+
+  // 2. reverse the array list in case of descending order scan data block
+  if (!asc) {
+    switch(pColData->info.type) {
+      case TSDB_DATA_TYPE_TIMESTAMP:
+      case TSDB_DATA_TYPE_DOUBLE:
+      case TSDB_DATA_TYPE_BIGINT:
+      case TSDB_DATA_TYPE_UBIGINT:
+      {
+        int32_t  mid = dumpedRows >> 1u;
+        int64_t* pts = (int64_t*)pColData->pData;
+        for (int32_t j = 0; j < mid; ++j) {
+          int64_t t = pts[j];
+          pts[j] = pts[dumpedRows - j - 1];
+          pts[dumpedRows - j - 1] = t;
+        }
+        break;
+      }
+
+      case TSDB_DATA_TYPE_BOOL:
+      case TSDB_DATA_TYPE_TINYINT:
+      case TSDB_DATA_TYPE_UTINYINT: {
+        int32_t  mid = dumpedRows >> 1u;
+        int8_t* pts = (int8_t*)pColData->pData;
+        for (int32_t j = 0; j < mid; ++j) {
+          int64_t t = pts[j];
+          pts[j] = pts[dumpedRows - j - 1];
+          pts[dumpedRows - j - 1] = t;
+        }
+        break;
+      }
+
+      case TSDB_DATA_TYPE_SMALLINT:
+      case TSDB_DATA_TYPE_USMALLINT: {
+        int32_t  mid = dumpedRows >> 1u;
+        int16_t* pts = (int16_t*)pColData->pData;
+        for (int32_t j = 0; j < mid; ++j) {
+          int64_t t = pts[j];
+          pts[j] = pts[dumpedRows - j - 1];
+          pts[dumpedRows - j - 1] = t;
+        }
+        break;
+      }
+
+      case TSDB_DATA_TYPE_FLOAT:
+      case TSDB_DATA_TYPE_INT:
+      case TSDB_DATA_TYPE_UINT: {
+        int32_t  mid = dumpedRows >> 1u;
+        int32_t* pts = (int32_t*)pColData->pData;
+        for (int32_t j = 0; j < mid; ++j) {
+          int64_t t = pts[j];
+          pts[j] = pts[dumpedRows - j - 1];
+          pts[dumpedRows - j - 1] = t;
+        }
+        break;
+      }
+    }
+  }
+
+  // 3. if the  null value exists, check items one-by-one
+  if (pData->flag != HAS_VALUE) {
+    int32_t rowIndex = 0;
+
+    for (int32_t j = pDumpInfo->rowIndex; rowIndex < dumpedRows; j += step, rowIndex++) {
+      uint8_t v = tColDataGetBitValue(pData, j);
+      if (v == 0 || v == 1) {
+        colDataSetNull_f(pColData->nullbitmap, rowIndex);
+        pColData->hasNull = true;
+      }
+    }
+  }
+}
+
 static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, STableBlockScanInfo* pBlockScanInfo) {
   SReaderStatus*  pStatus = &pReader->status;
   SDataBlockIter* pBlockIter = &pStatus->blockIter;
@@ -949,24 +1060,17 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, STableBlockScanIn
   }
 
   endIndex += step;
-  int32_t remain = asc ? (endIndex - pDumpInfo->rowIndex) : (pDumpInfo->rowIndex - endIndex);
-  if (remain > pReader->capacity) {  // output buffer check
-    remain = pReader->capacity;
+  int32_t dumpedRows = asc ? (endIndex - pDumpInfo->rowIndex) : (pDumpInfo->rowIndex - endIndex);
+  if (dumpedRows > pReader->capacity) {  // output buffer check
+    dumpedRows = pReader->capacity;
   }
 
+  int32_t i = 0;
   int32_t rowIndex = 0;
 
-  int32_t          i = 0;
   SColumnInfoData* pColData = taosArrayGet(pResBlock->pDataBlock, i);
   if (pColData->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
-    if (asc) {
-      memcpy(pColData->pData, &pBlockData->aTSKEY[pDumpInfo->rowIndex], remain * sizeof(int64_t));
-    } else {
-      for (int32_t j = pDumpInfo->rowIndex; rowIndex < remain; j += step) {
-        colDataAppendInt64(pColData, rowIndex++, &pBlockData->aTSKEY[j]);
-      }
-    }
-
+    copyPrimaryTsCol(pBlockData, pDumpInfo, pColData, dumpedRows, asc);
     i += 1;
   }
 
@@ -981,23 +1085,12 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, STableBlockScanIn
       colIndex += 1;
     } else if (pData->cid == pColData->info.colId) {
       if (pData->flag == HAS_NONE || pData->flag == HAS_NULL || pData->flag == (HAS_NULL | HAS_NONE)) {
-        colDataAppendNNULL(pColData, 0, remain);
+        colDataAppendNNULL(pColData, 0, dumpedRows);
       } else {
-        if (IS_NUMERIC_TYPE(pColData->info.type) && asc) {
-          uint8_t* p = pData->pData + tDataTypes[pData->type].bytes * pDumpInfo->rowIndex;
-          memcpy(pColData->pData, p, remain * tDataTypes[pData->type].bytes);
-
-          // null value exists, check one-by-one
-          if (pData->flag != HAS_VALUE) {
-            for (int32_t j = pDumpInfo->rowIndex; rowIndex < remain; j += step, rowIndex++) {
-              uint8_t v = tColDataGetBitValue(pData, j);
-              if (v == 0 || v == 1) {
-                colDataSetNull_f(pColData->nullbitmap, rowIndex);
-              }
-            }
-          }
-        } else {
-          for (int32_t j = pDumpInfo->rowIndex; rowIndex < remain; j += step) {
+        if (IS_MATHABLE_TYPE(pColData->info.type)) {
+          copyNumericCols(pData, pDumpInfo, pColData, dumpedRows, asc);
+        } else {  // varchar/nchar type
+          for (int32_t j = pDumpInfo->rowIndex; rowIndex < dumpedRows; j += step) {
             tColDataGetValue(pData, j, &cv);
             doCopyColVal(pColData, rowIndex++, i, &cv, pSupInfo);
           }
@@ -1007,7 +1100,7 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, STableBlockScanIn
       colIndex += 1;
       i += 1;
     } else {  // the specified column does not exist in file block, fill with null data
-      colDataAppendNNULL(pColData, 0, remain);
+      colDataAppendNNULL(pColData, 0, dumpedRows);
       i += 1;
     }
   }
@@ -1015,12 +1108,12 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, STableBlockScanIn
   // fill the mis-matched columns with null value
   while (i < numOfOutputCols) {
     pColData = taosArrayGet(pResBlock->pDataBlock, i);
-    colDataAppendNNULL(pColData, 0, remain);
+    colDataAppendNNULL(pColData, 0, dumpedRows);
     i += 1;
   }
 
-  pResBlock->info.rows = remain;
-  pDumpInfo->rowIndex += step * remain;
+  pResBlock->info.rows = dumpedRows;
+  pDumpInfo->rowIndex += step * dumpedRows;
 
   // check if current block are all handled
   if (pDumpInfo->rowIndex >= 0 && pDumpInfo->rowIndex < pBlock->nRow) {
@@ -1039,7 +1132,7 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, STableBlockScanIn
   int32_t unDumpedRows = asc ? pBlock->nRow - pDumpInfo->rowIndex : pDumpInfo->rowIndex + 1;
   tsdbDebug("%p copy file block to sdatablock, global index:%d, table index:%d, brange:%" PRId64 "-%" PRId64
             ", rows:%d, remain:%d, minVer:%" PRId64 ", maxVer:%" PRId64 ", elapsed time:%.2f ms, %s",
-            pReader, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlock->minKey.ts, pBlock->maxKey.ts, remain,
+            pReader, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlock->minKey.ts, pBlock->maxKey.ts, dumpedRows,
             unDumpedRows, pBlock->minVer, pBlock->maxVer, elapsedTime, pReader->idStr);
 
   return TSDB_CODE_SUCCESS;

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

@@ -438,8 +438,24 @@ static void vnodeBecomeLeader(const SSyncFSM *pFsm) {
 
 static bool vnodeApplyQueueEmpty(const SSyncFSM *pFsm) {
   SVnode *pVnode = pFsm->data;
-  int32_t itemSize = tmsgGetQueueSize(&pVnode->msgCb, pVnode->config.vgId, APPLY_QUEUE);
-  return (itemSize == 0);
+
+  if (pVnode != NULL && pVnode->msgCb.qsizeFp != NULL) {
+    int32_t itemSize = tmsgGetQueueSize(&pVnode->msgCb, pVnode->config.vgId, APPLY_QUEUE);
+    return (itemSize == 0);
+  } else {
+    return true;
+  }
+}
+
+static int32_t vnodeApplyQueueItems(const SSyncFSM *pFsm) {
+  SVnode *pVnode = pFsm->data;
+
+  if (pVnode != NULL && pVnode->msgCb.qsizeFp != NULL) {
+    int32_t itemSize = tmsgGetQueueSize(&pVnode->msgCb, pVnode->config.vgId, APPLY_QUEUE);
+    return itemSize;
+  } else {
+    return -1;
+  }
 }
 
 static SSyncFSM *vnodeSyncMakeFsm(SVnode *pVnode) {
@@ -452,6 +468,7 @@ static SSyncFSM *vnodeSyncMakeFsm(SVnode *pVnode) {
   pFsm->FpRestoreFinishCb = vnodeRestoreFinish;
   pFsm->FpLeaderTransferCb = NULL;
   pFsm->FpApplyQueueEmptyCb = vnodeApplyQueueEmpty;
+  pFsm->FpApplyQueueItems = vnodeApplyQueueItems;
   pFsm->FpBecomeLeaderCb = vnodeBecomeLeader;
   pFsm->FpBecomeFollowerCb = vnodeBecomeFollower;
   pFsm->FpReConfigCb = NULL;

source/libs/executor/inc/executorimpl.h

@@ -264,6 +264,7 @@ typedef struct SExchangeInfo {
   SLoadRemoteDataInfo loadInfo;
   uint64_t            self;
   SLimitInfo          limitInfo;
+  int64_t             openedTs;         // start exec time stamp
 } SExchangeInfo;
 
 typedef struct SScanInfo {

source/libs/executor/src/executorimpl.c

@@ -1846,40 +1846,41 @@ static void* setAllSourcesCompleted(SOperatorInfo* pOperator, int64_t startTs) {
   return NULL;
 }
 
-static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeInfo* pExchangeInfo,
-                                           SExecTaskInfo* pTaskInfo) {
-  int32_t code = 0;
-  int64_t startTs = taosGetTimestampUs();
-  size_t  totalSources = taosArrayGetSize(pExchangeInfo->pSources);
+
+static int32_t getCompletedSources(const SArray* pArray) {
+  size_t total = taosArrayGetSize(pArray);
 
   int32_t completed = 0;
-  for (int32_t k = 0; k < totalSources; ++k) {
-    SSourceDataInfo* p = taosArrayGet(pExchangeInfo->pSourceDataInfo, k);
+  for (int32_t k = 0; k < total; ++k) {
+    SSourceDataInfo* p = taosArrayGet(pArray, k);
     if (p->status == EX_SOURCE_DATA_EXHAUSTED) {
       completed += 1;
     }
   }
 
+  return completed;
+}
+
+static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeInfo* pExchangeInfo,
+                                           SExecTaskInfo* pTaskInfo) {
+  int32_t code = 0;
+  size_t  totalSources = taosArrayGetSize(pExchangeInfo->pSourceDataInfo);
+  int32_t completed = getCompletedSources(pExchangeInfo->pSourceDataInfo);
   if (completed == totalSources) {
-    setAllSourcesCompleted(pOperator, startTs);
+    setAllSourcesCompleted(pOperator, pExchangeInfo->openedTs);
     return;
   }
 
   while (1) {
-//    printf("1\n");
     tsem_wait(&pExchangeInfo->ready);
-//    printf("2\n");
 
     for (int32_t i = 0; i < totalSources; ++i) {
       SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, i);
       if (pDataInfo->status == EX_SOURCE_DATA_EXHAUSTED) {
-//        printf("========:%d is completed\n", i);
         continue;
       }
 
-//      printf("index:%d -  status:%d\n", i, pDataInfo->status);
       if (pDataInfo->status != EX_SOURCE_DATA_READY) {
-//        printf("-----------%d, status:%d, continue\n", i, pDataInfo->status);
         continue;
       }
 
@@ -1895,27 +1896,18 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
       SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
       if (pRsp->numOfRows == 0) {
         pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
-//        printf("%d completed, try next\n", i);
-
         qDebug("%s vgId:%d, taskId:0x%" PRIx64 " execId:%d index:%d completed, rowsOfSource:%" PRIu64
-               ", totalRows:%" PRIu64 ", completed:%d try next %d/%" PRIzu,
+               ", totalRows:%" PRIu64 ", try next %d/%" PRIzu,
                GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, i, pDataInfo->totalRows,
-               pExchangeInfo->loadInfo.totalRows, completed, i + 1, totalSources);
+               pExchangeInfo->loadInfo.totalRows, i + 1, totalSources);
         taosMemoryFreeClear(pDataInfo->pRsp);
-
-//        if (completed == totalSources) {
-//          return;
-//        } else {
-//          break;
-//        }
-         break;
+        break;
       }
 
       SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
       int32_t            index = 0;
       char*              pStart = pRetrieveRsp->data;
       while (index++ < pRetrieveRsp->numOfBlocks) {
-        printf("results, numOfBLock: %d\n", pRetrieveRsp->numOfBlocks);
         SSDataBlock* pb = createOneDataBlock(pExchangeInfo->pDummyBlock, false);
         code = extractDataBlockFromFetchRsp(pb, pStart, NULL, &pStart);
         if (code != 0) {
@@ -1926,25 +1918,16 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
         taosArrayPush(pExchangeInfo->pResultBlockList, &pb);
       }
 
-      updateLoadRemoteInfo(pLoadInfo, pRetrieveRsp->numOfRows, pRetrieveRsp->compLen, startTs, pOperator);
+      updateLoadRemoteInfo(pLoadInfo, pRetrieveRsp->numOfRows, pRetrieveRsp->compLen, pExchangeInfo->openedTs, pOperator);
 
-//      int32_t completed = 0;
       if (pRsp->completed == 1) {
         pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
-
-//        for (int32_t k = 0; k < totalSources; ++k) {
-//          SSourceDataInfo* p = taosArrayGet(pExchangeInfo->pSourceDataInfo, k);
-//          if (p->status == EX_SOURCE_DATA_EXHAUSTED) {
-//            completed += 1;
-//          }
-//        }
-
         qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64
                " execId:%d index:%d completed, blocks:%d, numOfRows:%d, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64
-               ", total:%.2f Kb, completed:%d try next %d/%" PRIzu,
+               ", total:%.2f Kb, try next %d/%" PRIzu,
                GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, i, pRsp->numOfBlocks,
                pRsp->numOfRows, pDataInfo->totalRows, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0,
-               completed, i + 1, totalSources);
+               i + 1, totalSources);
       } else {
         qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64
                " execId:%d blocks:%d, numOfRows:%d, totalRows:%" PRIu64 ", total:%.2f Kb",
@@ -1962,23 +1945,12 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
           goto _error;
         }
       }
-
-//      if (completed == totalSources) {
-//        setAllSourcesCompleted(pOperator, startTs);
-//      }
-
       return;
-    }
-
-    int32_t completed = 0;
-    for (int32_t k = 0; k < totalSources; ++k) {
-      SSourceDataInfo* p = taosArrayGet(pExchangeInfo->pSourceDataInfo, k);
-      if (p->status == EX_SOURCE_DATA_EXHAUSTED) {
-        completed += 1;
-      }
-    }
+    }  // end loop
 
-    if (completed == totalSources) {
+    int32_t complete1 = getCompletedSources(pExchangeInfo->pSourceDataInfo);
+    if (complete1 == totalSources) {
+      qDebug("all sources are completed, %s", GET_TASKID(pTaskInfo));
       return;
     }
   }
@@ -2097,6 +2069,7 @@ static int32_t prepareLoadRemoteData(SOperatorInfo* pOperator) {
     if (code != TSDB_CODE_SUCCESS) {
       return code;
     }
+    pExchangeInfo->openedTs = taosGetTimestampUs();
   }
 
   OPTR_SET_OPENED(pOperator);

source/libs/executor/src/groupoperator.c

@@ -918,6 +918,8 @@ static SSDataBlock* buildStreamPartitionResult(SOperatorInfo* pOperator) {
       blockDataDestroy(pResBlock);
     }
   }
+  taosArrayDestroy(pParInfo->rowIds);
+  pParInfo->rowIds = NULL;
   blockDataUpdateTsWindow(pDest, pInfo->tsColIndex);
   pDest->info.groupId = pParInfo->groupId;
   pOperator->resultInfo.totalRows += pDest->info.rows;
@@ -1016,6 +1018,7 @@ static void destroyStreamPartitionOperatorInfo(void* param) {
   cleanupExprSupp(&pInfo->tbnameCalSup);
   cleanupExprSupp(&pInfo->tagCalSup);
   blockDataDestroy(pInfo->pDelRes);
+  taosHashCleanup(pInfo->pPartitions);
   taosMemoryFreeClear(param);
 }
 

source/libs/executor/src/scanoperator.c

@@ -1438,7 +1438,7 @@ static int32_t generateSessionScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSr
     uint64_t groupId = getGroupIdByData(pInfo, uidCol[i], startData[i], version);
     // gap must be 0.
     SSessionKey startWin = {0};
-    getCurSessionWindow(pInfo->windowSup.pStreamAggSup, startData[i], endData[i], groupId, &startWin);
+    getCurSessionWindow(pInfo->windowSup.pStreamAggSup, startData[i], startData[i], groupId, &startWin);
     if (IS_INVALID_SESSION_WIN_KEY(startWin)) {
       // window has been closed.
       continue;

source/libs/executor/src/timewindowoperator.c

@@ -3550,7 +3550,7 @@ void getCurSessionWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endT
   pKey->win.skey = startTs;
   pKey->win.ekey = endTs;
   pKey->groupId = groupId;
-  int32_t code = streamStateSessionGetKey(pAggSup->pState, pKey, pKey);
+  int32_t code = streamStateSessionGetKeyByRange(pAggSup->pState, pKey, pKey);
   if (code != TSDB_CODE_SUCCESS) {
     SET_SESSION_WIN_KEY_INVALID(pKey);
   }
@@ -3561,10 +3561,11 @@ bool isInvalidSessionWin(SResultWindowInfo* pWinInfo) { return pWinInfo->session
 void setSessionOutputBuf(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId,
                          SResultWindowInfo* pCurWin) {
   pCurWin->sessionWin.groupId = groupId;
-  pCurWin->sessionWin.win.skey = startTs - pAggSup->gap;
-  pCurWin->sessionWin.win.ekey = endTs + pAggSup->gap;
+  pCurWin->sessionWin.win.skey = startTs;
+  pCurWin->sessionWin.win.ekey = endTs;
   int32_t size = pAggSup->resultRowSize;
-  int32_t code = streamStateSessionAddIfNotExist(pAggSup->pState, &pCurWin->sessionWin, &pCurWin->pOutputBuf, &size);
+  int32_t code =
+      streamStateSessionAddIfNotExist(pAggSup->pState, &pCurWin->sessionWin, pAggSup->gap, &pCurWin->pOutputBuf, &size);
   if (code == TSDB_CODE_SUCCESS) {
     pCurWin->isOutput = true;
   } else {
@@ -3575,7 +3576,7 @@ void setSessionOutputBuf(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endT
 
 int32_t getSessionWinBuf(SStreamAggSupporter* pAggSup, SStreamStateCur* pCur, SResultWindowInfo* pWinInfo) {
   int32_t size = 0;
-  int32_t code = streamStateSessionGetKVByCur(pCur, &pWinInfo->sessionWin, (const void**)&pWinInfo->pOutputBuf, &size);
+  int32_t code = streamStateSessionGetKVByCur(pCur, &pWinInfo->sessionWin, &pWinInfo->pOutputBuf, &size);
   if (code != TSDB_CODE_SUCCESS) {
     return code;
   }
@@ -3680,7 +3681,7 @@ SStreamStateCur* getNextSessionWinInfo(SStreamAggSupporter* pAggSup, SSHashObj*
   setSessionWinOutputInfo(pStUpdated, pNextWin);
   int32_t size = 0;
   pNextWin->sessionWin = pCurWin->sessionWin;
-  int32_t code = streamStateSessionGetKVByCur(pCur, &pNextWin->sessionWin, (const void**)&pNextWin->pOutputBuf, &size);
+  int32_t code = streamStateSessionGetKVByCur(pCur, &pNextWin->sessionWin, &pNextWin->pOutputBuf, &size);
   if (code != TSDB_CODE_SUCCESS) {
     SET_SESSION_WIN_INVALID(*pNextWin);
   }
@@ -3894,9 +3895,11 @@ static void rebuildSessionWindow(SOperatorInfo* pOperator, SArray* pWinArray, SS
       SOperatorInfo*                 pChild = taosArrayGetP(pInfo->pChildren, j);
       SStreamSessionAggOperatorInfo* pChInfo = pChild->info;
       SStreamAggSupporter*           pChAggSup = &pChInfo->streamAggSup;
-      SStreamStateCur*               pCur = streamStateSessionGetCur(pChAggSup->pState, pWinKey);
-      SResultRow*                    pResult = NULL;
-      SResultRow*                    pChResult = NULL;
+      SSessionKey                    chWinKey = *pWinKey;
+      chWinKey.win.ekey = chWinKey.win.skey;
+      SStreamStateCur* pCur = streamStateSessionSeekKeyCurrentNext(pChAggSup->pState, &chWinKey);
+      SResultRow*      pResult = NULL;
+      SResultRow*      pChResult = NULL;
       while (1) {
         SResultWindowInfo childWin = {0};
         childWin.sessionWin = *pWinKey;
@@ -4112,6 +4115,12 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
   initGroupResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
   blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
 
+#if 0
+  char* pBuf = streamStateSessionDump(pAggSup->pState);
+  qDebug("===stream===final session%s", pBuf);
+  taosMemoryFree(pBuf);
+#endif
+
   doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
   if (pInfo->pDelRes->info.rows > 0) {
     printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "final session" : "single session");
@@ -4306,6 +4315,12 @@ static SSDataBlock* doStreamSessionSemiAgg(SOperatorInfo* pOperator) {
   initGroupResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
   blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
 
+#if 0
+  char* pBuf = streamStateSessionDump(pAggSup->pState);
+  qDebug("===stream===semi session%s", pBuf);
+  taosMemoryFree(pBuf);
+#endif
+
   doBuildSessionResult(pOperator, pAggSup->pState, &pInfo->groupResInfo, pBInfo->pRes);
   if (pBInfo->pRes->info.rows > 0) {
     printDataBlock(pBInfo->pRes, "semi session");

source/libs/function/src/builtinsimpl.c

@@ -911,6 +911,7 @@ int32_t avgFunction(SqlFunctionCtx* pCtx) {
 
       case TSDB_DATA_TYPE_FLOAT: {
         float* plist = (float*)pCol->pData;
+//        float val = 0;
         for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
           if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
             continue;
@@ -918,8 +919,9 @@ int32_t avgFunction(SqlFunctionCtx* pCtx) {
 
           numOfElem += 1;
           pAvgRes->count += 1;
-          pAvgRes->sum.dsum += plist[i];
+          pAvgRes->sum.dsum  += plist[i];
         }
+//        pAvgRes->sum.dsum = val;
         break;
       }
 
@@ -1278,14 +1280,20 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
           pBuf->assign = true;
         } else {
           // ignore the equivalent data value
-          if ((*val) == pData[i]) {
-            continue;
-          }
-
-          if ((*val < pData[i]) ^ isMinFunc) {
-            *val = pData[i];
-            if (pCtx->subsidiaries.num > 0) {
-              updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+          // NOTE: An faster version to avoid one additional comparison with FPU.
+          if (isMinFunc) {  // min
+            if (*val > pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
+            }
+          } else {  // max
+            if (*val < pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
             }
           }
         }
@@ -1309,14 +1317,20 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
           pBuf->assign = true;
         } else {
           // ignore the equivalent data value
-          if ((*val) == pData[i]) {
-            continue;
-          }
-
-          if ((*val < pData[i]) ^ isMinFunc) {
-            *val = pData[i];
-            if (pCtx->subsidiaries.num > 0) {
-              updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+          // NOTE: An faster version to avoid one additional comparison with FPU.
+          if (isMinFunc) {  // min
+            if (*val > pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
+            }
+          } else {  // max
+            if (*val < pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
             }
           }
         }
@@ -1340,14 +1354,20 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
           pBuf->assign = true;
         } else {
           // ignore the equivalent data value
-          if ((*val) == pData[i]) {
-            continue;
-          }
-
-          if ((*val < pData[i]) ^ isMinFunc) {
-            *val = pData[i];
-            if (pCtx->subsidiaries.num > 0) {
-              updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+          // NOTE: An faster version to avoid one additional comparison with FPU.
+          if (isMinFunc) {  // min
+            if (*val > pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
+            }
+          } else {  // max
+            if (*val < pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
             }
           }
         }
@@ -1371,14 +1391,20 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
           pBuf->assign = true;
         } else {
           // ignore the equivalent data value
-          if ((*val) == pData[i]) {
-            continue;
-          }
-
-          if ((*val < pData[i]) ^ isMinFunc) {
-            *val = pData[i];
-            if (pCtx->subsidiaries.num > 0) {
-              updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+          // NOTE: An faster version to avoid one additional comparison with FPU.
+          if (isMinFunc) {  // min
+            if (*val > pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
+            }
+          } else {  // max
+            if (*val < pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
             }
           }
         }
@@ -1404,14 +1430,20 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
           pBuf->assign = true;
         } else {
           // ignore the equivalent data value
-          if ((*val) == pData[i]) {
-            continue;
-          }
-
-          if ((*val < pData[i]) ^ isMinFunc) {
-            *val = pData[i];
-            if (pCtx->subsidiaries.num > 0) {
-              updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+          // NOTE: An faster version to avoid one additional comparison with FPU.
+          if (isMinFunc) {  // min
+            if (*val > pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
+            }
+          } else {  // max
+            if (*val < pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
             }
           }
         }
@@ -1435,14 +1467,20 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
           pBuf->assign = true;
         } else {
           // ignore the equivalent data value
-          if ((*val) == pData[i]) {
-            continue;
-          }
-
-          if ((*val < pData[i]) ^ isMinFunc) {
-            *val = pData[i];
-            if (pCtx->subsidiaries.num > 0) {
-              updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+          // NOTE: An faster version to avoid one additional comparison with FPU.
+          if (isMinFunc) {  // min
+            if (*val > pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
+            }
+          } else {  // max
+            if (*val < pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
             }
           }
         }
@@ -1466,14 +1504,20 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
           pBuf->assign = true;
         } else {
           // ignore the equivalent data value
-          if ((*val) == pData[i]) {
-            continue;
-          }
-
-          if ((*val < pData[i]) ^ isMinFunc) {
-            *val = pData[i];
-            if (pCtx->subsidiaries.num > 0) {
-              updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+          // NOTE: An faster version to avoid one additional comparison with FPU.
+          if (isMinFunc) {  // min
+            if (*val > pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
+            }
+          } else {  // max
+            if (*val < pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
             }
           }
         }
@@ -1497,14 +1541,20 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
           pBuf->assign = true;
         } else {
           // ignore the equivalent data value
-          if ((*val) == pData[i]) {
-            continue;
-          }
-
-          if ((*val < pData[i]) ^ isMinFunc) {
-            *val = pData[i];
-            if (pCtx->subsidiaries.num > 0) {
-              updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+          // NOTE: An faster version to avoid one additional comparison with FPU.
+          if (isMinFunc) {  // min
+            if (*val > pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
+            }
+          } else {  // max
+            if (*val < pData[i]) {
+              *val = pData[i];
+              if (pCtx->subsidiaries.num > 0) {
+                updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+              }
             }
           }
         }
@@ -1529,14 +1579,20 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
         pBuf->assign = true;
       } else {
         // ignore the equivalent data value
-        if ((*val) == pData[i]) {
-          continue;
-        }
-
-        if ((*val < pData[i]) ^ isMinFunc) {
-          *val = pData[i];
-          if (pCtx->subsidiaries.num > 0) {
-            updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+        // NOTE: An faster version to avoid one additional comparison with FPU.
+        if (isMinFunc) {  // min
+          if (*val > pData[i]) {
+            *val = pData[i];
+            if (pCtx->subsidiaries.num > 0) {
+              updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+            }
+          }
+        } else {  // max
+          if (*val < pData[i]) {
+            *val = pData[i];
+            if (pCtx->subsidiaries.num > 0) {
+              updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+            }
           }
         }
       }
@@ -1559,7 +1615,7 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
         }
         pBuf->assign = true;
       } else {
-        // ignore the equivalent data value
+#if 0
         if ((*val) == pData[i]) {
           continue;
         }
@@ -1570,6 +1626,23 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
             updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
           }
         }
+#endif
+        // NOTE: An faster version to avoid one additional comparison with FPU.
+        if (isMinFunc) {  // min
+          if (*val > pData[i]) {
+            *val = pData[i];
+            if (pCtx->subsidiaries.num > 0) {
+              updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+            }
+          }
+        } else {  // max
+          if (*val < pData[i]) {
+            *val = pData[i];
+            if (pCtx->subsidiaries.num > 0) {
+              updateTupleData(pCtx, i, pCtx->pSrcBlock, &pBuf->tuplePos);
+            }
+          }
+        }
       }
 
       numOfElems += 1;
@@ -2934,6 +3007,7 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
     }
   }
 #else
+  int64_t* pts = (int64_t*) pInput->pPTS->pData;
   for (int32_t i = pInput->startRowIndex; i < pInput->startRowIndex + pInput->numOfRows; ++i) {
     if (pInputCol->hasNull && colDataIsNull(pInputCol, pInput->totalRows, i, pColAgg)) {
       continue;
@@ -2942,13 +3016,14 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
     numOfElems++;
 
     char* data = colDataGetData(pInputCol, i);
-    TSKEY cts = getRowPTs(pInput->pPTS, i);
+    TSKEY cts = pts[i];
     if (pResInfo->numOfRes == 0 || pInfo->ts > cts) {
       doSaveCurrentVal(pCtx, i, cts, pInputCol->info.type, data);
       pResInfo->numOfRes = 1;
     }
   }
 #endif
+
   if (numOfElems == 0) {
     // save selectivity value for column consisted of all null values
     firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo);
@@ -3020,6 +3095,7 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
     }
   }
 #else
+  int64_t* pts = (int64_t*)pInput->pPTS->pData;
   for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; ++i) {
     if (pInputCol->hasNull && colDataIsNull(pInputCol, pInput->totalRows, i, pColAgg)) {
       continue;
@@ -3028,15 +3104,16 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
     numOfElems++;
 
     char* data = colDataGetData(pInputCol, i);
-    TSKEY cts = getRowPTs(pInput->pPTS, i);
+    TSKEY cts = pts[i];
     if (pResInfo->numOfRes == 0 || pInfo->ts < cts) {
       doSaveCurrentVal(pCtx, i, cts, type, data);
       pResInfo->numOfRes = 1;
     }
   }
 #endif
+
+  // save selectivity value for column consisted of all null values
   if (numOfElems == 0) {
-    // save selectivity value for column consisted of all null values
     firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo);
   }
   SET_VAL(pResInfo, numOfElems, 1);
@@ -3216,11 +3293,13 @@ int32_t lastRowFunction(SqlFunctionCtx* pCtx) {
     }
   }
 #else
+
+  int64_t* pts = (int64_t*)pInput->pPTS->pData;
   for (int32_t i = pInput->startRowIndex; i < pInput->numOfRows + pInput->startRowIndex; ++i) {
     char* data = colDataGetData(pInputCol, i);
-    TSKEY cts = getRowPTs(pInput->pPTS, i);
-    numOfElems++;
+    TSKEY cts = pts[i];
 
+    numOfElems++;
     if (pResInfo->numOfRes == 0 || pInfo->ts < cts) {
       doSaveLastrow(pCtx, data, i, cts, pInfo);
       pResInfo->numOfRes = 1;

source/libs/function/src/tudf.c

@@ -131,7 +131,8 @@ static int32_t udfSpawnUdfd(SUdfdData *pData) {
 
   char   udfdPathLdLib[1024] = {0};
   size_t udfdLdLibPathLen = strlen(tsUdfdLdLibPath);
-  strncpy(udfdPathLdLib, tsUdfdLdLibPath, udfdLdLibPathLen);
+  strncpy(udfdPathLdLib, tsUdfdLdLibPath, tListLen(udfdPathLdLib));
+
   udfdPathLdLib[udfdLdLibPathLen] = ':';
   strncpy(udfdPathLdLib + udfdLdLibPathLen + 1, pathTaosdLdLib, sizeof(udfdPathLdLib) - udfdLdLibPathLen - 1);
   if (udfdLdLibPathLen + taosdLdLibPathLen < 1024) {

source/libs/parser/src/parTranslater.c

@@ -2393,6 +2393,9 @@ static int32_t translateTable(STranslateContext* pCxt, SNode* pTable) {
         if (TSDB_SUPER_TABLE == pRealTable->pMeta->tableType) {
           pCxt->stableQuery = true;
         }
+        if (TSDB_SYSTEM_TABLE == pRealTable->pMeta->tableType && isSelectStmt(pCxt->pCurrStmt)) {
+          ((SSelectStmt*)pCxt->pCurrStmt)->isTimeLineResult = false;
+        }
         code = addNamespace(pCxt, pRealTable);
       }
       break;
@@ -3428,6 +3431,19 @@ static SNode* createSetOperProject(const char* pTableAlias, SNode* pNode) {
   return (SNode*)pCol;
 }
 
+// 0 means equal, 1 means the left shall prevail, -1 means the right shall prevail
+static int32_t dataTypeComp(const SDataType* l, const SDataType* r) {
+  if (l->type != r->type) {
+    return 1;
+  }
+
+  if (l->bytes != r->bytes) {
+    return l->bytes > r->bytes ? 1 : -1;
+  }
+
+  return (l->precision == r->precision && l->scale == r->scale) ? 0 : 1;
+}
+
 static int32_t translateSetOperProject(STranslateContext* pCxt, SSetOperator* pSetOperator) {
   SNodeList* pLeftProjections = getProjectList(pSetOperator->pLeft);
   SNodeList* pRightProjections = getProjectList(pSetOperator->pRight);
@@ -3440,7 +3456,8 @@ static int32_t translateSetOperProject(STranslateContext* pCxt, SSetOperator* pS
   FORBOTH(pLeft, pLeftProjections, pRight, pRightProjections) {
     SExprNode* pLeftExpr = (SExprNode*)pLeft;
     SExprNode* pRightExpr = (SExprNode*)pRight;
-    if (!dataTypeEqual(&pLeftExpr->resType, &pRightExpr->resType)) {
+    int32_t    comp = dataTypeComp(&pLeftExpr->resType, &pRightExpr->resType);
+    if (comp > 0) {
       SNode*  pRightFunc = NULL;
       int32_t code = createCastFunc(pCxt, pRight, pLeftExpr->resType, &pRightFunc);
       if (TSDB_CODE_SUCCESS != code) {
@@ -3448,9 +3465,20 @@ static int32_t translateSetOperProject(STranslateContext* pCxt, SSetOperator* pS
       }
       REPLACE_LIST2_NODE(pRightFunc);
       pRightExpr = (SExprNode*)pRightFunc;
-    }
-    strcpy(pRightExpr->aliasName, pLeftExpr->aliasName);
-    pRightExpr->aliasName[strlen(pLeftExpr->aliasName)] = '\0';
+    } else if (comp < 0) {
+      SNode*  pLeftFunc = NULL;
+      int32_t code = createCastFunc(pCxt, pLeft, pRightExpr->resType, &pLeftFunc);
+      if (TSDB_CODE_SUCCESS != code) {
+        return code;
+      }
+      REPLACE_LIST1_NODE(pLeftFunc);
+      SExprNode* pLeftFuncExpr = (SExprNode*)pLeftFunc;
+      snprintf(pLeftFuncExpr->aliasName, sizeof(pLeftFuncExpr->aliasName), "%s", pLeftExpr->aliasName);
+      snprintf(pLeftFuncExpr->userAlias, sizeof(pLeftFuncExpr->userAlias), "%s", pLeftExpr->userAlias);
+      pLeft = pLeftFunc;
+      pLeftExpr = pLeftFuncExpr;
+    }
+    snprintf(pRightExpr->aliasName, sizeof(pRightExpr->aliasName), "%s", pLeftExpr->aliasName);
     if (TSDB_CODE_SUCCESS != nodesListMakeStrictAppend(&pSetOperator->pProjectionList,
                                                        createSetOperProject(pSetOperator->stmtName, pLeft))) {
       return TSDB_CODE_OUT_OF_MEMORY;

source/libs/parser/test/parSelectTest.cpp

@@ -425,6 +425,8 @@ TEST_F(ParserSelectTest, informationSchema) {
   run("SELECT * FROM ins_databases WHERE name = 'information_schema'");
 
   run("SELECT * FROM ins_tags WHERE db_name = 'test' and table_name = 'st1'");
+
+  run("SELECT * FROM (SELECT table_name FROM ins_tables) t WHERE table_name = 'a'");
 }
 
 TEST_F(ParserSelectTest, withoutFrom) {

source/libs/stream/src/streamState.c

@@ -29,7 +29,7 @@ typedef struct SStateSessionKey {
   int64_t     opNum;
 } SStateSessionKey;
 
-static inline int sessionKeyCmpr(const SSessionKey* pWin1, const SSessionKey* pWin2) {
+static inline int sessionRangeKeyCmpr(const SSessionKey* pWin1, const SSessionKey* pWin2) {
   if (pWin1->groupId > pWin2->groupId) {
     return 1;
   } else if (pWin1->groupId < pWin2->groupId) {
@@ -45,6 +45,28 @@ static inline int sessionKeyCmpr(const SSessionKey* pWin1, const SSessionKey* pW
   return 0;
 }
 
+static inline int sessionWinKeyCmpr(const SSessionKey* pWin1, const SSessionKey* pWin2) {
+  if (pWin1->groupId > pWin2->groupId) {
+    return 1;
+  } else if (pWin1->groupId < pWin2->groupId) {
+    return -1;
+  }
+
+  if (pWin1->win.skey > pWin2->win.skey) {
+    return 1;
+  } else if (pWin1->win.skey < pWin2->win.skey) {
+    return -1;
+  }
+
+  if (pWin1->win.ekey > pWin2->win.ekey) {
+    return 1;
+  } else if (pWin1->win.ekey < pWin2->win.ekey) {
+    return -1;
+  }
+
+  return 0;
+}
+
 static inline int stateSessionKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
   SStateSessionKey* pWin1 = (SStateSessionKey*)pKey1;
   SStateSessionKey* pWin2 = (SStateSessionKey*)pKey2;
@@ -55,7 +77,7 @@ static inline int stateSessionKeyCmpr(const void* pKey1, int kLen1, const void*
     return -1;
   }
 
-  return sessionKeyCmpr(&pWin1->key, &pWin2->key);
+  return sessionWinKeyCmpr(&pWin1->key, &pWin2->key);
 }
 
 static inline int stateKeyCmpr(const void* pKey1, int kLen1, const void* pKey2, int kLen2) {
@@ -400,7 +422,6 @@ SStreamStateCur* streamStateSeekKeyNext(SStreamState* pState, const SWinKey* key
   SStateKey sKey = {.key = *key, .opNum = pState->number};
   int32_t   c = 0;
   if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateKey), &c) < 0) {
-    tdbTbcClose(pCur->pCur);
     streamStateFreeCur(pCur);
     return NULL;
   }
@@ -426,7 +447,6 @@ SStreamStateCur* streamStateFillSeekKeyNext(SStreamState* pState, const SWinKey*
 
   int32_t c = 0;
   if (tdbTbcMoveTo(pCur->pCur, key, sizeof(SWinKey), &c) < 0) {
-    tdbTbcClose(pCur->pCur);
     streamStateFreeCur(pCur);
     return NULL;
   }
@@ -452,7 +472,6 @@ SStreamStateCur* streamStateFillSeekKeyPrev(SStreamState* pState, const SWinKey*
 
   int32_t c = 0;
   if (tdbTbcMoveTo(pCur->pCur, key, sizeof(SWinKey), &c) < 0) {
-    tdbTbcClose(pCur->pCur);
     streamStateFreeCur(pCur);
     return NULL;
   }
@@ -496,33 +515,18 @@ int32_t streamStateSessionPut(SStreamState* pState, const SSessionKey* key, cons
   return tdbTbUpsert(pState->pSessionStateDb, &sKey, sizeof(SStateSessionKey), value, vLen, &pState->txn);
 }
 
-SStreamStateCur* streamStateSessionGetRanomCur(SStreamState* pState, const SSessionKey* key) {
-  SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
-  if (pCur == NULL) return NULL;
-  tdbTbcOpen(pState->pSessionStateDb, &pCur->pCur, NULL);
-
-  int32_t          c = -2;
-  SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
-  tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateSessionKey), &c);
-  if (c != 0) {
-    streamStateFreeCur(pCur);
-    return NULL;
-  }
-  pCur->number = pState->number;
-  return pCur;
-}
-
 int32_t streamStateSessionGet(SStreamState* pState, SSessionKey* key, void** pVal, int32_t* pVLen) {
-  SStreamStateCur* pCur = streamStateSessionGetRanomCur(pState, key);
+  SStreamStateCur* pCur = streamStateSessionSeekKeyCurrentNext(pState, key);
+  SSessionKey      resKey = *key;
   void*            tmp = NULL;
-  if (streamStateSessionGetKVByCur(pCur, key, (const void**)&tmp, pVLen) == 0) {
+  int32_t          code = streamStateSessionGetKVByCur(pCur, &resKey, &tmp, pVLen);
+  if (code == 0) {
+    *key = resKey;
     *pVal = tdbRealloc(NULL, *pVLen);
     memcpy(*pVal, tmp, *pVLen);
-    streamStateFreeCur(pCur);
-    return 0;
   }
   streamStateFreeCur(pCur);
-  return -1;
+  return code;
 }
 
 int32_t streamStateSessionDel(SStreamState* pState, const SSessionKey* key) {
@@ -544,7 +548,6 @@ SStreamStateCur* streamStateSessionSeekKeyCurrentPrev(SStreamState* pState, cons
   SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
   int32_t          c = 0;
   if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateSessionKey), &c) < 0) {
-    tdbTbcClose(pCur->pCur);
     streamStateFreeCur(pCur);
     return NULL;
   }
@@ -558,6 +561,34 @@ SStreamStateCur* streamStateSessionSeekKeyCurrentPrev(SStreamState* pState, cons
   return pCur;
 }
 
+SStreamStateCur* streamStateSessionSeekKeyCurrentNext(SStreamState* pState, const SSessionKey* key) {
+  SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
+  if (pCur == NULL) {
+    return NULL;
+  }
+  pCur->number = pState->number;
+  if (tdbTbcOpen(pState->pSessionStateDb, &pCur->pCur, NULL) < 0) {
+    streamStateFreeCur(pCur);
+    return NULL;
+  }
+
+  SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
+  int32_t          c = 0;
+  if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateSessionKey), &c) < 0) {
+    streamStateFreeCur(pCur);
+    return NULL;
+  }
+
+  if (c <= 0) return pCur;
+
+  if (tdbTbcMoveToNext(pCur->pCur) < 0) {
+    streamStateFreeCur(pCur);
+    return NULL;
+  }
+
+  return pCur;
+}
+
 SStreamStateCur* streamStateSessionSeekKeyNext(SStreamState* pState, const SSessionKey* key) {
   SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
   if (pCur == NULL) {
@@ -572,7 +603,6 @@ SStreamStateCur* streamStateSessionSeekKeyNext(SStreamState* pState, const SSess
   SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
   int32_t          c = 0;
   if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateSessionKey), &c) < 0) {
-    tdbTbcClose(pCur->pCur);
     streamStateFreeCur(pCur);
     return NULL;
   }
@@ -586,13 +616,13 @@ SStreamStateCur* streamStateSessionSeekKeyNext(SStreamState* pState, const SSess
   return pCur;
 }
 
-int32_t streamStateSessionGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, const void** pVal, int32_t* pVLen) {
+int32_t streamStateSessionGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, void** pVal, int32_t* pVLen) {
   if (!pCur) {
     return -1;
   }
-  const SStateSessionKey* pKTmp = NULL;
-  int32_t                 kLen;
-  if (tdbTbcGet(pCur->pCur, (const void**)&pKTmp, &kLen, pVal, pVLen) < 0) {
+  SStateSessionKey* pKTmp = NULL;
+  int32_t           kLen;
+  if (tdbTbcGet(pCur->pCur, (const void**)&pKTmp, &kLen, (const void**)pVal, pVLen) < 0) {
     return -1;
   }
   if (pKTmp->opNum != pCur->number) {
@@ -606,15 +636,15 @@ int32_t streamStateSessionGetKVByCur(SStreamStateCur* pCur, SSessionKey* pKey, c
 }
 
 int32_t streamStateSessionClear(SStreamState* pState) {
-  SSessionKey key = {.win.skey = 0, .win.ekey = 0, .groupId = 0};
-  streamStateSessionPut(pState, &key, NULL, 0);
-  SStreamStateCur* pCur = streamStateSessionSeekKeyNext(pState, &key);
+  SSessionKey      key = {.win.skey = 0, .win.ekey = 0, .groupId = 0};
+  SStreamStateCur* pCur = streamStateSessionSeekKeyCurrentNext(pState, &key);
   while (1) {
     SSessionKey delKey = {0};
     void*       buf = NULL;
     int32_t     size = 0;
-    int32_t     code = streamStateSessionGetKVByCur(pCur, &delKey, buf, &size);
+    int32_t     code = streamStateSessionGetKVByCur(pCur, &delKey, &buf, &size);
     if (code == 0) {
+      ASSERT(size > 0);
       memset(buf, 0, size);
       streamStateSessionPut(pState, &delKey, buf, size);
     } else {
@@ -623,61 +653,104 @@ int32_t streamStateSessionClear(SStreamState* pState) {
     streamStateCurNext(pState, pCur);
   }
   streamStateFreeCur(pCur);
-  streamStateSessionDel(pState, &key);
   return 0;
 }
 
-SStreamStateCur* streamStateSessionGetCur(SStreamState* pState, const SSessionKey* key) {
-  SStreamStateCur* pCur = streamStateSessionGetRanomCur(pState, key);
-  SSessionKey      resKey = *key;
-  while (1) {
-    streamStateCurPrev(pState, pCur);
-    SSessionKey tmpKey = *key;
-    int32_t     code = streamStateSessionGetKVByCur(pCur, &tmpKey, NULL, 0);
-    if (code == 0 && sessionKeyCmpr(key, &tmpKey) == 0) {
-      resKey = tmpKey;
-    } else {
-      break;
-    }
+int32_t streamStateSessionGetKeyByRange(SStreamState* pState, const SSessionKey* key, SSessionKey* curKey) {
+  SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
+  if (pCur == NULL) {
+    return -1;
+  }
+  pCur->number = pState->number;
+  if (tdbTbcOpen(pState->pSessionStateDb, &pCur->pCur, NULL) < 0) {
+    streamStateFreeCur(pCur);
+    return -1;
   }
-  streamStateFreeCur(pCur);
-  return streamStateSessionGetRanomCur(pState, &resKey);
-}
 
-int32_t streamStateSessionGetKey(SStreamState* pState, const SSessionKey* key, SSessionKey* curKey) {
-  SStreamStateCur* pCur = streamStateSessionGetRanomCur(pState, key);
-  SSessionKey      resKey = *key;
-  int32_t          res = -1;
-  while (1) {
-    SSessionKey tmpKey = *key;
-    int32_t     code = streamStateSessionGetKVByCur(pCur, &tmpKey, NULL, 0);
-    if (code == 0 && sessionKeyCmpr(key, &tmpKey) == 0) {
-      res = 0;
-      resKey = tmpKey;
-      streamStateCurPrev(pState, pCur);
-    } else {
-      break;
+  SStateSessionKey sKey = {.key = *key, .opNum = pState->number};
+  int32_t          c = 0;
+  if (tdbTbcMoveTo(pCur->pCur, &sKey, sizeof(SStateSessionKey), &c) < 0) {
+    streamStateFreeCur(pCur);
+    return -1;
+  }
+
+  SSessionKey resKey = *key;
+  int32_t     code = streamStateSessionGetKVByCur(pCur, &resKey, NULL, 0);
+  if (code == 0 && sessionRangeKeyCmpr(key, &resKey) == 0) {
+    *curKey = resKey;
+    streamStateFreeCur(pCur);
+    return code;
+  }
+
+  if (c > 0) {
+    streamStateCurNext(pState, pCur);
+    code = streamStateSessionGetKVByCur(pCur, &resKey, NULL, 0);
+    if (code == 0 && sessionRangeKeyCmpr(key, &resKey) == 0) {
+      *curKey = resKey;
+      streamStateFreeCur(pCur);
+      return code;
+    }
+  } else if (c < 0) {
+    streamStateCurPrev(pState, pCur);
+    code = streamStateSessionGetKVByCur(pCur, &resKey, NULL, 0);
+    if (code == 0 && sessionRangeKeyCmpr(key, &resKey) == 0) {
+      *curKey = resKey;
+      streamStateFreeCur(pCur);
+      return code;
     }
   }
-  *curKey = resKey;
+
   streamStateFreeCur(pCur);
-  return res;
+  return -1;
 }
 
-int32_t streamStateSessionAddIfNotExist(SStreamState* pState, SSessionKey* key, void** pVal, int32_t* pVLen) {
+int32_t streamStateSessionAddIfNotExist(SStreamState* pState, SSessionKey* key, TSKEY gap, void** pVal,
+                                        int32_t* pVLen) {
   // todo refactor
-  SStreamStateCur* pCur = streamStateSessionGetRanomCur(pState, key);
-  int32_t          size = *pVLen;
-  void*            tmp = NULL;
-  *pVal = tdbRealloc(NULL, size);
-  memset(*pVal, 0, size);
-  if (streamStateSessionGetKVByCur(pCur, key, (const void**)&tmp, pVLen) == 0) {
-    memcpy(*pVal, tmp, *pVLen);
+  int32_t     res = 0;
+  SSessionKey originKey = *key;
+  SSessionKey searchKey = *key;
+  searchKey.win.skey = key->win.skey - gap;
+  searchKey.win.ekey = key->win.ekey + gap;
+  int32_t valSize = *pVLen;
+  void*   tmp = tdbRealloc(NULL, valSize);
+  if (!tmp) {
+    return -1;
+  }
+
+  SStreamStateCur* pCur = streamStateSessionSeekKeyCurrentPrev(pState, key);
+  int32_t          code = streamStateSessionGetKVByCur(pCur, key, pVal, pVLen);
+  if (code == 0) {
+    if (sessionRangeKeyCmpr(&searchKey, key) == 0) {
+      memcpy(tmp, *pVal, valSize);
+      streamStateSessionDel(pState, key);
+      goto _end;
+    }
+    streamStateCurNext(pState, pCur);
+  } else {
+    *key = originKey;
     streamStateFreeCur(pCur);
-    return 0;
+    pCur = streamStateSessionSeekKeyNext(pState, key);
   }
+
+  code = streamStateSessionGetKVByCur(pCur, key, pVal, pVLen);
+  if (code == 0) {
+    if (sessionRangeKeyCmpr(&searchKey, key) == 0) {
+      memcpy(tmp, *pVal, valSize);
+      streamStateSessionDel(pState, key);
+      goto _end;
+    }
+  }
+
+  *key = originKey;
+  res = 1;
+  memset(tmp, 0, valSize);
+
+_end:
+
+  *pVal = tmp;
   streamStateFreeCur(pCur);
-  return 1;
+  return res;
 }
 
 int32_t streamStateStateAddIfNotExist(SStreamState* pState, SSessionKey* key, char* pKeyData, int32_t keyDataLen,
@@ -692,16 +765,18 @@ int32_t streamStateStateAddIfNotExist(SStreamState* pState, SSessionKey* key, ch
   }
 
   SStreamStateCur* pCur = streamStateSessionSeekKeyCurrentPrev(pState, key);
-  int32_t          code = streamStateSessionGetKVByCur(pCur, key, (const void**)pVal, pVLen);
+  int32_t          code = streamStateSessionGetKVByCur(pCur, key, pVal, pVLen);
   if (code == 0) {
     if (key->win.skey <= tmpKey.win.skey && tmpKey.win.ekey <= key->win.ekey) {
       memcpy(tmp, *pVal, valSize);
+      streamStateSessionDel(pState, key);
       goto _end;
     }
 
     void* stateKey = (char*)(*pVal) + (valSize - keyDataLen);
     if (fn(pKeyData, stateKey) == true) {
       memcpy(tmp, *pVal, valSize);
+      streamStateSessionDel(pState, key);
       goto _end;
     }
 
@@ -712,11 +787,12 @@ int32_t streamStateStateAddIfNotExist(SStreamState* pState, SSessionKey* key, ch
     pCur = streamStateSessionSeekKeyNext(pState, key);
   }
 
-  code = streamStateSessionGetKVByCur(pCur, key, (const void**)pVal, pVLen);
+  code = streamStateSessionGetKVByCur(pCur, key, pVal, pVLen);
   if (code == 0) {
     void* stateKey = (char*)(*pVal) + (valSize - keyDataLen);
     if (fn(pKeyData, stateKey) == true) {
       memcpy(tmp, *pVal, valSize);
+      streamStateSessionDel(pState, key);
       goto _end;
     }
   }
@@ -746,8 +822,11 @@ char* streamStateSessionDump(SStreamState* pState) {
   tdbTbcMoveToFirst(pCur->pCur);
 
   SSessionKey key = {0};
-  int32_t     code = streamStateSessionGetKVByCur(pCur, &key, NULL, 0);
+  void*       buf = NULL;
+  int32_t     bufSize = 0;
+  int32_t     code = streamStateSessionGetKVByCur(pCur, &key, &buf, &bufSize);
   if (code != 0) {
+    streamStateFreeCur(pCur);
     return NULL;
   }
 
@@ -762,12 +841,14 @@ char* streamStateSessionDump(SStreamState* pState) {
     key = (SSessionKey){0};
     code = streamStateSessionGetKVByCur(pCur, &key, NULL, 0);
     if (code != 0) {
+      streamStateFreeCur(pCur);
       return dumpBuf;
     }
     len += snprintf(dumpBuf + len, size - len, "||s:%15" PRId64 ",", key.win.skey);
     len += snprintf(dumpBuf + len, size - len, "e:%15" PRId64 ",", key.win.ekey);
     len += snprintf(dumpBuf + len, size - len, "g:%15" PRId64 "||", key.groupId);
   }
+  streamStateFreeCur(pCur);
   return dumpBuf;
 }
 #endif

source/libs/sync/inc/syncRespMgr.h

@@ -41,13 +41,12 @@ typedef struct SSyncRespMgr {
 
 SSyncRespMgr *syncRespMgrCreate(void *data, int64_t ttl);
 void          syncRespMgrDestroy(SSyncRespMgr *pObj);
-int64_t       syncRespMgrAdd(SSyncRespMgr *pObj, SRespStub *pStub);
-int32_t       syncRespMgrDel(SSyncRespMgr *pObj, uint64_t index);
-int32_t       syncRespMgrGet(SSyncRespMgr *pObj, uint64_t index, SRespStub *pStub);
-int32_t       syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t index, SRespStub *pStub);
+uint64_t      syncRespMgrAdd(SSyncRespMgr *pObj, const SRespStub *pStub);
+int32_t       syncRespMgrDel(SSyncRespMgr *pObj, uint64_t seq);
+int32_t       syncRespMgrGet(SSyncRespMgr *pObj, uint64_t seq, SRespStub *pStub);
+int32_t       syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t seq, SRpcHandleInfo *pInfo);
 void          syncRespClean(SSyncRespMgr *pObj);
 void          syncRespCleanRsp(SSyncRespMgr *pObj);
-void          syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp);
 
 #ifdef __cplusplus
 }

source/libs/sync/inc/syncTools.h

@@ -26,9 +26,6 @@ typedef struct SRaftId {
   SyncGroupId vgId;
 } SRaftId;
 
-// for compatibility, the same as syncPropose
-int32_t syncForwardToPeer(int64_t rid, SRpcMsg* pMsg, bool isWeak);
-
 // ------------------ for debug -------------------
 void syncRpcMsgPrint(SRpcMsg* pMsg);
 void syncRpcMsgPrint2(char* s, SRpcMsg* pMsg);

source/libs/sync/src/syncAppendEntries.c

@@ -246,7 +246,8 @@ int32_t syncNodeOnAppendEntries(SSyncNode* ths, SyncAppendEntries* pMsg) {
       } else {
         // error
         char logBuf[128];
-        snprintf(logBuf, sizeof(logBuf), "ignore, get local entry error, append-index:%" PRId64, appendIndex);
+        snprintf(logBuf, sizeof(logBuf), "ignore, get local entry error, append-index:%" PRId64 " err:%d", appendIndex,
+                 terrno);
         syncLogRecvAppendEntries(ths, pMsg, logBuf);
 
         syncEntryDestory(pLocalEntry);

source/libs/sync/src/syncRespMgr.c

@@ -13,21 +13,22 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#define _DEFAULT_SOURCE
 #include "syncRespMgr.h"
 #include "syncRaftEntry.h"
 #include "syncRaftStore.h"
 
 SSyncRespMgr *syncRespMgrCreate(void *data, int64_t ttl) {
-  SSyncRespMgr *pObj = (SSyncRespMgr *)taosMemoryMalloc(sizeof(SSyncRespMgr));
+  SSyncRespMgr *pObj = taosMemoryCalloc(1, sizeof(SSyncRespMgr));
   if (pObj == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return NULL;
   }
-  memset(pObj, 0, sizeof(SSyncRespMgr));
 
   pObj->pRespHash =
       taosHashInit(sizeof(uint64_t), taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
-  ASSERT(pObj->pRespHash != NULL);
+  if (pObj->pRespHash == NULL) return NULL;
+
   pObj->ttl = ttl;
   pObj->data = data;
   pObj->seqNum = 0;
@@ -38,93 +39,84 @@ SSyncRespMgr *syncRespMgrCreate(void *data, int64_t ttl) {
 
 void syncRespMgrDestroy(SSyncRespMgr *pObj) {
   if (pObj != NULL) {
-    taosThreadMutexLock(&(pObj->mutex));
+    taosThreadMutexLock(&pObj->mutex);
     taosHashCleanup(pObj->pRespHash);
-    taosThreadMutexUnlock(&(pObj->mutex));
+    taosThreadMutexUnlock(&pObj->mutex);
     taosThreadMutexDestroy(&(pObj->mutex));
     taosMemoryFree(pObj);
   }
 }
 
-int64_t syncRespMgrAdd(SSyncRespMgr *pObj, SRespStub *pStub) {
-  taosThreadMutexLock(&(pObj->mutex));
+uint64_t syncRespMgrAdd(SSyncRespMgr *pObj, const SRespStub *pStub) {
+  taosThreadMutexLock(&pObj->mutex);
 
-  uint64_t keyCode = ++(pObj->seqNum);
-  taosHashPut(pObj->pRespHash, &keyCode, sizeof(keyCode), pStub, sizeof(SRespStub));
+  uint64_t seq = ++(pObj->seqNum);
+  int32_t  code = taosHashPut(pObj->pRespHash, &seq, sizeof(uint64_t), pStub, sizeof(SRespStub));
+  sNTrace(pObj->data, "save message handle:%p, type:%s seq:%" PRIu64 " code:0x%x", pStub->rpcMsg.info.handle,
+          TMSG_INFO(pStub->rpcMsg.msgType), seq, code);
 
-  sNTrace(pObj->data, "save message handle, type:%s seq:%" PRIu64 " handle:%p", TMSG_INFO(pStub->rpcMsg.msgType),
-          keyCode, pStub->rpcMsg.info.handle);
-  taosThreadMutexUnlock(&(pObj->mutex));
-  return keyCode;
+  taosThreadMutexUnlock(&pObj->mutex);
+  return seq;
 }
 
-int32_t syncRespMgrDel(SSyncRespMgr *pObj, uint64_t index) {
-  taosThreadMutexLock(&(pObj->mutex));
+int32_t syncRespMgrDel(SSyncRespMgr *pObj, uint64_t seq) {
+  taosThreadMutexLock(&pObj->mutex);
 
-  taosHashRemove(pObj->pRespHash, &index, sizeof(index));
+  int32_t code = taosHashRemove(pObj->pRespHash, &seq, sizeof(seq));
+  sNTrace(pObj->data, "remove message handle, seq:%" PRIu64 " code:%d", seq, code);
 
-  taosThreadMutexUnlock(&(pObj->mutex));
-  return 0;
+  taosThreadMutexUnlock(&pObj->mutex);
+  return code;
 }
 
-int32_t syncRespMgrGet(SSyncRespMgr *pObj, uint64_t index, SRespStub *pStub) {
-  taosThreadMutexLock(&(pObj->mutex));
+int32_t syncRespMgrGet(SSyncRespMgr *pObj, uint64_t seq, SRespStub *pStub) {
+  taosThreadMutexLock(&pObj->mutex);
 
-  void *pTmp = taosHashGet(pObj->pRespHash, &index, sizeof(index));
+  SRespStub *pTmp = taosHashGet(pObj->pRespHash, &seq, sizeof(uint64_t));
   if (pTmp != NULL) {
     memcpy(pStub, pTmp, sizeof(SRespStub));
+    sNTrace(pObj->data, "get message handle, type:%s seq:%" PRIu64 " handle:%p", TMSG_INFO(pStub->rpcMsg.msgType), seq,
+            pStub->rpcMsg.info.handle);
 
-    sNTrace(pObj->data, "get message handle, type:%s seq:%" PRIu64 " handle:%p", TMSG_INFO(pStub->rpcMsg.msgType),
-            index, pStub->rpcMsg.info.handle);
-    taosThreadMutexUnlock(&(pObj->mutex));
+    taosThreadMutexUnlock(&pObj->mutex);
     return 1;  // get one object
   }
-  taosThreadMutexUnlock(&(pObj->mutex));
+
+  taosThreadMutexUnlock(&pObj->mutex);
   return 0;  // get none object
 }
 
-int32_t syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t index, SRespStub *pStub) {
-  taosThreadMutexLock(&(pObj->mutex));
+int32_t syncRespMgrGetAndDel(SSyncRespMgr *pObj, uint64_t seq, SRpcHandleInfo *pInfo) {
+  taosThreadMutexLock(&pObj->mutex);
 
-  void *pTmp = taosHashGet(pObj->pRespHash, &index, sizeof(index));
-  if (pTmp != NULL) {
-    memcpy(pStub, pTmp, sizeof(SRespStub));
+  SRespStub *pStub = taosHashGet(pObj->pRespHash, &seq, sizeof(uint64_t));
+  if (pStub != NULL) {
+    *pInfo = pStub->rpcMsg.info;
+    sNTrace(pObj->data, "get-and-del message handle:%p, type:%s seq:%" PRIu64, pStub->rpcMsg.info.handle,
+            TMSG_INFO(pStub->rpcMsg.msgType), seq);
+    taosHashRemove(pObj->pRespHash, &seq, sizeof(uint64_t));
 
-    sNTrace(pObj->data, "get-and-del message handle, type:%s seq:%" PRIu64 " handle:%p",
-            TMSG_INFO(pStub->rpcMsg.msgType), index, pStub->rpcMsg.info.handle);
-    taosHashRemove(pObj->pRespHash, &index, sizeof(index));
-    taosThreadMutexUnlock(&(pObj->mutex));
+    taosThreadMutexUnlock(&pObj->mutex);
     return 1;  // get one object
   }
-  taosThreadMutexUnlock(&(pObj->mutex));
-  return 0;  // get none object
-}
 
-void syncRespCleanRsp(SSyncRespMgr *pObj) {
-  taosThreadMutexLock(&(pObj->mutex));
-  syncRespCleanByTTL(pObj, -1, true);
-  taosThreadMutexUnlock(&(pObj->mutex));
-}
-
-void syncRespClean(SSyncRespMgr *pObj) {
-  taosThreadMutexLock(&(pObj->mutex));
-  syncRespCleanByTTL(pObj, pObj->ttl, false);
-  taosThreadMutexUnlock(&(pObj->mutex));
+  taosThreadMutexUnlock(&pObj->mutex);
+  return 0;  // get none object
 }
 
-void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) {
+static void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) {
   SRespStub *pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, NULL);
   int        cnt = 0;
   int        sum = 0;
   SSyncNode *pSyncNode = pObj->data;
 
-  SArray *delIndexArray = taosArrayInit(0, sizeof(uint64_t));
-  ASSERT(delIndexArray != NULL);
-  sDebug("vgId:%d, resp mgr begin clean by ttl", pSyncNode->vgId);
+  SArray *delIndexArray = taosArrayInit(4, sizeof(uint64_t));
+  if (delIndexArray == NULL) return;
 
+  sDebug("vgId:%d, resp mgr begin clean by ttl", pSyncNode->vgId);
   while (pStub) {
     size_t    len;
-    void *    key = taosHashGetKey(pStub, &len);
+    void     *key = taosHashGetKey(pStub, &len);
     uint64_t *pSeqNum = (uint64_t *)key;
     sum++;
 
@@ -149,15 +141,15 @@ void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) {
       pStub->rpcMsg.contLen = 0;
 
       // TODO: and make rpcMsg body, call commit cb
-      // pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &(pStub->rpcMsg), cbMeta);
+      // pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &pStub->rpcMsg, cbMeta);
 
       pStub->rpcMsg.code = TSDB_CODE_SYN_NOT_LEADER;
       if (pStub->rpcMsg.info.handle != NULL) {
-        tmsgSendRsp(&(pStub->rpcMsg));
+        tmsgSendRsp(&pStub->rpcMsg);
       }
     }
 
-    pStub = (SRespStub *)taosHashIterate(pObj->pRespHash, pStub);
+    pStub = taosHashIterate(pObj->pRespHash, pStub);
   }
 
   int32_t arraySize = taosArrayGetSize(delIndexArray);
@@ -170,3 +162,15 @@ void syncRespCleanByTTL(SSyncRespMgr *pObj, int64_t ttl, bool rsp) {
   }
   taosArrayDestroy(delIndexArray);
 }
+
+void syncRespCleanRsp(SSyncRespMgr *pObj) {
+  taosThreadMutexLock(&pObj->mutex);
+  syncRespCleanByTTL(pObj, -1, true);
+  taosThreadMutexUnlock(&pObj->mutex);
+}
+
+void syncRespClean(SSyncRespMgr *pObj) {
+  taosThreadMutexLock(&pObj->mutex);
+  syncRespCleanByTTL(pObj, pObj->ttl, false);
+  taosThreadMutexUnlock(&pObj->mutex);
+}

source/libs/sync/src/syncUtil.c

@@ -205,7 +205,7 @@ static void syncPeerState2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLen) {
 
     if (i < pSyncNode->replicaNum - 1) {
       len += snprintf(buf + len, bufLen - len, "%d:%" PRId64 " %" PRId64 ", ", i, pState->lastSendIndex,
-                     pState->lastSendTime);
+                      pState->lastSendTime);
     } else {
       len += snprintf(buf + len, bufLen - len, "%d:%" PRId64 " %" PRId64 "}", i, pState->lastSendIndex,
                       pState->lastSendTime);
@@ -216,6 +216,9 @@ static void syncPeerState2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLen) {
 void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNode* pNode, const char* format, ...) {
   if (pNode == NULL || pNode->pRaftCfg != NULL && pNode->pRaftStore == NULL || pNode->pLogStore == NULL) return;
 
+  // save error code, otherwise it will be overwritten
+  int32_t errCode = terrno;
+
   SSnapshot snapshot = {.data = NULL, .lastApplyIndex = -1, .lastApplyTerm = 0};
   if (pNode->pFsm != NULL && pNode->pFsm->FpGetSnapshotInfo != NULL) {
     pNode->pFsm->FpGetSnapshotInfo(pNode->pFsm, &snapshot);
@@ -242,16 +245,21 @@ void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNo
   int32_t writeLen = vsnprintf(eventLog, sizeof(eventLog), format, argpointer);
   va_end(argpointer);
 
+  int32_t aqItems = pNode->pFsm->FpApplyQueueItems(pNode->pFsm);
+
+  // restore error code
+  terrno = errCode;
+
   taosPrintLog(flags, level, dflag,
                "vgId:%d, sync %s "
                "%s"
                ", tm:%" PRIu64 ", cmt:%" PRId64 ", fst:%" PRId64 ", lst:%" PRId64 ", min:%" PRId64 ", snap:%" PRId64
-               ", snap-tm:%" PRIu64 ", sby:%d, stgy:%d, bch:%d, r-num:%d, lcfg:%" PRId64
+               ", snap-tm:%" PRIu64 ", sby:%d, aq:%d, bch:%d, r-num:%d, lcfg:%" PRId64
                ", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s",
                pNode->vgId, syncStr(pNode->state), eventLog, pNode->pRaftStore->currentTerm, pNode->commitIndex,
                logBeginIndex, logLastIndex, pNode->minMatchIndex, snapshot.lastApplyIndex, snapshot.lastApplyTerm,
-               pNode->pRaftCfg->isStandBy, pNode->pRaftCfg->snapshotStrategy, pNode->pRaftCfg->batchSize,
-               pNode->replicaNum, pNode->pRaftCfg->lastConfigIndex, pNode->changing, pNode->restoreFinish, quorum,
+               pNode->pRaftCfg->isStandBy, aqItems, pNode->pRaftCfg->batchSize, pNode->replicaNum,
+               pNode->pRaftCfg->lastConfigIndex, pNode->changing, pNode->restoreFinish, quorum,
                pNode->electTimerLogicClock, pNode->heartbeatTimerLogicClockUser, peerStr, cfgStr);
 }
 

source/libs/transport/src/transCli.c

@@ -474,7 +474,7 @@ void* destroyConnPool(void* pool) {
 }
 
 static SCliConn* getConnFromPool(void* pool, char* ip, uint32_t port) {
-  char key[32] = {0};
+  char key[TSDB_FQDN_LEN + 64] = {0};
   CONN_CONSTRUCT_HASH_KEY(key, ip, port);
 
   SConnList* plist = taosHashGet((SHashObj*)pool, key, strlen(key));
@@ -525,7 +525,7 @@ static void addConnToPool(void* pool, SCliConn* conn) {
   conn->status = ConnInPool;
 
   if (conn->list == NULL) {
-    char key[32] = {0};
+    char key[TSDB_FQDN_LEN + 64] = {0};
     CONN_CONSTRUCT_HASH_KEY(key, conn->ip, conn->port);
     tTrace("%s conn %p added to conn pool, read buf cap:%d", CONN_GET_INST_LABEL(conn), conn, conn->readBuf.cap);
     conn->list = taosHashGet((SHashObj*)pool, key, strlen(key));

source/util/src/tlog.c

@@ -317,14 +317,14 @@ static void taosGetLogFileName(char *fn) {
     for (int32_t i = 0; i < tsLogObj.fileNum; i++) {
       char fileName[LOG_FILE_NAME_LEN];
 
-      sprintf(fileName, "%s%d.0", fn, i);
+      snprintf(fileName, LOG_FILE_NAME_LEN, "%s%d.0", fn, i);
       bool file1open = taosCheckFileIsOpen(fileName);
 
-      sprintf(fileName, "%s%d.1", fn, i);
+      snprintf(fileName, LOG_FILE_NAME_LEN, "%s%d.1", fn, i);
       bool file2open = taosCheckFileIsOpen(fileName);
 
       if (!file1open && !file2open) {
-        sprintf(tsLogObj.logName, "%s%d", fn, i);
+        snprintf(tsLogObj.logName, LOG_FILE_NAME_LEN, "%s%d", fn, i);
         return;
       }
     }
@@ -598,7 +598,7 @@ static int32_t taosPushLogBuffer(SLogBuff *pLogBuf, const char *msg, int32_t msg
   remainSize = (start > end) ? (start - end - 1) : (start + LOG_BUF_SIZE(pLogBuf) - end - 1);
 
   if (lostLine > 0) {
-    sprintf(tmpBuf, "...Lost %" PRId64 " lines here...\n", lostLine);
+    snprintf(tmpBuf, tListLen(tmpBuf), "...Lost %" PRId64 " lines here...\n", lostLine);
     tmpBufLen = (int32_t)strlen(tmpBuf);
   }
 

tests/script/tsim/stream/basic1.sim

@@ -23,443 +23,516 @@ 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
+
+$loop_count = 0
+
+loop0:
+sleep 200
+
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
-if $rows != 4 then
-  print ======$rows
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
   return -1
 endi
 
+if $rows != 4 then
+  print =====rows=$rows
+  goto loop0
+endi
+
 # row 0
 if $data01 != 2 then
-  print ======$data01
-  return -1
+  print =====data01=$data01
+  goto loop0
 endi
 
 if $data02 != 2 then
-  print ======$data02
-  return -1
+  print =====data02=$data02
+  goto loop0
 endi
 
 if $data03 != 5 then
-  print ======$data03
-  return -1
+  print =====data03=$data03
+  goto loop0
 endi
 
 if $data04 != 2 then
-  print ======$data04
-  return -1
+  print =====data04=$data04
+  goto loop0
 endi
 
 if $data05 != 3 then
-  print ======$data05
-  return -1
+  print =====data05=$data05
+  goto loop0
 endi
 
 # row 1
 if $data11 != 1 then
-  print ======$data11
-  return -1
+  print =====data11=$data11
+  goto loop0
 endi
 
 if $data12 != 1 then
-  print ======$data12
-  return -1
+  print =====data12=$data12
+  goto loop0
 endi
 
 if $data13 != 2 then
-  print ======$data13
-  return -1
+  print =====data13=$data13
+  goto loop0
 endi
 
 if $data14 != 2 then
-  print ======$data14
-  return -1
+  print =====data14=$data14
+  goto loop0
 endi
 
 if $data15 != 3 then
-  print ======$data15
-  return -1
+  print =====data15=$data15
+  goto loop0
 endi
 
 # row 2
 if $data21 != 1 then
-  print ======$data21
+  print =====data21=$data21
   return -1
 endi
 
 if $data22 != 1 then
-  print ======$data22
+  print =====data22=$data22
   return -1
 endi
 
 if $data23 != 3 then
-  print ======$data23
+  print =====data23=$data23
   return -1
 endi
 
 if $data24 != 2 then
-  print ======$data24
+  print =====data24=$data24
   return -1
 endi
 
 if $data25 != 3 then
-  print ======$data25
+  print =====data25=$data25
   return -1
 endi
 
 # row 3
 if $data31 != 1 then
-  print ======$data31
+  print =====data31=$data31
   return -1
 endi
 
 if $data32 != 1 then
-  print ======$data32
+  print =====data32=$data32
   return -1
 endi
 
 if $data33 != 4 then
-  print ======$data33
+  print =====data33=$data33
   return -1
 endi
 
 if $data34 != 2 then
-  print ======$data34
+  print =====data34=$data34
   return -1
 endi
 
 if $data35 != 3 then
-  print ======$data35
+  print =====data35=$data35
   return -1
 endi
 
 sql insert into t1 values(1648791223001,12,14,13,11.1);
-sleep 500
+
+$loop_count = 0
+loop1:
+sleep 200
+
 sql select * from streamt;
 
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
 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
+  goto loop1
 endi
 
 # row 0
 if $data01 != 2 then
-  print ======$data01
-  return -1
+  print =====data01=$data01
+  goto loop1
 endi
 
 if $data02 != 2 then
-  print ======$data02
-  return -1
+  print =====data02=$data02
+  goto loop1
 endi
 
 if $data03 != 5 then
-  print ======$data03
-  return -1
+  print =====data03=$data03
+  goto loop1
 endi
 
 if $data04 != 2 then
-  print ======$data04
-  return -1
+  print =====data04=$data04
+  goto loop1
 endi
 
 if $data05 != 3 then
-  print ======$data05
-  return -1
+  print =====data05=$data05
+  goto loop1
 endi
 
 # row 1
 if $data11 != 1 then
-  print ======$data11
-  return -1
+  print =====data11=$data11
+  goto loop1
 endi
 
 if $data12 != 1 then
-  print ======$data12
-  return -1
+  print =====data12=$data12
+  goto loop1
 endi
 
 if $data13 != 12 then
-  print ======$data13
-  return -1
+  print =====data13=$data13
+  goto loop1
 endi
 
 if $data14 != 14 then
-  print ======$data14
-  return -1
+  print =====data14=$data14
+  goto loop1
 endi
 
 if $data15 != 13 then
-  print ======$data15
-  return -1
+  print =====data15=$data15
+  goto loop1
 endi
 
 # row 2
 if $data21 != 1 then
-  print ======$data21
+  print =====data21=$data21
   return -1
 endi
 
 if $data22 != 1 then
-  print ======$data22
+  print =====data22=$data22
   return -1
 endi
 
 if $data23 != 3 then
-  print ======$data23
+  print =====data23=$data23
   return -1
 endi
 
 if $data24 != 2 then
-  print ======$data24
+  print =====data24=$data24
   return -1
 endi
 
 if $data25 != 3 then
-  print ======$data25
+  print =====data25=$data25
   return -1
 endi
 
 # row 3
 if $data31 != 1 then
-  print ======$data31
+  print =====data31=$data31
   return -1
 endi
 
 if $data32 != 1 then
-  print ======$data32
+  print =====data32=$data32
   return -1
 endi
 
 if $data33 != 4 then
-  print ======$data33
+  print =====data33=$data33
   return -1
 endi
 
 if $data34 != 2 then
-  print ======$data34
+  print =====data34=$data34
   return -1
 endi
 
 if $data35 != 3 then
-  print ======$data35
+  print =====data35=$data35
   return -1
 endi
 
 sql insert into t1 values(1648791223002,12,14,13,11.1);
-sleep 100
+
+$loop_count = 0
+loop2:
+sleep 200
+
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
 # row 1
 if $data11 != 2 then
-  print ======$data11
-  return -1
+  print =====data11=$data11
+  goto loop2
 endi
 
 if $data12 != 2 then
-  print ======$data12
-  return -1
+  print =====data12=$data12
+  goto loop2
 endi
 
 if $data13 != 24 then
-  print ======$data13
-  return -1
+  print =====data13=$data13
+  goto loop2
 endi
 
 if $data14 != 14 then
-  print ======$data14
-  return -1
+  print =====data14=$data14
+  goto loop2
 endi
 
 if $data15 != 13 then
-  print ======$data15
-  return -1
+  print =====data15=$data15
+  goto loop2
 endi
 
 sql insert into t1 values(1648791223003,12,14,13,11.1);
-sleep 100
+
+$loop_count = 0
+loop3:
+sleep 200
+
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
 # row 1
 if $data11 != 3 then
-  print ======$data11
-  return -1
+  print =====data11=$data11
+  goto loop3
 endi
 
 if $data12 != 3 then
-  print ======$data12
-  return -1
+  print =====data12=$data12
+  goto loop3
 endi
 
 if $data13 != 36 then
-  print ======$data13
-  return -1
+  print =====data13=$data13
+  goto loop3
 endi
 
 if $data14 != 14 then
-  print ======$data14
-  return -1
+  print =====data14=$data14
+  goto loop3
 endi
 
 if $data15 != 13 then
-  print ======$data15
-  return -1
+  print =====data15=$data15
+  goto loop3
 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
+
+$loop_count = 0
+loop4:
+sleep 200
+
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
 # row 1
 if $data11 != 3 then
-  print ======$data11
-  return -1
+  print =====data11=$data11
+  goto loop4
 endi
 
 if $data12 != 3 then
-  print ======$data12
-  return -1
+  print =====data12=$data12
+  goto loop4
 endi
 
 if $data13 != 6 then
-  print ======$data13
-  return -1
+  print =====data13=$data13
+  goto loop4
 endi
 
 if $data14 != 3 then
-  print ======$data14
-  return -1
+  print =====data14=$data14
+  goto loop4
 endi
 
 if $data15 != 1 then
-  print ======$data15
-  return -1
+  print =====data15=$data15
+  goto loop4
 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
+
+$loop_count = 0
+loop5:
+sleep 200
+
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
 # row 2
 if $data21 != 2 then
-  print ======$data21
-  return -1
+  print =====data21=$data21
+  goto loop5
 endi
 
 if $data22 != 2 then
-  print ======$data22
-  return -1
+  print =====data22=$data22
+  goto loop5
 endi
 
 if $data23 != 6 then
-  print ======$data23
-  return -1
+  print =====data23=$data23
+  goto loop5
 endi
 
 if $data24 != 2 then
-  print ======$data24
-  return -1
+  print =====data24=$data24
+  goto loop5
 endi
 
 if $data25 != 3 then
-  print ======$data25
-  return -1
+  print =====data25=$data25
+  goto loop5
 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
+
+$loop_count = 0
+loop6:
+sleep 200
+
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
 # row 0
 if $data01 != 4 then
-  print ======$data01
-  return -1
+  print =====data01=$data01
+  goto loop6
 endi
 
 if $data02 != 4 then
-  print ======$data02
-  return -1
+  print =====data02=$data02
+  goto loop6
 endi
 
 if $data03 != 50 then
-  print ======$data03 != 50
-  return -1
+  print =====data03=$data03 != 50
+  goto loop6
 endi
 
 if $data04 != 20 then
-  print ======$data04 != 20
-  return -1
+  print =====data04=$data04 != 20
+  goto loop6
 endi
 
 if $data05 != 3 then
-  print ======$data05
-  return -1
+  print =====data05=$data05
+  goto loop6
 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
+
+$loop_count = 0
+loop7:
+sleep 200
+
 sql select `_wstart`, c1, c2 ,c3 ,c4, c5 from streamt;
 
+$loop_count = $loop_count + 1
+if $loop_count == 20 then
+  return -1
+endi
+
 # row 1
 if $data11 != 4 then
-  print ======$data11
-  return -1
+  print =====data11=$data11
+  goto loop7
 endi
 
 if $data12 != 4 then
-  print ======$data12
-  return -1
+  print =====data12=$data12
+  goto loop7
 endi
 
 if $data13 != 46 then
-  print ======$data13 != 46
-  return -1
+  print =====data13=$data13 != 46
+  goto loop7
 endi
 
 if $data14 != 20 then
-  print ======$data14 != 20
-  return -1
+  print =====data14=$data14 != 20
+  goto loop7
 endi
 
 if $data15 != 1 then
-  print ======$data15
-  return -1
+  print =====data15=$data15
+  goto loop7
 endi
 
 # row 2
 if $data21 != 4 then
-  print ======$data21
-  return -1
+  print =====data21=$data21
+  goto loop7
 endi
 
 if $data22 != 4 then
-  print ======$data22
-  return -1
+  print =====data22=$data22
+  goto loop7
 endi
 
 if $data23 != 15 then
-  print ======$data23
-    return -1
+  print =====data23=$data23
+  goto loop7
 endi
 
 if $data24 != 4 then
-  print ======$data24
-  return -1
+  print =====data24=$data24
+  goto loop7
 endi
 
 if $data25 != 3 then
-  print ======$data25
-  return -1
+  print =====data25=$data25
+  goto loop7
 endi
 
 sql create database test2 vgroups 1;
@@ -479,11 +552,11 @@ sql insert into t1 values(1648791213000,1,1,1,1.0) t2 values(1648791213000,2,2,2
 
 $loop_count = 0
 
-loop0:
-sleep 300
+loop8:
+sleep 200
 
 $loop_count = $loop_count + 1
-if $loop_count == 10 then
+if $loop_count == 20 then
   return -1
 endi
 
@@ -491,7 +564,7 @@ sql select * from streamt;
 
 if $rows != 4 then
   print =====rows=$rows
-  goto loop0
+  goto loop8
 endi
 
 sql insert into t1 values(1648791213000,5,5,5,5.0) t2 values(1648791213000,6,6,6,6.0) t5 values(1648791213000,7,7,7,7.0);
@@ -499,11 +572,11 @@ sql insert into t1 values(1648791213000,5,5,5,5.0) t2 values(1648791213000,6,6,6
 
 $loop_count = 0
 
-loop1:
-sleep 300
+loop9:
+sleep 200
 
 $loop_count = $loop_count + 1
-if $loop_count == 10 then
+if $loop_count == 20 then
   return -1
 endi
 
@@ -511,51 +584,51 @@ sql select * from streamt order by c4 desc;
 
 if $rows != 5 then
   print =====rows=$rows
-  goto loop1
+  goto loop9
 endi
 
 # row 0
 if $data01 != 1 then
   print =====data01=$data01
-  goto loop1
+  goto loop9
 endi
 
 if $data02 != 7 then
   print =====data02=$data02
-  goto loop1
+  goto loop9
 endi
 
 # row 1
 if $data11 != 1 then
   print =====data11=$data11
-  goto loop1
+  goto loop9
 endi
 
 if $data12 != 6 then
   print =====data12=$data12
-  goto loop1
+  goto loop9
 endi
 
 # row 2
 if $data21 != 1 then
   print =====data21=$data21
-  goto loop1
+  goto loop9
 endi
 
 if $data22 != 5 then
   print =====data22=$data22
-  goto loop1
+  goto loop9
 endi
 
 sql insert into t1 values(1648791213000,8,8,8,8.0);
 
 $loop_count = 0
 
-loop2:
-sleep 300
+loop10:
+sleep 200
 
 $loop_count = $loop_count + 1
-if $loop_count == 10 then
+if $loop_count == 20 then
   return -1
 endi
 
@@ -564,28 +637,29 @@ sql select * from streamt order by c4 desc;
 # row 0
 if $data01 != 1 then
   print =====data01=$data01
-  goto loop2
+  goto loop10
 endi
 
 if $data02 != 8 then
   print =====data02=$data02
-  goto loop2
+  goto loop10
 endi
 
 $loop_count = 0
-loop3:
-sleep 300
+loop11:
+sleep 200
+
+sql select count(*) from streamt3;
 
 $loop_count = $loop_count + 1
-if $loop_count == 10 then
+if $loop_count == 20 then
   return -1
 endi
 
-sql select count(*) from streamt3;
 # row 0
 if $data00 != 5 then
   print =====data00=$data00
-  goto loop3
+  goto loop11
 endi
 
 #max,min selectivity
@@ -601,25 +675,26 @@ sql insert into ts1 values(1648791222001,2,2,3);
 sleep 50
 
 $loop_count = 0
-loop3:
+loop12:
+sleep 200
+
 sql select * from streamtST3;
 
-sleep 300
 $loop_count = $loop_count + 1
-if $loop_count == 10 then
+if $loop_count == 20 then
   return -1
 endi
 
 # row 0
 if $data02 != 1 then
   print =====data02=$data02
-  goto loop3
+  goto loop12
 endi
 
 # row 1
 if $data12 != 2 then
   print =====data12=$data12
-  goto loop3
+  goto loop12
 endi
 
 
@@ -629,19 +704,22 @@ sql create table t1(ts timestamp, a int, b int , c int, d double);
 sql create stream streams4 trigger at_once  into streamt4 as select  _wstart, count(*) c1 from t1 where a > 5 interval(10s);
 sql insert into t1 values(1648791213000,1,2,3,1.0);
 
+$loop_count = 0
+loop13:
 sleep 200
+
 sql select * from streamt4;
 
 # row 0
 if $rows != 0 then
   print =====rows=$rows
-  return -1
+  goto loop13
 endi
 
 sql insert into t1 values(1648791213000,6,2,3,1.0);
 
 $loop_count = 0
-loop4:
+loop14:
 sleep 200
 sql select * from streamt4;
 
@@ -652,13 +730,13 @@ endi
 
 if $data01 != 1 then
   print =====data01=$data01
-  goto loop4
+  goto loop14
 endi
 
 sql insert into t1 values(1648791213000,2,2,3,1.0);
 
 $loop_count = 0
-loop5:
+loop15:
 sleep 200
 sql select * from streamt4;
 
@@ -669,7 +747,7 @@ endi
 
 if $rows != 0 then
   print =====rows=$rows
-  goto loop5
+  goto loop15
 endi
 
 

tests/script/tsim/stream/session0.sim

@@ -201,7 +201,7 @@ if $loop_count == 10 then
 endi
 
 if $rows != 1 then
-  print ======$rows
+  print =====rows=$rows
   goto loop2
 endi