From eac0d148685fd67faa91cdc1b92d2a4813d7a74c Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Wed, 10 Jan 2024 11:02:44 +0800
Subject: [PATCH 1/8] fix(stream): add check when merging rows from files and
 buffer.

---
 source/dnode/vnode/src/tsdb/tsdbRead2.c    | 104 +++------------------
 source/dnode/vnode/src/tsdb/tsdbReadUtil.c |   9 +-
 source/dnode/vnode/src/tsdb/tsdbReadUtil.h |   6 ++
 3 files changed, 21 insertions(+), 98 deletions(-)

diff --git a/source/dnode/vnode/src/tsdb/tsdbRead2.c b/source/dnode/vnode/src/tsdb/tsdbRead2.c
index 255d5d4105..68ba00eefc 100644
--- a/source/dnode/vnode/src/tsdb/tsdbRead2.c
+++ b/source/dnode/vnode/src/tsdb/tsdbRead2.c
@@ -1490,7 +1490,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo*
   SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
 
   int64_t tsLast = INT64_MIN;
-  if (hasDataInSttBlock(pBlockScanInfo)) {
+  if (hasDataInSttBlock(pBlockScanInfo) && (!pBlockScanInfo->cleanSttBlocks)) {
     tsLast = getCurrentKeyInSttBlock(pSttBlockReader);
   }
 
@@ -1509,7 +1509,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo*
   int64_t minKey = 0;
   if (pReader->info.order == TSDB_ORDER_ASC) {
     minKey = INT64_MAX;  // chosen the minimum value
-    if (minKey > tsLast && hasDataInSttBlock(pBlockScanInfo)) {
+    if (minKey > tsLast && hasDataInSttBlock(pBlockScanInfo) && (!pBlockScanInfo->cleanSttBlocks)) {
       minKey = tsLast;
     }
 
@@ -1522,7 +1522,7 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo*
     }
   } else {
     minKey = INT64_MIN;
-    if (minKey < tsLast && hasDataInSttBlock(pBlockScanInfo)) {
+    if (minKey < tsLast && hasDataInSttBlock(pBlockScanInfo) && (!pBlockScanInfo->cleanSttBlocks)) {
       minKey = tsLast;
     }
 
@@ -1626,91 +1626,6 @@ static int32_t doMergeBufAndFileRows(STsdbReader* pReader, STableBlockScanInfo*
   return code;
 }
 
-static int32_t doMergeFileBlockAndLastBlock(SSttBlockReader* pSttBlockReader, STsdbReader* pReader,
-                                            STableBlockScanInfo* pBlockScanInfo, SBlockData* pBlockData,
-                                            bool mergeBlockData) {
-  SRowMerger*         pMerger = &pReader->status.merger;
-  SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
-
-  int64_t  tsLastBlock = getCurrentKeyInSttBlock(pSttBlockReader);
-  bool     copied = false;
-  int32_t  code = TSDB_CODE_SUCCESS;
-  SRow*    pTSRow = NULL;
-  TSDBROW* pRow = tMergeTreeGetRow(&pSttBlockReader->mergeTree);
-
-  // create local variable to hold the row value
-  TSDBROW fRow = {.iRow = pRow->iRow, .type = TSDBROW_COL_FMT, .pBlockData = pRow->pBlockData};
-
-  tsdbTrace("fRow ptr:%p, %d, uid:%" PRIu64 ", ts:%" PRId64 " %s", pRow->pBlockData, pRow->iRow, pSttBlockReader->uid,
-            fRow.pBlockData->aTSKEY[fRow.iRow], pReader->idStr);
-
-  // only stt block exists
-  if ((!mergeBlockData) || (tsLastBlock != pBlockData->aTSKEY[pDumpInfo->rowIndex])) {
-    code = tryCopyDistinctRowFromSttBlock(&fRow, pSttBlockReader, pBlockScanInfo, tsLastBlock, pReader, &copied);
-    if (code) {
-      return code;
-    }
-
-    if (copied) {
-      pBlockScanInfo->lastProcKey = tsLastBlock;
-      return TSDB_CODE_SUCCESS;
-    } else {
-      code = tsdbRowMergerAdd(pMerger, &fRow, NULL);
-      if (code != TSDB_CODE_SUCCESS) {
-        return code;
-      }
-
-      TSDBROW* pRow1 = tMergeTreeGetRow(&pSttBlockReader->mergeTree);
-      tsdbRowMergerAdd(pMerger, pRow1, NULL);
-      doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, tsLastBlock, pMerger, &pReader->info.verRange,
-                            pReader->idStr);
-
-      code = tsdbRowMergerGetRow(pMerger, &pTSRow);
-      if (code != TSDB_CODE_SUCCESS) {
-        return code;
-      }
-
-      code = doAppendRowFromTSRow(pReader->resBlockInfo.pResBlock, pReader, pTSRow, pBlockScanInfo);
-
-      taosMemoryFree(pTSRow);
-      tsdbRowMergerClear(pMerger);
-
-      if (code != TSDB_CODE_SUCCESS) {
-        return code;
-      }
-    }
-  } else {  // not merge block data
-    code = tsdbRowMergerAdd(pMerger, &fRow, NULL);
-    if (code != TSDB_CODE_SUCCESS) {
-      return code;
-    }
-
-    doMergeRowsInSttBlock(pSttBlockReader, pBlockScanInfo, tsLastBlock, pMerger, &pReader->info.verRange,
-                          pReader->idStr);
-
-    // merge with block data if ts == key
-    if (tsLastBlock == pBlockData->aTSKEY[pDumpInfo->rowIndex]) {
-      doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader);
-    }
-
-    code = tsdbRowMergerGetRow(pMerger, &pTSRow);
-    if (code != TSDB_CODE_SUCCESS) {
-      return code;
-    }
-
-    code = doAppendRowFromTSRow(pReader->resBlockInfo.pResBlock, pReader, pTSRow, pBlockScanInfo);
-
-    taosMemoryFree(pTSRow);
-    tsdbRowMergerClear(pMerger);
-
-    if (code != TSDB_CODE_SUCCESS) {
-      return code;
-    }
-  }
-
-  return TSDB_CODE_SUCCESS;
-}
-
 static int32_t mergeFileBlockAndSttBlock(STsdbReader* pReader, SSttBlockReader* pSttBlockReader, int64_t key,
                                          STableBlockScanInfo* pBlockScanInfo, SBlockData* pBlockData) {
   SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
@@ -1816,7 +1731,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
   TSDBROW* piRow = getValidMemRow(&pBlockScanInfo->iiter, pDelList, pReader);
 
   int64_t tsLast = INT64_MIN;
-  if (hasDataInSttBlock(pBlockScanInfo)) {
+  if (hasDataInSttBlock(pBlockScanInfo) && (!pBlockScanInfo->cleanSttBlocks)) {
     tsLast = getCurrentKeyInSttBlock(pSttBlockReader);
   }
 
@@ -1865,7 +1780,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
       minKey = key;
     }
 
-    if (minKey > tsLast && hasDataInSttBlock(pBlockScanInfo)) {
+    if (minKey > tsLast && hasDataInSttBlock(pBlockScanInfo) && (!pBlockScanInfo->cleanSttBlocks)) {
       minKey = tsLast;
     }
   } else {
@@ -1882,7 +1797,7 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo*
       minKey = key;
     }
 
-    if (minKey < tsLast && hasDataInSttBlock(pBlockScanInfo)) {
+    if (minKey < tsLast && hasDataInSttBlock(pBlockScanInfo) && (!pBlockScanInfo->cleanSttBlocks)) {
       minKey = tsLast;
     }
   }
@@ -2169,10 +2084,15 @@ static bool initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScan
       pScanInfo->sttKeyInfo.nextProcKey =
           ASCENDING_TRAVERSE(pReader->info.order) ? pScanInfo->sttWindow.skey : pScanInfo->sttWindow.ekey;
       hasData = true;
-    } else {
+    } else {  // not clean stt blocks
+      INIT_TIMEWINDOW(pScanInfo->sttWindow);  //reset the time window
+      pScanInfo->sttBlockReturned = false;
       hasData = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, &pReader->info.verRange);
     }
   } else {
+    pScanInfo->cleanSttBlocks = false;
+    INIT_TIMEWINDOW(pScanInfo->sttWindow);  //reset the time window
+    pScanInfo->sttBlockReturned = false;
     hasData = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, &pReader->info.verRange);
   }
 
diff --git a/source/dnode/vnode/src/tsdb/tsdbReadUtil.c b/source/dnode/vnode/src/tsdb/tsdbReadUtil.c
index b4b92c0008..33604d21de 100644
--- a/source/dnode/vnode/src/tsdb/tsdbReadUtil.c
+++ b/source/dnode/vnode/src/tsdb/tsdbReadUtil.c
@@ -22,12 +22,6 @@
 #include "tsdbUtil2.h"
 #include "tsimplehash.h"
 
-#define INIT_TIMEWINDOW(_w) \
-  do {                      \
-    (_w)->skey = INT64_MAX; \
-    (_w)->ekey = INT64_MIN; \
-  } while (0);
-
 static bool overlapWithDelSkylineWithoutVer(STableBlockScanInfo* pBlockScanInfo, const SBrinRecord* pRecord,
                                             int32_t order);
 
@@ -165,6 +159,9 @@ SSHashObj* createDataBlockScanInfo(STsdbReader* pTsdbReader, SBlockInfoBuf* pBuf
     INIT_TIMEWINDOW(&pScanInfo->sttWindow);
     INIT_TIMEWINDOW(&pScanInfo->filesetWindow);
 
+    pScanInfo->cleanSttBlocks = false;
+    pScanInfo->sttBlockReturned = false;
+
     pUidList->tableUidList[j] = idList[j].uid;
 
     if (ASCENDING_TRAVERSE(pTsdbReader->info.order)) {
diff --git a/source/dnode/vnode/src/tsdb/tsdbReadUtil.h b/source/dnode/vnode/src/tsdb/tsdbReadUtil.h
index cd9942a0a6..7a93e73a07 100644
--- a/source/dnode/vnode/src/tsdb/tsdbReadUtil.h
+++ b/source/dnode/vnode/src/tsdb/tsdbReadUtil.h
@@ -26,6 +26,12 @@ extern "C" {
 
 #define ASCENDING_TRAVERSE(o) (o == TSDB_ORDER_ASC)
 
+#define INIT_TIMEWINDOW(_w) \
+  do {                      \
+    (_w)->skey = INT64_MAX; \
+    (_w)->ekey = INT64_MIN; \
+  } while (0);
+
 typedef enum {
   READER_STATUS_SUSPEND = 0x1,
   READER_STATUS_NORMAL = 0x2,

From 1b6d55616fd4c5e036b525d13464449db115768f Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Wed, 10 Jan 2024 11:28:06 +0800
Subject: [PATCH 2/8] fix(stream): fix syntax error.

---
 source/dnode/vnode/src/tsdb/tsdbRead2.c | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/source/dnode/vnode/src/tsdb/tsdbRead2.c b/source/dnode/vnode/src/tsdb/tsdbRead2.c
index 68ba00eefc..7360f59301 100644
--- a/source/dnode/vnode/src/tsdb/tsdbRead2.c
+++ b/source/dnode/vnode/src/tsdb/tsdbRead2.c
@@ -2085,13 +2085,13 @@ static bool initSttBlockReader(SSttBlockReader* pSttBlockReader, STableBlockScan
           ASCENDING_TRAVERSE(pReader->info.order) ? pScanInfo->sttWindow.skey : pScanInfo->sttWindow.ekey;
       hasData = true;
     } else {  // not clean stt blocks
-      INIT_TIMEWINDOW(pScanInfo->sttWindow);  //reset the time window
+      INIT_TIMEWINDOW(&pScanInfo->sttWindow);  //reset the time window
       pScanInfo->sttBlockReturned = false;
       hasData = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, &pReader->info.verRange);
     }
   } else {
     pScanInfo->cleanSttBlocks = false;
-    INIT_TIMEWINDOW(pScanInfo->sttWindow);  //reset the time window
+    INIT_TIMEWINDOW(&pScanInfo->sttWindow);  //reset the time window
     pScanInfo->sttBlockReturned = false;
     hasData = nextRowFromSttBlocks(pSttBlockReader, pScanInfo, &pReader->info.verRange);
   }

From 7744a31deb3dce64623334a67aaa0949b6f4cd8b Mon Sep 17 00:00:00 2001
From: Alex Duan <417921451@qq.com>
Date: Wed, 10 Jan 2024 11:29:24 +0800
Subject: [PATCH 3/8] case: fix memory leak with return cause

---
 tests/army/enterprise/s3/s3_basic.py | 33 ++++++++++++++--------------
 1 file changed, 16 insertions(+), 17 deletions(-)

diff --git a/tests/army/enterprise/s3/s3_basic.py b/tests/army/enterprise/s3/s3_basic.py
index 112cb65b15..8e7c1f2d35 100644
--- a/tests/army/enterprise/s3/s3_basic.py
+++ b/tests/army/enterprise/s3/s3_basic.py
@@ -86,30 +86,29 @@ class TDTestCase(TBase):
         tdLog.debug(f"start to excute {__file__}")
         if eos.isArm64Cpu():
             tdLog.success(f"{__file__} arm64 ignore executed")
-            return
+        else:
+            # insert data
+            self.insertData()
 
-        # insert data
-        self.insertData()
+            # check insert data correct
+            self.checkInsertCorrect()
 
-        # check insert data correct
-        self.checkInsertCorrect()
+            # save
+            self.snapshotAgg()
 
-        # save
-        self.snapshotAgg()
+            # do action
+            self.doAction()
 
-        # do action
-        self.doAction()
+            # check save agg result correct
+            self.checkAggCorrect()
 
-        # check save agg result correct
-        self.checkAggCorrect()
+            # check insert correct again
+            self.checkInsertCorrect()
 
-        # check insert correct again
-        self.checkInsertCorrect()
+            # drop database and free s3 file
+            self.dropDb()
 
-        # drop database and free s3 file
-        self.dropDb()
-
-        tdLog.success(f"{__file__} successfully executed")
+            tdLog.success(f"{__file__} successfully executed")
 
         
 

From b0202bb8536718d117e89aa527b77603ddaf9c57 Mon Sep 17 00:00:00 2001
From: Haojun Liao <hjliao@taosdata.com>
Date: Tue, 9 Jan 2024 21:40:27 +0800
Subject: [PATCH 4/8] fix(tsdb): open reader during resume task for main reader
 after suspending reader.

---
 source/dnode/vnode/src/tsdb/tsdbRead2.c | 100 ++++++++++++++----------
 1 file changed, 59 insertions(+), 41 deletions(-)

diff --git a/source/dnode/vnode/src/tsdb/tsdbRead2.c b/source/dnode/vnode/src/tsdb/tsdbRead2.c
index 7360f59301..62be0968e3 100644
--- a/source/dnode/vnode/src/tsdb/tsdbRead2.c
+++ b/source/dnode/vnode/src/tsdb/tsdbRead2.c
@@ -676,6 +676,10 @@ static int32_t doCopyColVal(SColumnInfoData* pColInfoData, int32_t rowIndex, int
 }
 
 static SFileDataBlockInfo* getCurrentBlockInfo(SDataBlockIter* pBlockIter) {
+  if (pBlockIter->blockList == NULL) {
+    return NULL;
+  }
+
   size_t num = TARRAY_SIZE(pBlockIter->blockList);
   if (num == 0) {
     ASSERT(pBlockIter->numOfBlocks == num);
@@ -1969,30 +1973,34 @@ static int32_t initMemDataIterator(STableBlockScanInfo* pBlockScanInfo, STsdbRea
   return TSDB_CODE_SUCCESS;
 }
 
-static bool isValidFileBlockRow(SBlockData* pBlockData, SFileBlockDumpInfo* pDumpInfo,
-                                STableBlockScanInfo* pBlockScanInfo, STsdbReader* pReader) {
+static bool isValidFileBlockRow(SBlockData* pBlockData, int32_t rowIndex, STableBlockScanInfo* pBlockScanInfo, bool asc,
+                                STsdbReaderInfo* pInfo) {
   // it is an multi-table data block
   if (pBlockData->aUid != NULL) {
-    uint64_t uid = pBlockData->aUid[pDumpInfo->rowIndex];
+    uint64_t uid = pBlockData->aUid[rowIndex];
     if (uid != pBlockScanInfo->uid) {  // move to next row
       return false;
     }
   }
 
   // check for version and time range
-  int64_t ver = pBlockData->aVersion[pDumpInfo->rowIndex];
-  if (ver > pReader->info.verRange.maxVer || ver < pReader->info.verRange.minVer) {
+  int64_t ver = pBlockData->aVersion[rowIndex];
+  if (ver > pInfo->verRange.maxVer || ver < pInfo->verRange.minVer) {
     return false;
   }
 
-  int64_t ts = pBlockData->aTSKEY[pDumpInfo->rowIndex];
-  if (ts > pReader->info.window.ekey || ts < pReader->info.window.skey) {
+  int64_t ts = pBlockData->aTSKEY[rowIndex];
+  if (ts > pInfo->window.ekey || ts < pInfo->window.skey) {
+    return false;
+  }
+
+  if ((asc && (ts <= pBlockScanInfo->lastProcKey)) || ((!asc) && (ts >= pBlockScanInfo->lastProcKey))) {
     return false;
   }
 
   if (pBlockScanInfo->delSkyline != NULL && TARRAY_SIZE(pBlockScanInfo->delSkyline) > 0) {
     bool dropped = hasBeenDropped(pBlockScanInfo->delSkyline, &pBlockScanInfo->fileDelIndex, ts, ver,
-                                  pReader->info.order, &pReader->info.verRange);
+                                  pInfo->order, &pInfo->verRange);
     if (dropped) {
       return false;
     }
@@ -2344,7 +2352,7 @@ static int32_t buildComposedDataBlock(STsdbReader* pReader) {
     {
       while (pBlockData->nRow > 0 && pBlockData->uid == pBlockScanInfo->uid) {
         // find the first qualified row in data block
-        if (isValidFileBlockRow(pBlockData, pDumpInfo, pBlockScanInfo, pReader)) {
+        if (isValidFileBlockRow(pBlockData, pDumpInfo->rowIndex, pBlockScanInfo, asc, &pReader->info)) {
           hasBlockData = true;
           break;
         }
@@ -4130,37 +4138,21 @@ void tsdbReaderClose2(STsdbReader* pReader) {
   taosMemoryFreeClear(pReader);
 }
 
-int32_t tsdbReaderSuspend2(STsdbReader* pReader) {
-  // save reader's base state & reset top state to be reconstructed from base state
-  int32_t              code = 0;
-  SReaderStatus*       pStatus = &pReader->status;
-  STableBlockScanInfo* pBlockScanInfo = NULL;
-
-  pReader->status.suspendInvoked = true;  // record the suspend status
+static int32_t doSuspendCurrentReader(STsdbReader* pCurrentReader) {
+  SReaderStatus* pStatus = &pCurrentReader->status;
 
   if (pStatus->loadFromFile) {
-    SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(&pReader->status.blockIter);
-    if (pBlockInfo != NULL) {
-      pBlockScanInfo = getTableBlockScanInfo(pStatus->pTableMap, pBlockInfo->uid, pReader->idStr);
-      if (pBlockScanInfo == NULL) {
-        goto _err;
-      }
-    } else {
-      pBlockScanInfo = *pStatus->pTableIter;
-    }
+    tsdbDataFileReaderClose(&pCurrentReader->pFileReader);
 
-    tsdbDataFileReaderClose(&pReader->pFileReader);
-
-    SReadCostSummary* pCost = &pReader->cost;
-    pReader->status.pLDataIterArray = destroySttBlockReader(pReader->status.pLDataIterArray, &pCost->sttCost);
-    pReader->status.pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
+    SReadCostSummary* pCost = &pCurrentReader->cost;
+    pStatus->pLDataIterArray = destroySttBlockReader(pStatus->pLDataIterArray, &pCost->sttCost);
+    pStatus->pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
   }
 
   // resetDataBlockScanInfo excluding lastKey
   STableBlockScanInfo** p = NULL;
 
-  int32_t step = ASCENDING_TRAVERSE(pReader->info.order) ? 1 : -1;
-
+  int32_t step = ASCENDING_TRAVERSE(pCurrentReader->info.order) ? 1 : -1;
   int32_t iter = 0;
   while ((p = tSimpleHashIterate(pStatus->pTableMap, p, &iter)) != NULL) {
     STableBlockScanInfo* pInfo = *(STableBlockScanInfo**)p;
@@ -4171,6 +4163,26 @@ int32_t tsdbReaderSuspend2(STsdbReader* pReader) {
   pStatus->uidList.currentIndex = 0;
   initReaderStatus(pStatus);
 
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t tsdbReaderSuspend2(STsdbReader* pReader) {
+  // save reader's base state & reset top state to be reconstructed from base state
+  int32_t code = 0;
+  pReader->status.suspendInvoked = true;  // record the suspend status
+
+  if (pReader->type == TIMEWINDOW_RANGE_EXTERNAL) {
+    if (pReader->step == EXTERNAL_ROWS_PREV) {
+      doSuspendCurrentReader(pReader->innerReader[0]);
+    } else if (pReader->step == EXTERNAL_ROWS_MAIN) {
+      doSuspendCurrentReader(pReader);
+    } else {
+      doSuspendCurrentReader(pReader->innerReader[1]);
+    }
+  } else {
+    doSuspendCurrentReader(pReader);
+  }
+
   tsdbUntakeReadSnap2(pReader, pReader->pReadSnap, false);
   pReader->pReadSnap = NULL;
   if (pReader->bFilesetDelimited) {
@@ -4179,16 +4191,16 @@ int32_t tsdbReaderSuspend2(STsdbReader* pReader) {
   }
   pReader->flag = READER_STATUS_SUSPEND;
 
+  if (pReader->type == TIMEWINDOW_RANGE_EXTERNAL) {
+    clearSharedPtr(pReader->innerReader[0]);
+    clearSharedPtr(pReader->innerReader[1]);
+  }
+
 #if SUSPEND_RESUME_TEST
   tsem_post(&pReader->resumeAfterSuspend);
 #endif
 
-  tsdbDebug("reader: %p suspended uid %" PRIu64 " in this query %s", pReader, pBlockScanInfo ? pBlockScanInfo->uid : 0,
-            pReader->idStr);
-  return code;
-
-_err:
-  tsdbError("failed to suspend data reader, code:%s %s", tstrerror(code), pReader->idStr);
+  tsdbDebug("reader: %p suspended in this query %s, step:%d", pReader, pReader->idStr, pReader->step);
   return code;
 }
 
@@ -4219,8 +4231,7 @@ int32_t tsdbReaderResume2(STsdbReader* pReader) {
   int32_t               code = 0;
   STableBlockScanInfo** pBlockScanInfo = pReader->status.pTableIter;
 
-  //  restore reader's state
-  //  task snapshot
+  //  restore reader's state, task snapshot
   int32_t numOfTables = tSimpleHashGetSize(pReader->status.pTableMap);
   if (numOfTables > 0) {
     qTrace("tsdb/reader: %p, take snapshot", pReader);
@@ -4245,7 +4256,14 @@ int32_t tsdbReaderResume2(STsdbReader* pReader) {
       pNextReader->resBlockInfo.capacity = 1;
       setSharedPtr(pNextReader, pReader);
 
-      code = doOpenReaderImpl(pPrevReader);
+      if (pReader->step == 0 || pReader->step == EXTERNAL_ROWS_PREV) {
+        code = doOpenReaderImpl(pPrevReader);
+      } else if (pReader->step == EXTERNAL_ROWS_MAIN) {
+        code = doOpenReaderImpl(pReader);
+      } else {
+        code = doOpenReaderImpl(pNextReader);
+      }
+
       if (code != TSDB_CODE_SUCCESS) {
         return code;
       }

From 27b63b721fb40e96a74f920a50edbf9040d5d5bf Mon Sep 17 00:00:00 2001
From: wangmm0220 <wangmm0220@gmail.com>
Date: Wed, 10 Jan 2024 17:36:51 +0800
Subject: [PATCH 5/8] fix:decode mqDataMsg error for the message with old
 version

---
 source/common/src/tmsg.c | 24 +++++++++++++++++++-----
 1 file changed, 19 insertions(+), 5 deletions(-)

diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c
index 3ff814dda9..c9e2908e8a 100644
--- a/source/common/src/tmsg.c
+++ b/source/common/src/tmsg.c
@@ -8334,7 +8334,7 @@ int32_t tDecodeMqMetaRsp(SDecoder *pDecoder, SMqMetaRsp *pRsp) {
   return 0;
 }
 
-int32_t tEncodeMqDataRsp(SEncoder *pEncoder, const SMqDataRsp *pRsp) {
+int32_t tEncodeMqDataRspCommon(SEncoder *pEncoder, const SMqDataRsp *pRsp) {
   if (tEncodeSTqOffsetVal(pEncoder, &pRsp->reqOffset) < 0) return -1;
   if (tEncodeSTqOffsetVal(pEncoder, &pRsp->rspOffset) < 0) return -1;
   if (tEncodeI32(pEncoder, pRsp->blockNum) < 0) return -1;
@@ -8356,11 +8356,16 @@ int32_t tEncodeMqDataRsp(SEncoder *pEncoder, const SMqDataRsp *pRsp) {
       }
     }
   }
+  return 0;
+}
+
+int32_t tEncodeMqDataRsp(SEncoder *pEncoder, const SMqDataRsp *pRsp) {
+  if (tEncodeMqDataRspCommon(pEncoder, pRsp) < 0) return -1;
   if (tEncodeI64(pEncoder, pRsp->sleepTime) < 0) return -1;
   return 0;
 }
 
-int32_t tDecodeMqDataRsp(SDecoder *pDecoder, SMqDataRsp *pRsp) {
+int32_t tDecodeMqDataRspCommon(SDecoder *pDecoder, SMqDataRsp *pRsp) {
   if (tDecodeSTqOffsetVal(pDecoder, &pRsp->reqOffset) < 0) return -1;
   if (tDecodeSTqOffsetVal(pDecoder, &pRsp->rspOffset) < 0) return -1;
   if (tDecodeI32(pDecoder, &pRsp->blockNum) < 0) return -1;
@@ -8402,7 +8407,15 @@ int32_t tDecodeMqDataRsp(SDecoder *pDecoder, SMqDataRsp *pRsp) {
       }
     }
   }
-  if (tDecodeI64(pDecoder, &pRsp->sleepTime) < 0) return -1;
+
+  return 0;
+}
+
+int32_t tDecodeMqDataRsp(SDecoder *pDecoder, SMqDataRsp *pRsp) {
+  if (tDecodeMqDataRspCommon(pDecoder, pRsp) < 0) return -1;
+  if (!tDecodeIsEnd(pDecoder)) {
+    if (tDecodeI64(pDecoder, &pRsp->sleepTime) < 0) return -1;
+  }
 
   return 0;
 }
@@ -8418,7 +8431,7 @@ void tDeleteMqDataRsp(SMqDataRsp *pRsp) {
 }
 
 int32_t tEncodeSTaosxRsp(SEncoder *pEncoder, const STaosxRsp *pRsp) {
-  if (tEncodeMqDataRsp(pEncoder, (const SMqDataRsp *)pRsp) < 0) return -1;
+  if (tEncodeMqDataRspCommon(pEncoder, (const SMqDataRsp *)pRsp) < 0) return -1;
 
   if (tEncodeI32(pEncoder, pRsp->createTableNum) < 0) return -1;
   if (pRsp->createTableNum) {
@@ -8432,7 +8445,7 @@ int32_t tEncodeSTaosxRsp(SEncoder *pEncoder, const STaosxRsp *pRsp) {
 }
 
 int32_t tDecodeSTaosxRsp(SDecoder *pDecoder, STaosxRsp *pRsp) {
-  if (tDecodeMqDataRsp(pDecoder, (SMqDataRsp *)pRsp) < 0) return -1;
+  if (tDecodeMqDataRspCommon(pDecoder, (SMqDataRsp*)pRsp) < 0) return -1;
 
   if (tDecodeI32(pDecoder, &pRsp->createTableNum) < 0) return -1;
   if (pRsp->createTableNum) {
@@ -8447,6 +8460,7 @@ int32_t tDecodeSTaosxRsp(SDecoder *pDecoder, STaosxRsp *pRsp) {
       taosArrayPush(pRsp->createTableReq, &pCreate);
     }
   }
+
   return 0;
 }
 

From df69a89da202a60df6915c2653e9af00f74aa6a7 Mon Sep 17 00:00:00 2001
From: wangmm0220 <wangmm0220@gmail.com>
Date: Wed, 10 Jan 2024 18:46:10 +0800
Subject: [PATCH 6/8] fix:[TD-28185]add pause & resume logic for snode

---
 include/dnode/vnode/tqCommon.h             |   2 +
 source/dnode/snode/src/snode.c             |   6 +-
 source/dnode/vnode/src/tq/tq.c             | 101 +-------------------
 source/dnode/vnode/src/tqCommon/tqCommon.c | 105 +++++++++++++++++++++
 source/libs/stream/src/streamMeta.c        |   8 +-
 5 files changed, 118 insertions(+), 104 deletions(-)

diff --git a/include/dnode/vnode/tqCommon.h b/include/dnode/vnode/tqCommon.h
index da2325f006..82c4102369 100644
--- a/include/dnode/vnode/tqCommon.h
+++ b/include/dnode/vnode/tqCommon.h
@@ -33,5 +33,7 @@ int32_t tqStreamTaskProcessRunReq(SStreamMeta* pMeta, SRpcMsg* pMsg, bool isLead
 int32_t tqStreamTaskResetStatus(SStreamMeta* pMeta);
 int32_t tqStartTaskCompleteCallback(SStreamMeta* pMeta);
 int32_t tqStreamTaskProcessTaskResetReq(SStreamMeta* pMeta, SRpcMsg* pMsg);
+int32_t tqStreamTaskProcessTaskPauseReq(SStreamMeta* pMeta, char* pMsg);
+int32_t tqStreamTaskProcessTaskResumeReq(void* handle, int64_t sversion, char* pMsg, bool fromVnode);
 
 #endif  // TDENGINE_TQ_COMMON_H
diff --git a/source/dnode/snode/src/snode.c b/source/dnode/snode/src/snode.c
index 33eb9cd3ed..9177559f61 100644
--- a/source/dnode/snode/src/snode.c
+++ b/source/dnode/snode/src/snode.c
@@ -195,7 +195,7 @@ int32_t sndProcessWriteMsg(SSnode *pSnode, SRpcMsg *pMsg, SRpcMsg *pRsp) {
     case TDMT_STREAM_TASK_DEPLOY: {
       void *  pReq = POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead));
       int32_t len = pMsg->contLen - sizeof(SMsgHead);
-      return tqStreamTaskProcessDeployReq(pSnode->pMeta, -1, pReq, len, true, true);
+      return tqStreamTaskProcessDeployReq(pSnode->pMeta, pMsg->info.conn.applyIndex, pReq, len, true, true);
     }
 
     case TDMT_STREAM_TASK_DROP:
@@ -204,6 +204,10 @@ int32_t sndProcessWriteMsg(SSnode *pSnode, SRpcMsg *pMsg, SRpcMsg *pRsp) {
       return tqStreamTaskProcessUpdateReq(pSnode->pMeta, &pSnode->msgCb, pMsg, true);
     case TDMT_VND_STREAM_TASK_RESET:
       return tqStreamTaskProcessTaskResetReq(pSnode->pMeta, pMsg);
+    case TDMT_STREAM_TASK_PAUSE:
+      return tqStreamTaskProcessTaskPauseReq(pSnode->pMeta, pMsg->pCont);
+    case TDMT_STREAM_TASK_RESUME:
+      return tqStreamTaskProcessTaskResumeReq(pSnode->pMeta, pMsg->info.conn.applyIndex, pMsg->pCont, false);
     default:
       ASSERT(0);
   }
diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c
index 65cc56c0b4..275bbffe25 100644
--- a/source/dnode/vnode/src/tq/tq.c
+++ b/source/dnode/vnode/src/tq/tq.c
@@ -1094,108 +1094,11 @@ int32_t tqProcessTaskDropReq(STQ* pTq, char* msg, int32_t msgLen) {
 }
 
 int32_t tqProcessTaskPauseReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
-  SVPauseStreamTaskReq* pReq = (SVPauseStreamTaskReq*)msg;
-
-  SStreamMeta* pMeta = pTq->pStreamMeta;
-  SStreamTask* pTask = streamMetaAcquireTask(pMeta, pReq->streamId, pReq->taskId);
-  if (pTask == NULL) {
-    tqError("vgId:%d process pause req, failed to acquire task:0x%x, it may have been dropped already", pMeta->vgId,
-            pReq->taskId);
-    // since task is in [STOP|DROPPING] state, it is safe to assume the pause is active
-    return TSDB_CODE_SUCCESS;
-  }
-
-  tqDebug("s-task:%s receive pause msg from mnode", pTask->id.idStr);
-  streamTaskPause(pTask, pMeta);
-
-  SStreamTask* pHistoryTask = NULL;
-  if (HAS_RELATED_FILLHISTORY_TASK(pTask)) {
-    pHistoryTask = streamMetaAcquireTask(pMeta, pTask->hTaskInfo.id.streamId, pTask->hTaskInfo.id.taskId);
-    if (pHistoryTask == NULL) {
-      tqError("vgId:%d process pause req, failed to acquire fill-history task:0x%" PRIx64
-              ", it may have been dropped already",
-              pMeta->vgId, pTask->hTaskInfo.id.taskId);
-      streamMetaReleaseTask(pMeta, pTask);
-
-      // since task is in [STOP|DROPPING] state, it is safe to assume the pause is active
-      return TSDB_CODE_SUCCESS;
-    }
-
-    tqDebug("s-task:%s fill-history task handle paused along with related stream task", pHistoryTask->id.idStr);
-
-    streamTaskPause(pHistoryTask, pMeta);
-    streamMetaReleaseTask(pMeta, pHistoryTask);
-  }
-
-  streamMetaReleaseTask(pMeta, pTask);
-  return TSDB_CODE_SUCCESS;
-}
-
-int32_t tqProcessTaskResumeImpl(STQ* pTq, SStreamTask* pTask, int64_t sversion, int8_t igUntreated) {
-  int32_t vgId = pTq->pStreamMeta->vgId;
-  if (pTask == NULL) {
-    return -1;
-  }
-
-  streamTaskResume(pTask);
-  ETaskStatus status = streamTaskGetStatus(pTask, NULL);
-
-  int32_t level = pTask->info.taskLevel;
-  if (level == TASK_LEVEL__SINK) {
-    if (status == TASK_STATUS__UNINIT) {
-    }
-    streamMetaReleaseTask(pTq->pStreamMeta, pTask);
-    return 0;
-  }
-
-  if (status == TASK_STATUS__READY || status == TASK_STATUS__SCAN_HISTORY || status == TASK_STATUS__CK) {
-    // no lock needs to secure the access of the version
-    if (igUntreated && level == TASK_LEVEL__SOURCE && !pTask->info.fillHistory) {
-      // discard all the data  when the stream task is suspended.
-      walReaderSetSkipToVersion(pTask->exec.pWalReader, sversion);
-      tqDebug("vgId:%d s-task:%s resume to exec, prev paused version:%" PRId64 ", start from vnode ver:%" PRId64
-              ", schedStatus:%d",
-              vgId, pTask->id.idStr, pTask->chkInfo.nextProcessVer, sversion, pTask->status.schedStatus);
-    } else {  // from the previous paused version and go on
-      tqDebug("vgId:%d s-task:%s resume to exec, from paused ver:%" PRId64 ", vnode ver:%" PRId64 ", schedStatus:%d",
-              vgId, pTask->id.idStr, pTask->chkInfo.nextProcessVer, sversion, pTask->status.schedStatus);
-    }
-
-    if (level == TASK_LEVEL__SOURCE && pTask->info.fillHistory && status == TASK_STATUS__SCAN_HISTORY) {
-      streamStartScanHistoryAsync(pTask, igUntreated);
-    } else if (level == TASK_LEVEL__SOURCE && (streamQueueGetNumOfItems(pTask->inputq.queue) == 0)) {
-      tqScanWalAsync(pTq, false);
-    } else {
-      streamSchedExec(pTask);
-    }
-  } else if (status == TASK_STATUS__UNINIT) {
-    // todo: fill-history task init ?
-    if (pTask->info.fillHistory == 0) {
-      EStreamTaskEvent event = /*HAS_RELATED_FILLHISTORY_TASK(pTask) ? TASK_EVENT_INIT_STREAM_SCANHIST : */TASK_EVENT_INIT;
-      streamTaskHandleEvent(pTask->status.pSM, event);
-    }
-  }
-
-  streamMetaReleaseTask(pTq->pStreamMeta, pTask);
-  return 0;
+  return tqStreamTaskProcessTaskPauseReq(pTq->pStreamMeta, msg);
 }
 
 int32_t tqProcessTaskResumeReq(STQ* pTq, int64_t sversion, char* msg, int32_t msgLen) {
-  SVResumeStreamTaskReq* pReq = (SVResumeStreamTaskReq*)msg;
-
-  SStreamTask* pTask = streamMetaAcquireTask(pTq->pStreamMeta, pReq->streamId, pReq->taskId);
-  int32_t      code = tqProcessTaskResumeImpl(pTq, pTask, sversion, pReq->igUntreated);
-  if (code != 0) {
-    return code;
-  }
-
-  STaskId*     pHTaskId = &pTask->hTaskInfo.id;
-  SStreamTask* pHistoryTask = streamMetaAcquireTask(pTq->pStreamMeta, pHTaskId->streamId, pHTaskId->taskId);
-  if (pHistoryTask) {
-    code = tqProcessTaskResumeImpl(pTq, pHistoryTask, sversion, pReq->igUntreated);
-  }
-
-  return code;
+  return tqStreamTaskProcessTaskResumeReq(pTq, sversion, msg, true);
 }
 
 int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg) {
diff --git a/source/dnode/vnode/src/tqCommon/tqCommon.c b/source/dnode/vnode/src/tqCommon/tqCommon.c
index a3d860fd78..70bfa00609 100644
--- a/source/dnode/vnode/src/tqCommon/tqCommon.c
+++ b/source/dnode/vnode/src/tqCommon/tqCommon.c
@@ -752,4 +752,109 @@ int32_t tqStreamTaskProcessTaskResetReq(SStreamMeta* pMeta, SRpcMsg* pMsg) {
 
   streamMetaReleaseTask(pMeta, pTask);
   return TSDB_CODE_SUCCESS;
+}
+
+int32_t tqStreamTaskProcessTaskPauseReq(SStreamMeta* pMeta, char* pMsg){
+  SVPauseStreamTaskReq* pReq = (SVPauseStreamTaskReq*)pMsg;
+
+  SStreamTask* pTask = streamMetaAcquireTask(pMeta, pReq->streamId, pReq->taskId);
+  if (pTask == NULL) {
+    tqError("vgId:%d process pause req, failed to acquire task:0x%x, it may have been dropped already", pMeta->vgId,
+            pReq->taskId);
+    // since task is in [STOP|DROPPING] state, it is safe to assume the pause is active
+    return TSDB_CODE_SUCCESS;
+  }
+
+  tqDebug("s-task:%s receive pause msg from mnode", pTask->id.idStr);
+  streamTaskPause(pTask, pMeta);
+
+  SStreamTask* pHistoryTask = NULL;
+  if (HAS_RELATED_FILLHISTORY_TASK(pTask)) {
+    pHistoryTask = streamMetaAcquireTask(pMeta, pTask->hTaskInfo.id.streamId, pTask->hTaskInfo.id.taskId);
+    if (pHistoryTask == NULL) {
+      tqError("vgId:%d process pause req, failed to acquire fill-history task:0x%" PRIx64
+                  ", it may have been dropped already",
+              pMeta->vgId, pTask->hTaskInfo.id.taskId);
+      streamMetaReleaseTask(pMeta, pTask);
+
+      // since task is in [STOP|DROPPING] state, it is safe to assume the pause is active
+      return TSDB_CODE_SUCCESS;
+    }
+
+    tqDebug("s-task:%s fill-history task handle paused along with related stream task", pHistoryTask->id.idStr);
+
+    streamTaskPause(pHistoryTask, pMeta);
+    streamMetaReleaseTask(pMeta, pHistoryTask);
+  }
+
+  streamMetaReleaseTask(pMeta, pTask);
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t tqProcessTaskResumeImpl(void* handle, SStreamTask* pTask, int64_t sversion, int8_t igUntreated, bool fromVnode) {
+  SStreamMeta *pMeta = fromVnode ? ((STQ*)handle)->pStreamMeta : handle;
+  int32_t vgId = pMeta->vgId;
+  if (pTask == NULL) {
+    return -1;
+  }
+
+  streamTaskResume(pTask);
+  ETaskStatus status = streamTaskGetStatus(pTask, NULL);
+
+  int32_t level = pTask->info.taskLevel;
+  if (level == TASK_LEVEL__SINK) {
+    if (status == TASK_STATUS__UNINIT) {
+    }
+    streamMetaReleaseTask(pMeta, pTask);
+    return 0;
+  }
+
+  if (status == TASK_STATUS__READY || status == TASK_STATUS__SCAN_HISTORY || status == TASK_STATUS__CK) {
+    // no lock needs to secure the access of the version
+    if (igUntreated && level == TASK_LEVEL__SOURCE && !pTask->info.fillHistory) {
+      // discard all the data  when the stream task is suspended.
+      walReaderSetSkipToVersion(pTask->exec.pWalReader, sversion);
+      tqDebug("vgId:%d s-task:%s resume to exec, prev paused version:%" PRId64 ", start from vnode ver:%" PRId64
+                  ", schedStatus:%d",
+              vgId, pTask->id.idStr, pTask->chkInfo.nextProcessVer, sversion, pTask->status.schedStatus);
+    } else {  // from the previous paused version and go on
+      tqDebug("vgId:%d s-task:%s resume to exec, from paused ver:%" PRId64 ", vnode ver:%" PRId64 ", schedStatus:%d",
+              vgId, pTask->id.idStr, pTask->chkInfo.nextProcessVer, sversion, pTask->status.schedStatus);
+    }
+
+    if (level == TASK_LEVEL__SOURCE && pTask->info.fillHistory && status == TASK_STATUS__SCAN_HISTORY) {
+      streamStartScanHistoryAsync(pTask, igUntreated);
+    } else if (level == TASK_LEVEL__SOURCE && (streamQueueGetNumOfItems(pTask->inputq.queue) == 0)) {
+      tqScanWalAsync((STQ*)handle, false);
+    } else {
+      streamSchedExec(pTask);
+    }
+  } else if (status == TASK_STATUS__UNINIT) {
+    // todo: fill-history task init ?
+    if (pTask->info.fillHistory == 0) {
+      EStreamTaskEvent event = /*HAS_RELATED_FILLHISTORY_TASK(pTask) ? TASK_EVENT_INIT_STREAM_SCANHIST : */TASK_EVENT_INIT;
+      streamTaskHandleEvent(pTask->status.pSM, event);
+    }
+  }
+
+  streamMetaReleaseTask(pMeta, pTask);
+  return 0;
+}
+
+int32_t tqStreamTaskProcessTaskResumeReq(void* handle, int64_t sversion, char* msg, bool fromVnode){
+  SVResumeStreamTaskReq* pReq = (SVResumeStreamTaskReq*)msg;
+  SStreamMeta *pMeta = fromVnode ? ((STQ*)handle)->pStreamMeta : handle;
+  SStreamTask* pTask = streamMetaAcquireTask(pMeta, pReq->streamId, pReq->taskId);
+  int32_t      code = tqProcessTaskResumeImpl(handle, pTask, sversion, pReq->igUntreated, fromVnode);
+  if (code != 0) {
+    return code;
+  }
+
+  STaskId*     pHTaskId = &pTask->hTaskInfo.id;
+  SStreamTask* pHistoryTask = streamMetaAcquireTask(pMeta, pHTaskId->streamId, pHTaskId->taskId);
+  if (pHistoryTask) {
+    code = tqProcessTaskResumeImpl(pMeta, pHistoryTask, sversion, pReq->igUntreated, fromVnode);
+  }
+
+  return code;
 }
\ No newline at end of file
diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c
index 7ea80ad8d8..4e30dc2555 100644
--- a/source/libs/stream/src/streamMeta.c
+++ b/source/libs/stream/src/streamMeta.c
@@ -377,16 +377,16 @@ SStreamMeta* streamMetaOpen(const char* path, void* ahandle, FTaskExpand expandF
   pMeta->rid = taosAddRef(streamMetaId, pMeta);
 
   // set the attribute when running on Linux OS
-#if defined LINUX
   TdThreadRwlockAttr attr;
   taosThreadRwlockAttrInit(&attr);
 
+#ifdef LINUX
   pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP);
-  taosThreadRwlockInit(&pMeta->lock, &attr);
-
-  taosThreadRwlockAttrDestroy(&attr);
 #endif
 
+  taosThreadRwlockInit(&pMeta->lock, &attr);
+  taosThreadRwlockAttrDestroy(&attr);
+
   int64_t* pRid = taosMemoryMalloc(sizeof(int64_t));
   memcpy(pRid, &pMeta->rid, sizeof(pMeta->rid));
   metaRefMgtAdd(pMeta->vgId, pRid);

From fb42c6afd5a06d7a748afa69df6e9436a31528f6 Mon Sep 17 00:00:00 2001
From: Alex Duan <417921451@qq.com>
Date: Wed, 10 Jan 2024 19:59:44 +0800
Subject: [PATCH 7/8] coverage: update coverage script file and BlockSMA
 trigger is 1

---
 tests/script/coverage_test.sh         | 363 +++++++++++---------------
 tests/system-test/2-query/blockSMA.py |   2 +-
 2 files changed, 160 insertions(+), 205 deletions(-)
 mode change 100755 => 100644 tests/script/coverage_test.sh

diff --git a/tests/script/coverage_test.sh b/tests/script/coverage_test.sh
old mode 100755
new mode 100644
index 535b6802f2..1b9e9753d3
--- a/tests/script/coverage_test.sh
+++ b/tests/script/coverage_test.sh
@@ -12,7 +12,6 @@ fi
 today=`date +"%Y%m%d"`
 TDENGINE_DIR=/root/TDengine
 JDBC_DIR=/root/taos-connector-jdbc
-TAOSKEEPER_DIR=/root/taoskeeper
 TDENGINE_COVERAGE_REPORT=$TDENGINE_DIR/tests/coverage-report-$today.log
 
 # Color setting
@@ -23,52 +22,52 @@ GREEN_UNDERLINE='\033[4;32m'
 NC='\033[0m'
 
 function buildTDengine() {
-	echo "check if TDengine need build"
-	cd $TDENGINE_DIR
-  	git remote prune origin > /dev/null
-	git remote update > /dev/null
-	REMOTE_COMMIT=`git rev-parse --short remotes/origin/$branch`
-	LOCAL_COMMIT=`git rev-parse --short @`
-	echo " LOCAL: $LOCAL_COMMIT"
-	echo "REMOTE: $REMOTE_COMMIT"
+    echo "check if TDengine need build"
+    cd $TDENGINE_DIR
+    git remote prune origin > /dev/null
+    git remote update > /dev/null
+    REMOTE_COMMIT=`git rev-parse --short remotes/origin/$branch`
+    LOCAL_COMMIT=`git rev-parse --short @`
+    echo " LOCAL: $LOCAL_COMMIT"
+    echo "REMOTE: $REMOTE_COMMIT"
 
-	# reset counter
-	lcov -d . --zerocounters
+    # reset counter
+    lcov -d . --zerocounters
 
-	if [ "$LOCAL_COMMIT" == "$REMOTE_COMMIT" ]; then
-		echo "repo up-to-date"
-	else
-		echo "repo need to pull"
-	fi
+    if [ "$LOCAL_COMMIT" == "$REMOTE_COMMIT" ]; then
+        echo "repo up-to-date"
+    else
+        echo "repo need to pull"
+    fi
 
-	git reset --hard
-	git checkout -- .
-	git checkout $branch
-	git checkout -- .
-	git clean -dfx
-	git pull    
- 	
-  	[ -d $TDENGINE_DIR/debug ] || mkdir $TDENGINE_DIR/debug
-  	cd $TDENGINE_DIR/debug
+    git reset --hard
+    git checkout -- .
+    git checkout $branch
+    git checkout -- .
+    git clean -dfx
+    git pull
+
+      [ -d $TDENGINE_DIR/debug ] || mkdir $TDENGINE_DIR/debug
+      cd $TDENGINE_DIR/debug
 
     echo "rebuild.."
-	LOCAL_COMMIT=`git rev-parse --short @`
+    LOCAL_COMMIT=`git rev-parse --short @`
 
-	rm -rf *
-	if [ "$branch" == "3.0" ]; then
-		echo "3.0 ============="
-		cmake -DCOVER=true -DBUILD_TEST=true -DBUILD_HTTP=false -DBUILD_TOOLS=true .. > /dev/null
-	else
-		cmake -DCOVER=true -DBUILD_TEST=true -DBUILD_TOOLS=true -DBUILD_HTTP=false .. > /dev/null
-	fi
-	make -j
-	make install
+    rm -rf *
+    if [ "$branch" == "3.0" ]; then
+        echo "3.0 ============="
+        cmake -DCOVER=true -DBUILD_TEST=true -DBUILD_HTTP=false -DBUILD_TOOLS=true -DBUILD_GEOS=true -DBUILD_CONTRIB=true ..
+    else
+        cmake -DCOVER=true -DBUILD_TOOLS=true -DBUILD_HTTP=false .. > /dev/null
+    fi
+    make -j
+    make install
 }
 
 function runCasesOneByOne () {
-	while read -r line; do
-		if [[ "$line" != "#"* ]]; then
-			cmd=`echo $line | cut -d',' -f 5`
+    while read -r line; do
+        if [[ "$line" != "#"* ]]; then
+            cmd=`echo $line | cut -d',' -f 5`
             if [[ "$2" == "sim" ]] && [[ $line == *"script"* ]]; then
                 case=`echo $cmd | cut -d' ' -f 3`
                 start_time=`date +%s`
@@ -77,18 +76,7 @@ function runCasesOneByOne () {
                 || echo -e "${RED}$case failed${NC}" | tee -a  $TDENGINE_COVERAGE_REPORT
                 end_time=`date +%s`
                 echo execution time of $case was `expr $end_time - $start_time`s. | tee -a $TDENGINE_COVERAGE_REPORT
-            elif [[ "$2" == "system-test" ]] && [[ $line == *"system-test"* ]]; then
-                if [[ "$cmd" == *"pytest.sh"* ]]; then
-                    cmd=`echo $cmd | cut -d' ' -f 2-20`
-                fi
-                case=`echo $cmd | cut -d' ' -f 4-20`
-                start_time=`date +%s`
-                date +%F\ %T | tee -a $TDENGINE_COVERAGE_REPORT && $cmd > /dev/null 2>&1 && \
-                echo -e "${GREEN}$case success${NC}" | tee -a $TDENGINE_COVERAGE_REPORT || \
-                echo -e "${RED}$case failed${NC}" | tee -a $TDENGINE_COVERAGE_REPORT
-                end_time=`date +%s`
-                echo execution time of $case was `expr $end_time - $start_time`s. | tee -a $TDENGINE_COVERAGE_REPORT
-            elif [[ "$2" == "develop-test" ]] && [[ $line == *"develop-test"* ]]; then
+            elif [$line == *"$2"* ]; then
                 if [[ "$cmd" == *"pytest.sh"* ]]; then
                     cmd=`echo $cmd | cut -d' ' -f 2-20`
                 fi
@@ -101,209 +89,176 @@ function runCasesOneByOne () {
                 echo execution time of $case was `expr $end_time - $start_time`s. | tee -a $TDENGINE_COVERAGE_REPORT
             fi
         fi
-	done < $1
+    done < $1
 }
 
 function runUnitTest() {
-	echo "=== Run unit test case ==="
-	echo " $TDENGINE_DIR/debug"
-	cd $TDENGINE_DIR/debug
-	ctest -j12
-
-	echo " $TDENGINE_DIR/tests/script/api"
-	cd $TDENGINE_DIR/tests/script/api
-	make clean && make
-	
-	stopTaosd
-	stopTaosadapter
-
-	nohup taosd -c /etc/taos >> /dev/null 2>&1 &
-	./batchprepare 127.0.0.1
-	./dbTableRoute 127.0.0.1
-	./stopquery 127.0.0.1 demo t1
-	
-	echo "3.0 unit test done"
+    echo "=== Run unit test case ==="
+    echo " $TDENGINE_DIR/debug"
+    cd $TDENGINE_DIR/debug
+    ctest -j12
+    echo "3.0 unit test done"
 }
 
 function runSimCases() {
-	echo "=== Run sim cases ==="
-	
-	cd $TDENGINE_DIR/tests/script
-	runCasesOneByOne ../parallel_test/cases.task sim
-	
-	totalSuccess=`grep 'sim success' $TDENGINE_COVERAGE_REPORT | wc -l`
-	if [ "$totalSuccess" -gt "0" ]; then
-		echo "### Total $totalSuccess SIM test case(s) succeed! ###" | tee -a $TDENGINE_COVERAGE_REPORT
-	fi
+    echo "=== Run sim cases ==="
 
-	totalFailed=`grep 'sim failed\|fault' $TDENGINE_COVERAGE_REPORT | wc -l`
-	if [ "$totalFailed" -ne "0" ]; then
-		echo "### Total $totalFailed SIM test case(s) failed! ###" | tee -a $TDENGINE_COVERAGE_REPORT
-	fi
+    cd $TDENGINE_DIR/tests/script
+    runCasesOneByOne $TDENGINE_DIR/tests/parallel_test/cases.task sim
+
+    totalSuccess=`grep 'sim success' $TDENGINE_COVERAGE_REPORT | wc -l`
+    if [ "$totalSuccess" -gt "0" ]; then
+        echo "### Total $totalSuccess SIM test case(s) succeed! ###" | tee -a $TDENGINE_COVERAGE_REPORT
+    fi
+
+    totalFailed=`grep 'sim failed\|fault' $TDENGINE_COVERAGE_REPORT | wc -l`
+    if [ "$totalFailed" -ne "0" ]; then
+        echo "### Total $totalFailed SIM test case(s) failed! ###" | tee -a $TDENGINE_COVERAGE_REPORT
+    fi
 }
 
 function runPythonCases() {
-	echo "=== Run python cases ==="
+    echo "=== Run python cases ==="
 
-	cd $TDENGINE_DIR/tests/parallel_test
-	sed -i '/compatibility.py/d' cases.task
-		
-	cd $TDENGINE_DIR/tests/system-test
-	runCasesOneByOne ../parallel_test/cases.task system-test
+    cd $TDENGINE_DIR/tests/parallel_test
+    sed -i '/compatibility.py/d' cases.task
 
-	cd $TDENGINE_DIR/tests/develop-test
-	runCasesOneByOne ../parallel_test/cases.task develop-test
+    # army
+    cd $TDENGINE_DIR/tests/army
+    runCasesOneByOne ../parallel_test/cases.task army
 
-	totalSuccess=`grep 'py success' $TDENGINE_COVERAGE_REPORT | wc -l`
-	if [ "$totalSuccess" -gt "0" ]; then
-		echo "### Total $totalSuccess python test case(s) succeed! ###" | tee -a $TDENGINE_COVERAGE_REPORT
-	fi
+    # system-test
+    cd $TDENGINE_DIR/tests/system-test
+    runCasesOneByOne ../parallel_test/cases.task system-test
 
-	totalFailed=`grep 'py failed\|fault' $TDENGINE_COVERAGE_REPORT | wc -l`
-	if [ "$totalFailed" -ne "0" ]; then
-		echo "### Total $totalFailed python test case(s) failed! ###" | tee -a $TDENGINE_COVERAGE_REPORT
-	fi
+    # develop-test
+    cd $TDENGINE_DIR/tests/develop-test
+    runCasesOneByOne ../parallel_test/cases.task develop-test
+
+    totalSuccess=`grep 'py success' $TDENGINE_COVERAGE_REPORT | wc -l`
+    if [ "$totalSuccess" -gt "0" ]; then
+        echo "### Total $totalSuccess python test case(s) succeed! ###" | tee -a $TDENGINE_COVERAGE_REPORT
+    fi
+
+    totalFailed=`grep 'py failed\|fault' $TDENGINE_COVERAGE_REPORT | wc -l`
+    if [ "$totalFailed" -ne "0" ]; then
+        echo "### Total $totalFailed python test case(s) failed! ###" | tee -a $TDENGINE_COVERAGE_REPORT
+    fi
 }
 
 function runJDBCCases() {
-	echo "=== Run JDBC cases ==="
+    echo "=== Run JDBC cases ==="
 
-	cd $JDBC_DIR
-	git checkout -- .
-	git reset --hard HEAD
-	git checkout main
-	git pull
+    cd $JDBC_DIR
+    git checkout -- .
+    git reset --hard HEAD
+    git checkout main
+    git pull
 
-	stopTaosd
-	stopTaosadapter
+    stopTaosd
+    stopTaosadapter
 
-	nohup taosd -c /etc/taos >> /dev/null 2>&1 &
-	nohup taosadapter >> /dev/null 2>&1 &
+    nohup $TDENGINE_DIR/debug/build/bin/taosd -c /etc/taos >> /dev/null 2>&1 &
+    nohup taosadapter >> /dev/null 2>&1 &
 
-	mvn clean test > result.txt 2>&1
-	summary=`grep "Tests run:" result.txt | tail -n 1`		
-	echo -e "### JDBC test result: $summary ###" | tee -a $TDENGINE_COVERAGE_REPORT
-}
-
-function runTaosKeeperCases() {
-	echo "=== Run taoskeeper cases ==="
-
-	cd $TAOSKEEPER_DIR
-	git checkout -- .
-	git reset --hard HEAD
-	git checkout master
-	git pull
-
-	stopTaosd
-	stopTaosadapter
-
-	taosd -c /etc/taos >> /dev/null 2>&1 &
-	taosadapter >> /dev/null 2>&1 &
-
-	go mod tidy && go test -v ./...
+    mvn clean test > result.txt 2>&1
+    summary=`grep "Tests run:" result.txt | tail -n 1`
+    echo -e "### JDBC test result: $summary ###" | tee -a $TDENGINE_COVERAGE_REPORT
 }
 
 function runTest() {
-	echo "run Test"
-	
-	cd $TDENGINE_DIR
-	[ -d sim ] && rm -rf sim
+    echo "run Test"
+
+    cd $TDENGINE_DIR
+    [ -d sim ] && rm -rf sim
     [ -f $TDENGINE_COVERAGE_REPORT ] && rm $TDENGINE_COVERAGE_REPORT
 
-	runUnitTest
-	runSimCases
-	runPythonCases
-	runJDBCCases
-	runTaosKeeperCases
-	
-	stopTaosd	
-	cd $TDENGINE_DIR/tests/script
-	find . -name '*.sql' | xargs rm -f
+    runUnitTest
+    runSimCases
+    runPythonCases
+    runJDBCCases
 
-	cd $TDENGINE_DIR/tests/pytest
-	find . -name '*.sql' | xargs rm -f
+    stopTaosd  
+    cd $TDENGINE_DIR/tests/script
+    find . -name '*.sql' | xargs rm -f
+
+    cd $TDENGINE_DIR/tests/pytest
+    find . -name '*.sql' | xargs rm -f
 }
 
 function lcovFunc {
-	echo "collect data by lcov"
-	cd $TDENGINE_DIR
+    echo "collect data by lcov"
+    cd $TDENGINE_DIR
 
-	# collect data
-	lcov -d . --capture --rc lcov_branch_coverage=1 --rc genhtml_branch_coverage=1 --no-external -b $TDENGINE_DIR -o coverage.info
+    # collect data
+    lcov --ignore-errors negative --ignore-errors mismatch -d . --capture --rc lcov_branch_coverage=1 --rc branch_coverage=1 --no-external -b $TDENGINE_DIR -o coverage.info
 
-	# remove exclude paths
-	if [ "$branch" == "main" ] ; then
-		lcov --remove coverage.info \
-			'*/contrib/*' '*/tests/*' '*/test/*' '*/tools/*' '*/libs/sync/*'\
-			'*/AccessBridgeCalls.c' '*/ttszip.c' '*/dataInserter.c' '*/tlinearhash.c' '*/tsimplehash.c' '*/tsdbDiskData.c'\
-			'*/texpr.c' '*/runUdf.c' '*/schDbg.c' '*/syncIO.c' '*/tdbOs.c' '*/pushServer.c' '*/osLz4.c'\
-			'*/tbase64.c' '*/tbuffer.c' '*/tdes.c' '*/texception.c' '*/tidpool.c' '*/tmempool.c'\
-			'*/clientJniConnector.c' '*/clientTmqConnector.c' '*/version.c' '*/shellAuto.c' '*/shellTire.c'\
-			'*/tthread.c' '*/tversion.c' '*/ctgDbg.c' '*/schDbg.c' '*/qwDbg.c' '*/tencode.h' '*/catalog.c'\
-			'*/tqSnapshot.c' '*/tsdbSnapshot.c''*/metaSnapshot.c' '*/smaSnapshot.c' '*/tqOffsetSnapshot.c'\
-			'*/vnodeSnapshot.c' '*/metaSnapshot.c' '*/tsdbSnapshot.c' '*/mndGrant.c' '*/mndSnode.c' '*/streamRecover.c'\
-			'*/osAtomic.c' '*/osDir.c' '*/osFile.c' '*/osMath.c' '*/osSignal.c' '*/osSleep.c' '*/osString.c' '*/osSystem.c'\
-			'*/osThread.c' '*/osTime.c' '*/osTimezone.c' \
-		       	--rc lcov_branch_coverage=1 -o coverage.info
-	else	
-		lcov --remove coverage.info \
-    		'*/tests/*' '*/test/*' '*/deps/*' '*/plugins/*' '*/taosdef.h' '*/ttype.h' '*/tarithoperator.c' '*/TSDBJNIConnector.c' '*/taosdemo.c' '*/clientJniConnector.c'\
-    		--rc lcov_branch_coverage=1 -o coverage.info
-	fi
+    # remove exclude paths
+    lcov --remove coverage.info \
+        '*/contrib/*' '*/tests/*' '*/test/*' '*/packaging/*' '*/taos-tools/*' '*/taosadapter/*' '*/TSZ/*' \
+        '*/AccessBridgeCalls.c' '*/ttszip.c' '*/dataInserter.c' '*/tlinearhash.c' '*/tsimplehash.c' '*/tsdbDiskData.c'\
+        '*/texpr.c' '*/runUdf.c' '*/schDbg.c' '*/syncIO.c' '*/tdbOs.c' '*/pushServer.c' '*/osLz4.c'\
+        '*/tbase64.c' '*/tbuffer.c' '*/tdes.c' '*/texception.c' '*/examples/*' '*/tidpool.c' '*/tmempool.c'\
+        '*/clientJniConnector.c' '*/clientTmqConnector.c' '*/version.c'\
+        '*/tthread.c' '*/tversion.c'  '*/ctgDbg.c' '*/schDbg.c' '*/qwDbg.c' '*/tencode.h' \
+        '*/shellAuto.c' '*/shellTire.c' '*/shellCommand.c'\
+         --rc branch_coverage=1 -o coverage.info
 
+    # generate result
+    echo "generate result"
+    lcov -l --rc branch_coverage=1 coverage.info | tee -a $TDENGINE_COVERAGE_REPORT
 
-	# generate result
-	echo "generate result"
-	lcov -l --rc lcov_branch_coverage=1 coverage.info | tee -a $TDENGINE_COVERAGE_REPORT
+    sed -i 's/\/root\/TDengine\/sql.c/\/root\/TDengine\/source\/libs\/parser\/inc\/sql.c/g' coverage.info
+    sed -i 's/\/root\/TDengine\/sql.y/\/root\/TDengine\/source\/libs\/parser\/inc\/sql.y/g' coverage.info
 
-	# push result to coveralls.io
-	echo "push result to coveralls.io"
-	/usr/local/bin/coveralls-lcov coverage.info -t o7uY02qEAgKyJHrkxLGiCOTfL3IGQR2zm | tee -a $TDENGINE_COVERAGE_REPORT
+    # push result to coveralls.io
+    echo "push result to coveralls.io"
+    /usr/local/bin/coveralls-lcov coverage.info -t o7uY02qEAgKyJHrkxLGiCOTfL3IGQR2zm | tee -a $TDENGINE_COVERAGE_REPORT
+
+    #/root/pxiao/checkCoverageFile.sh -s $TDENGINE_DIR/source -f $TDENGINE_COVERAGE_REPORT
+    #cat /root/pxiao/fileListNoCoverage.log | tee -a $TDENGINE_COVERAGE_REPORT
+    cat $TDENGINE_COVERAGE_REPORT | grep "| 0.0%" | awk -F "%" '{print $1}' | awk -F "|" '{if($2==0.0)print $1}' | tee -a $TDENGINE_COVERAGE_REPORT
 
-	#/root/pxiao/checkCoverageFile.sh -s $TDENGINE_DIR/source -f $TDENGINE_COVERAGE_REPORT
-	#cat /root/pxiao/fileListNoCoverage.log | tee -a $TDENGINE_COVERAGE_REPORT
-	cat $TDENGINE_COVERAGE_REPORT | grep "| 0.0%" | awk -F "%" '{print $1}' | awk -F "|" '{if($2==0.0)print $1}' | tee -a $TDENGINE_COVERAGE_REPORT
-	
 }
 
 function sendReport {
-	echo "send report"
-	receiver="develop@taosdata.com"
-	mimebody="MIME-Version: 1.0\nContent-Type: text/html; charset=utf-8\n"
+    echo "send report"
+    receiver="develop@taosdata.com"
+    mimebody="MIME-Version: 1.0\nContent-Type: text/html; charset=utf-8\n"
 
-	cd $TDENGINE_DIR
+    cd $TDENGINE_DIR
 
-	sed -i 's/\x1b\[[0-9;]*m//g' $TDENGINE_COVERAGE_REPORT
-	BODY_CONTENT=`cat $TDENGINE_COVERAGE_REPORT`
-	echo -e "from: <support@taosdata.com>\nto: ${receiver}\nsubject: Coverage test report ${branch} ${today}, commit ID: ${LOCAL_COMMIT}\n\n${today}:\n${BODY_CONTENT}" | \
-	(cat - && uuencode $TDENGINE_COVERAGE_REPORT coverage-report-$today.log) | \
-	/usr/sbin/ssmtp "${receiver}" && echo "Report Sent!"
+    sed -i 's/\x1b\[[0-9;]*m//g' $TDENGINE_COVERAGE_REPORT
+    BODY_CONTENT=`cat $TDENGINE_COVERAGE_REPORT`
+    echo -e "from: <support@taosdata.com>\nto: ${receiver}\nsubject: Coverage test report ${branch} ${today}, commit ID: ${LOCAL_COMMIT}\n\n${today}:\n${BODY_CONTENT}" | \
+    (cat - && uuencode $TDENGINE_COVERAGE_REPORT coverage-report-$today.log) | \
+    /usr/sbin/ssmtp "${receiver}" && echo "Report Sent!"
 }
 
 function stopTaosd {
-	echo "Stop taosd start"
+    echo "Stop taosd start"
         systemctl stop taosd
-  	PID=`ps -ef|grep -w taosd | grep -v grep | awk '{print $2}'`
-	while [ -n "$PID" ]
-	do
+      PID=`ps -ef|grep -w taosd | grep -v grep | awk '{print $2}'`
+    while [ -n "$PID" ]
+    do
     pkill -TERM -x taosd
     sleep 1
-  	PID=`ps -ef|grep -w taosd | grep -v grep | awk '{print $2}'`
-	done
-	echo "Stop tasod end"
+      PID=`ps -ef|grep -w taosd | grep -v grep | awk '{print $2}'`
+    done
+    echo "Stop tasod end"
 }
 
 function stopTaosadapter {
-	echo "Stop taosadapter"
-	systemctl stop taosadapter.service
-	PID=`ps -ef|grep -w taosadapter | grep -v grep | awk '{print $2}'`
-	while [ -n "$PID" ]
-	do
-		pkill -TERM -x taosadapter
-		sleep 1
-		PID=`ps -ef|grep -w taosd | grep -v grep | awk '{print $2}'`
-	done
-	echo "Stop tasoadapter end"
+    echo "Stop taosadapter"
+    systemctl stop taosadapter.service
+    PID=`ps -ef|grep -w taosadapter | grep -v grep | awk '{print $2}'`
+    while [ -n "$PID" ]
+    do
+        pkill -TERM -x taosadapter
+        sleep 1
+        PID=`ps -ef|grep -w taosd | grep -v grep | awk '{print $2}'`
+    done
+    echo "Stop tasoadapter end"
 
 }
 
@@ -318,7 +273,7 @@ buildTDengine
 runTest
 lcovFunc
 
-sendReport
+#sendReport
 stopTaosd
 
 date >> $WORK_DIR/cron.log
diff --git a/tests/system-test/2-query/blockSMA.py b/tests/system-test/2-query/blockSMA.py
index 85c0189e27..605542cb3c 100644
--- a/tests/system-test/2-query/blockSMA.py
+++ b/tests/system-test/2-query/blockSMA.py
@@ -16,7 +16,7 @@ class TDTestCase:
 
     def run(self):
         dbname = "db"
-        tdSql.prepare()
+        tdSql.prepare(dbname=dbname, drop=True, stt_trigger=1)
 
         tdSql.execute(f'''create table {dbname}.ntb(ts timestamp, col1 tinyint, col2 smallint, col3 int, col4 bigint, col5 float, col6 double,
                     col7 bool, col8 binary(20), col9 nchar(20), col11 tinyint unsigned, col12 smallint unsigned, col13 int unsigned, col14 bigint unsigned)''')

From 216830f8b4e35254e707fbf23189e77d4ee05f83 Mon Sep 17 00:00:00 2001
From: wangmm0220 <wangmm0220@gmail.com>
Date: Wed, 10 Jan 2024 20:33:13 +0800
Subject: [PATCH 8/8] fix:[TD-28185]add pause & resume logic for snode

---
 source/dnode/vnode/src/tqCommon/tqCommon.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/source/dnode/vnode/src/tqCommon/tqCommon.c b/source/dnode/vnode/src/tqCommon/tqCommon.c
index 70bfa00609..399788e731 100644
--- a/source/dnode/vnode/src/tqCommon/tqCommon.c
+++ b/source/dnode/vnode/src/tqCommon/tqCommon.c
@@ -853,7 +853,7 @@ int32_t tqStreamTaskProcessTaskResumeReq(void* handle, int64_t sversion, char* m
   STaskId*     pHTaskId = &pTask->hTaskInfo.id;
   SStreamTask* pHistoryTask = streamMetaAcquireTask(pMeta, pHTaskId->streamId, pHTaskId->taskId);
   if (pHistoryTask) {
-    code = tqProcessTaskResumeImpl(pMeta, pHistoryTask, sversion, pReq->igUntreated, fromVnode);
+    code = tqProcessTaskResumeImpl(handle, pHistoryTask, sversion, pReq->igUntreated, fromVnode);
   }
 
   return code;