Merge pull request #22355 from taosdata/3.0

merge 3.0 to TD-22970

docs/en/07-develop/01-connect/index.md

@@ -83,7 +83,7 @@ If `maven` is used to manage the projects, what needs to be done is only adding
 <dependency>
   <groupId>com.taosdata.jdbc</groupId>
   <artifactId>taos-jdbcdriver</artifactId>
-  <version>3.2.1</version>
+  <version>3.2.4</version>
 </dependency>
 ```
 

docs/en/20-third-party/50-qstudio.md

@@ -0,0 +1,40 @@
+---
+sidebar_label: qStudio
+title: qStudio
+description: Step-by-Step Guide to Accessing TDengine Data with qStudio
+---
+
+qStudio is a free cross-platform SQL data analysis tool that allows easy browsing of tables, variables, functions, and configuration settings in a database. The latest version of qStudio includes built-in support for TDengine.
+
+## Prerequisites
+
+To connect TDengine using qStudio, you need to complete the following preparations:
+
+- Install qStudio: qStudio supports major operating systems, including Windows, macOS, and Linux. Please ensure you download the correct installation package for your platform from the [download page](https://www.timestored.com/qstudio/download/).
+- Set up TDengine instance: Make sure TDengine is installed and running correctly, and the taosAdapter is installed and running. For detailed information, refer to the taosAdapter User Manual.
+
+## Connecting to TDengine with qStudio
+
+1. Launch the qStudio application and select "Server" and then "Add Server..." from the menu. Choose TDengine from the Server Type dropdown.
+
+ ![qConnecting TDengine with qStudio](./qstudio/qstudio-connect-tdengine.webp)
+
+2. Configure the TDengine connection by entering the host address, port number, username, and password. If TDengine is deployed on the local machine, you can fill in the username and password only. The default username is "root," and the default password is "taosdata." Click "Test" to test the connection's availability. If the TDengine Java connector is not installed on the local machine, qStudio will prompt you to download and install it.
+
+ ![Download Java Connector](./qstudio/qstudio-jdbc-connector-download.webp)
+
+3. Once connected successfully, the screen will display as shown below. If the connection fails, check that the TDengine service and taosAdapter are running correctly, and ensure that the host address, port number, username, and password are correct.
+
+ ![Successful Connection](./qstudio/qstudio-connect-tdengine-test.webp)
+
+4. Use qStudio to select databases and tables to browse data from the TDengine server.
+
+ ![Browsing TDengine Data with qStudio](./qstudio/qstudio-browse-data.webp)
+
+5. You can also perform operations on TDengine data by executing SQL commands.
+
+ ![qStudio SQL Commands](./qstudio/qstudio-sql-execution.webp)
+
+6. qStudio supports charting functions based on the data. For more information, please refer to the [qStudio documentation](https://www.timestored.com/qstudio/help).
+
+ ![qStudio Chart](./qstudio/qstudio-chart.webp)

docs/examples/java/pom.xml

@@ -22,7 +22,7 @@
     <dependency>
       <groupId>com.taosdata.jdbc</groupId>
       <artifactId>taos-jdbcdriver</artifactId>
-      <version>3.2.1</version>
+      <version>3.2.4</version>
     </dependency>
     <!--    ANCHOR_END: dep-->
     <dependency>
@@ -33,4 +33,4 @@
     </dependency>
   </dependencies>
 
-</project>
+</project>

docs/zh/07-develop/01-connect/index.md

@@ -82,7 +82,7 @@ TDengine 提供了丰富的应用程序开发接口，为了便于用户快速
 <dependency>
   <groupId>com.taosdata.jdbc</groupId>
   <artifactId>taos-jdbcdriver</artifactId>
-  <version>3.2.1</version>
+  <version>3.2.4</version>
 </dependency>
 ```
 

docs/zh/08-connector/14-java.mdx

@@ -446,7 +446,7 @@ TDengine 的 JDBC 原生连接实现大幅改进了参数绑定方式对数据
 **注意**：
 
 - JDBC REST 连接目前不支持参数绑定
-- 以下示例代码基于 taos-jdbcdriver-3.2.1
+- 以下示例代码基于 taos-jdbcdriver-3.2.4
 - binary 类型数据需要调用 setString 方法，nchar 类型数据需要调用 setNString 方法
 - 预处理语句中指定数据库与子表名称不要使用 `db.?`，应直接使用 `?`，然后在 setTableName 中指定数据库，如：`prepareStatement.setTableName("db.t1")`。
 

docs/zh/20-third-party/50-qstudio.md

@@ -0,0 +1,41 @@
+---
+sidebar_label: qStudio
+title: qStudio
+description: 使用 qStudio 存取 TDengine 数据的详细指南
+---
+
+qStudio 是一款免费的多平台 SQL 数据分析工具，可以轻松浏览数据库中的表、变量、函数和配置设置。最新版本 qStudio 内嵌支持 TDengine。
+
+## 前置条件
+
+使用 qStudio 连接 TDengine 需要以下几方面的准备工作。
+
+- 安装 qStudio。qStudio 支持主流操作系统包括 Windows、macOS 和 Linux。请注意[下载](https://www.timestored.com/qstudio/download/)正确平台的安装包。
+- 安装 TDengine 实例，请确认 TDengine 正常运行，并且 taosAdapter 已经安装并正常运行，具体细节请参考 [taosAdapter 的使用手册](/reference/taosadapter)。
+
+## 使用 qStudio 连接 TDengine
+
+1. 启动 qStudio 应用，从菜单项选择“Server” 和 “Add Server...”，然后在 Server Type 下拉框中选择 TDengine。
+
+ ![qStudio 连接 TDengine](./qstudio/qstudio-connect-tdengine.webp)
+
+2. 配置 TDengine 连接，填入主机地址、端口号、用户名和密码。如果 TDengine 部署在本机，可以只填用户名和密码，默认用户名为 root，默认密码为 taosdata。点击“Test”可以对连接是否可用进行测试。如果本机没有安装 TDengine Java
+ 连接器，qStudio 会提示下载安装。
+
+ ![下载 Java 连接器](./qstudio/qstudio-jdbc-connector-download.webp)
+
+3. 连接成功将显示如下图所示。如果显示连接失败，请检查 TDengine 服务和 taosAdapter 是否正确运行，主机地址、端口号、用户名和密码是否正确。
+
+ ![连接成功](./qstudio/qstudio-connect-tdengine-test.webp)
+
+4. 使用 qStudio 选择数据库和表可以浏览 TDengine 服务的数据。
+
+ ![qStudio 浏览 TDengine 数据](./qstudio/qstudio-browse-data.webp)
+
+5. 也可以通过执行 SQL 命令的方式对 TDengine 数据进行操作。
+
+ ![qStudio SQL 命令](./qstudio/qstudio-sql-execution.webp)
+
+6. qStudio 支持根据数据绘制图表等功能，请参考 [qStudio 的帮助文档](https://www.timestored.com/qstudio/help)
+
+ ![qStudio Chart 功能](./qstudio/qstudio-chart.webp)

source/client/test/clientTests.cpp

@@ -1442,4 +1442,178 @@ TEST(clientCase, sub_tb_mt_test) {
   }
 }
 
+TEST(clientCase, ts_3756) {
+//  taos_options(TSDB_OPTION_CONFIGDIR, "~/first/cfg");
+
+  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  ASSERT_NE(pConn, nullptr);
+
+  tmq_conf_t* conf = tmq_conf_new();
+
+  tmq_conf_set(conf, "enable.auto.commit", "false");
+  tmq_conf_set(conf, "auto.commit.interval.ms", "2000");
+  tmq_conf_set(conf, "group.id", "group_id_2");
+  tmq_conf_set(conf, "td.connect.user", "root");
+  tmq_conf_set(conf, "td.connect.pass", "taosdata");
+  tmq_conf_set(conf, "auto.offset.reset", "latest");
+  tmq_conf_set(conf, "msg.with.table.name", "false");
+
+  tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
+  tmq_conf_destroy(conf);
+
+  // 创建订阅 topics 列表
+  tmq_list_t* topicList = tmq_list_new();
+  tmq_list_append(topicList, "tp");
+
+  // 启动订阅
+  tmq_subscribe(tmq, topicList);
+  tmq_list_destroy(topicList);
+
+  TAOS_FIELD* fields = NULL;
+  int32_t     numOfFields = 0;
+  int32_t     precision = 0;
+  int32_t     totalRows = 0;
+  int32_t     msgCnt = 0;
+  int32_t     timeout = 200;
+
+  int32_t count = 0;
+
+  tmq_topic_assignment* pAssign = NULL;
+  int32_t numOfAssign = 0;
+
+  int32_t code = tmq_get_topic_assignment(tmq, "tp", &pAssign, &numOfAssign);
+  if (code != 0) {
+    printf("error occurs:%s\n", tmq_err2str(code));
+    tmq_free_assignment(pAssign);
+    tmq_consumer_close(tmq);
+    taos_close(pConn);
+    fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
+    return;
+  }
+
+  for(int i = 0; i < numOfAssign; i++){
+    printf("assign i:%d, vgId:%d, offset:%lld, start:%lld, end:%lld\n", i, pAssign[i].vgId, pAssign[i].currentOffset, pAssign[i].begin, pAssign[i].end);
+  }
+
+//  tmq_offset_seek(tmq, "tp", pAssign[0].vgId, 4);
+  tmq_free_assignment(pAssign);
+
+  code = tmq_get_topic_assignment(tmq, "tp", &pAssign, &numOfAssign);
+  if (code != 0) {
+    printf("error occurs:%s\n", tmq_err2str(code));
+    tmq_free_assignment(pAssign);
+    tmq_consumer_close(tmq);
+    taos_close(pConn);
+    fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
+    return;
+  }
+
+  for(int i = 0; i < numOfAssign; i++){
+    printf("assign i:%d, vgId:%d, offset:%lld, start:%lld, end:%lld\n", i, pAssign[i].vgId, pAssign[i].currentOffset, pAssign[i].begin, pAssign[i].end);
+  }
+
+  tmq_free_assignment(pAssign);
+
+  code = tmq_get_topic_assignment(tmq, "tp", &pAssign, &numOfAssign);
+  if (code != 0) {
+    printf("error occurs:%s\n", tmq_err2str(code));
+    tmq_free_assignment(pAssign);
+    tmq_consumer_close(tmq);
+    taos_close(pConn);
+    fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
+    return;
+  }
+
+  for(int i = 0; i < numOfAssign; i++){
+    printf("assign i:%d, vgId:%d, offset:%lld, start:%lld, end:%lld\n", i, pAssign[i].vgId, pAssign[i].currentOffset, pAssign[i].begin, pAssign[i].end);
+  }
+
+  while (1) {
+    printf("start to poll\n");
+    TAOS_RES* pRes = tmq_consumer_poll(tmq, timeout);
+    if (pRes) {
+      char buf[128];
+
+      const char* topicName = tmq_get_topic_name(pRes);
+//      const char* dbName = tmq_get_db_name(pRes);
+//      int32_t     vgroupId = tmq_get_vgroup_id(pRes);
+//
+//      printf("topic: %s\n", topicName);
+//      printf("db: %s\n", dbName);
+//      printf("vgroup id: %d\n", vgroupId);
+
+      printSubResults(pRes, &totalRows);
+
+      tmq_topic_assignment* pAssignTmp = NULL;
+      int32_t numOfAssignTmp = 0;
+
+      code = tmq_get_topic_assignment(tmq, "tp", &pAssignTmp, &numOfAssignTmp);
+      if (code != 0) {
+        printf("error occurs:%s\n", tmq_err2str(code));
+        tmq_free_assignment(pAssign);
+        tmq_consumer_close(tmq);
+        taos_close(pConn);
+        fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
+        return;
+      }
+
+      for(int i = 0; i < numOfAssign; i++){
+        printf("assign i:%d, vgId:%d, offset:%lld, start:%lld, end:%lld\n", i, pAssignTmp[i].vgId, pAssignTmp[i].currentOffset, pAssignTmp[i].begin, pAssignTmp[i].end);
+      }
+      if(numOfAssign != 0){
+        int i = 0;
+        for(; i < numOfAssign; i++){
+          if(pAssign[i].currentOffset != pAssignTmp[i].currentOffset){
+            break;
+          }
+        }
+        if(i == numOfAssign){
+          printf("all position is same\n");
+          break;
+        }
+        tmq_free_assignment(pAssign);
+      }
+      numOfAssign = numOfAssignTmp;
+      pAssign = pAssignTmp;
+
+    } else {
+//      tmq_offset_seek(tmq, "tp", pAssign[0].vgId, pAssign[0].currentOffset);
+//      tmq_offset_seek(tmq, "tp", pAssign[1].vgId, pAssign[1].currentOffset);
+//      tmq_commit_sync(tmq, pRes);
+      continue;
+    }
+
+//    tmq_commit_sync(tmq, pRes);
+    if (pRes != NULL) {
+      taos_free_result(pRes);
+      //      if ((++count) > 1) {
+      //        break;
+      //      }
+    } else {
+//      break;
+    }
+
+//    tmq_offset_seek(tmq, "tp", pAssign[0].vgId, pAssign[0].begin);
+  }
+
+  tmq_free_assignment(pAssign);
+
+  code = tmq_get_topic_assignment(tmq, "tp", &pAssign, &numOfAssign);
+  if (code != 0) {
+    printf("error occurs:%s\n", tmq_err2str(code));
+    tmq_free_assignment(pAssign);
+    tmq_consumer_close(tmq);
+    taos_close(pConn);
+    fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
+    return;
+  }
+
+  for(int i = 0; i < numOfAssign; i++){
+    printf("assign i:%d, vgId:%d, offset:%lld, start:%lld, end:%lld\n", i, pAssign[i].vgId, pAssign[i].currentOffset, pAssign[i].begin, pAssign[i].end);
+  }
+
+  tmq_consumer_close(tmq);
+  taos_close(pConn);
+  fprintf(stderr, "%d msg consumed, include %d rows\n", msgCnt, totalRows);
+}
 #pragma GCC diagnostic pop

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

@@ -117,9 +117,10 @@ static void vmProcessFetchQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
     const STraceId *trace = &pMsg->info.traceId;
     dGTrace("vgId:%d, msg:%p get from vnode-fetch queue", pVnode->vgId, pMsg);
 
+    terrno = 0;
     int32_t code = vnodeProcessFetchMsg(pVnode->pImpl, pMsg, pInfo);
     if (code != 0) {
-      if (terrno != 0) {
+      if (code == -1 && terrno != 0) {
         code = terrno;
       }
 

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

@@ -863,6 +863,7 @@ static int32_t mndProcessAlterUserReq(SRpcMsg *pReq) {
         mndReleaseDb(pMnode, pDb);
         goto _OVER;
       }
+      mndReleaseDb(pMnode, pDb);
     } else {
       while (1) {
         SDbObj *pDb = NULL;
@@ -887,6 +888,7 @@ static int32_t mndProcessAlterUserReq(SRpcMsg *pReq) {
         mndReleaseDb(pMnode, pDb);
         goto _OVER;
       }
+      mndReleaseDb(pMnode, pDb);
     } else {
       while (1) {
         SDbObj *pDb = NULL;
@@ -908,6 +910,7 @@ static int32_t mndProcessAlterUserReq(SRpcMsg *pReq) {
         goto _OVER;
       }
       taosHashRemove(newUser.readDbs, alterReq.objname, len);
+      mndReleaseDb(pMnode, pDb);
     } else {
       taosHashClear(newUser.readDbs);
     }
@@ -922,6 +925,7 @@ static int32_t mndProcessAlterUserReq(SRpcMsg *pReq) {
         goto _OVER;
       }
       taosHashRemove(newUser.writeDbs, alterReq.objname, len);
+      mndReleaseDb(pMnode, pDb);
     } else {
       taosHashClear(newUser.writeDbs);
     }

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

@@ -183,7 +183,7 @@ static bool hasStreamTaskInTimer(SStreamMeta* pMeta) {
     }
 
     SStreamTask* pTask = *(SStreamTask**)pIter;
-    if (pTask->status.timerActive == 1) {
+    if (pTask->status.timerActive >= 1) {
       inTimer = true;
     }
   }
@@ -646,7 +646,8 @@ int32_t tqProcessVgCommittedInfoReq(STQ* pTq, SRpcMsg* pMsg) {
   SDecoder decoder;
   tDecoderInit(&decoder, (uint8_t*)data, len);
   if (tDecodeMqVgOffset(&decoder, &vgOffset) < 0) {
-    return TSDB_CODE_OUT_OF_MEMORY;
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return terrno;
   }
 
   tDecoderClear(&decoder);
@@ -654,19 +655,22 @@ int32_t tqProcessVgCommittedInfoReq(STQ* pTq, SRpcMsg* pMsg) {
   STqOffset* pOffset = &vgOffset.offset;
   STqOffset* pSavedOffset = tqOffsetRead(pTq->pOffsetStore, pOffset->subKey);
   if (pSavedOffset == NULL) {
-    return TSDB_CODE_TMQ_NO_COMMITTED;
+    terrno = TSDB_CODE_TMQ_NO_COMMITTED;
+    return terrno;
   }
   vgOffset.offset = *pSavedOffset;
 
   int32_t code = 0;
   tEncodeSize(tEncodeMqVgOffset, &vgOffset, len, code);
   if (code < 0) {
-    return TSDB_CODE_INVALID_PARA;
+    terrno = TSDB_CODE_INVALID_PARA;
+    return terrno;
   }
 
   void* buf = rpcMallocCont(len);
   if (buf == NULL) {
-    return TSDB_CODE_OUT_OF_MEMORY;
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return terrno;
   }
   SEncoder encoder;
   tEncoderInit(&encoder, buf, len);

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

@@ -15,6 +15,7 @@
 
 #include "tsdb.h"
 #include "tsdbFSet2.h"
+#include "tsdbMerge.h"
 #include "tsdbReadUtil.h"
 #include "tsdbSttFileRW.h"
 
@@ -352,10 +353,14 @@ static int32_t extractSttBlockInfo(SLDataIter *pIter, const TSttBlkArray *pArray
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t uidComparFn(const void *p1, const void *p2) {
-  const uint64_t *uid1 = p1;
+static int32_t suidComparFn(const void *target, const void *p2) {
+  const uint64_t *targetUid = target;
   const uint64_t *uid2 = p2;
-  return (*uid1) - (*uid2);
+  if (*uid2 == (*targetUid)) {
+    return 0;
+  } else {
+    return (*targetUid) < (*uid2) ? -1:1;
+  }
 }
 
 static bool existsFromSttBlkStatis(const TStatisBlkArray *pStatisBlkArray, uint64_t suid, uint64_t uid,
@@ -372,29 +377,55 @@ static bool existsFromSttBlkStatis(const TStatisBlkArray *pStatisBlkArray, uint6
     }
   }
 
-  //  for (; i < TARRAY2_SIZE(pStatisBlkArray); ++i) {
-  //    SStatisBlk *p = &pStatisBlkArray->data[i];
-  //    if (p->minTbid.uid <= uid && p->maxTbid.uid >= uid) {
-  //      break;
-  //    }
-  //
-  //    if (p->maxTbid.uid < uid) {
-  //      break;
-  //    }
-  //  }
-
   if (i >= TARRAY2_SIZE(pStatisBlkArray)) {
     return false;
   }
 
-  SStatisBlk    *p = &pStatisBlkArray->data[i];
-  STbStatisBlock block = {0};
-  tsdbSttFileReadStatisBlock(pReader, p, &block);
+  while(i < TARRAY2_SIZE(pStatisBlkArray)) {
+    SStatisBlk *p = &pStatisBlkArray->data[i];
+    if (p->minTbid.suid > suid) {
+      return false;
+    }
 
-  int32_t index = tarray2SearchIdx(block.uid, &uid, sizeof(int64_t), uidComparFn, TD_EQ);
-  tStatisBlockDestroy(&block);
+    STbStatisBlock block = {0};
+    tsdbSttFileReadStatisBlock(pReader, p, &block);
 
-  return (index != -1);
+    int32_t index = tarray2SearchIdx(block.suid, &suid, sizeof(int64_t), suidComparFn, TD_EQ);
+    if (index == -1) {
+      tStatisBlockDestroy(&block);
+      return false;
+    }
+
+    int32_t j = index;
+    if (block.uid->data[j] == uid) {
+      tStatisBlockDestroy(&block);
+      return true;
+    } else if (block.uid->data[j] > uid) {
+      while (j >= 0 && block.suid->data[j] == suid) {
+        if (block.uid->data[j] == uid) {
+          tStatisBlockDestroy(&block);
+          return true;
+        } else {
+          j -= 1;
+        }
+      }
+    } else {
+      j = index + 1;
+      while (j < block.suid->size && block.suid->data[j] == suid) {
+        if (block.uid->data[j] == uid) {
+          tStatisBlockDestroy(&block);
+          return true;
+        } else {
+          j += 1;
+        }
+      }
+    }
+
+    tStatisBlockDestroy(&block);
+    i += 1;
+  }
+
+  return false;
 }
 
 int32_t tLDataIterOpen2(struct SLDataIter *pIter, SSttFileReader *pSttFileReader, int32_t iStt, int8_t backward,
@@ -452,12 +483,12 @@ int32_t tLDataIterOpen2(struct SLDataIter *pIter, SSttFileReader *pSttFileReader
     tsdbDebug("load the stt file info completed, elapsed time:%.2fms, %s", el, idStr);
   }
 
-  //  bool exists = existsFromSttBlkStatis(pBlockLoadInfo->pSttStatisBlkArray, suid, uid, pIter->pReader);
-  //  if (!exists) {
-  //    pIter->iSttBlk = -1;
-  //    pIter->pSttBlk = NULL;
-  //    return TSDB_CODE_SUCCESS;
-  //  }
+  bool exists = existsFromSttBlkStatis(pBlockLoadInfo->pSttStatisBlkArray, suid, uid, pIter->pReader);
+  if (!exists) {
+    pIter->iSttBlk = -1;
+    pIter->pSttBlk = NULL;
+    return TSDB_CODE_SUCCESS;
+  }
 
   // find the start block, actually we could load the position to avoid repeatly searching for the start position when
   // the skey is updated.

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

@@ -439,7 +439,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, void
   return code;
 
 _end:
-  tsdbReaderClose(pReader);
+  tsdbReaderClose2(pReader);
   *ppReader = NULL;
   return code;
 }
@@ -1729,41 +1729,45 @@ static int32_t mergeFileBlockAndLastBlock(STsdbReader* pReader, SLastBlockReader
 
     // row in last file block
     TSDBROW fRow = tsdbRowFromBlockData(pBlockData, pDumpInfo->rowIndex);
-    int64_t ts = getCurrentKeyInLastBlock(pLastBlockReader);
-
+    int64_t tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
     if (ASCENDING_TRAVERSE(pReader->info.order)) {
-      if (key < ts) {  // imem, mem are all empty, file blocks (data blocks and last block) exist
+      if (key < tsLast) {
         return mergeRowsInFileBlocks(pBlockData, pBlockScanInfo, key, pReader);
-      } else if (key == ts) {
-        SRow*   pTSRow = NULL;
-        int32_t code = tsdbRowMergerAdd(pMerger, &fRow, pReader->info.pSchema);
-        if (code != TSDB_CODE_SUCCESS) {
-          return code;
-        }
-
-        doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader);
-
-        TSDBROW* pRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
-        tsdbRowMergerAdd(pMerger, pRow1, NULL);
-
-        doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, ts, 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);
-        return code;
-      } else {  // key > ts
+      } else if (key > tsLast) {
+        return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, NULL, false);
+      }
+    } else {
+      if (key > tsLast) {
+        return mergeRowsInFileBlocks(pBlockData, pBlockScanInfo, key, pReader);
+      } else if (key < tsLast) {
         return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, NULL, false);
       }
-    } else {  // desc order
-      return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, pBlockData, true);
     }
+    // the following for key == tsLast
+    SRow*   pTSRow = NULL;
+    int32_t code = tsdbRowMergerAdd(pMerger, &fRow, pReader->info.pSchema);
+    if (code != TSDB_CODE_SUCCESS) {
+      return code;
+    }
+
+    doMergeRowsInFileBlocks(pBlockData, pBlockScanInfo, pReader);
+
+    TSDBROW* pRow1 = tMergeTreeGetRow(&pLastBlockReader->mergeTree);
+    tsdbRowMergerAdd(pMerger, pRow1, NULL);
+
+    doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLast, 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);
+    return code;
+
   } else {  // only last block exists
     return doMergeFileBlockAndLastBlock(pLastBlockReader, pReader, pBlockScanInfo, NULL, false);
   }
@@ -2190,7 +2194,8 @@ static int32_t buildComposedDataBlockImpl(STsdbReader* pReader, STableBlockScanI
   SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
 
   TSDBROW *pRow = NULL, *piRow = NULL;
-  int64_t key = (pBlockData->nRow > 0 && (!pDumpInfo->allDumped)) ? pBlockData->aTSKEY[pDumpInfo->rowIndex] : INT64_MIN;
+  int64_t key = (pBlockData->nRow > 0 && (!pDumpInfo->allDumped)) ? pBlockData->aTSKEY[pDumpInfo->rowIndex] :
+                                                                   (ASCENDING_TRAVERSE(pReader->info.order) ? INT64_MAX : INT64_MIN);
   if (pBlockScanInfo->iter.hasVal) {
     pRow = getValidMemRow(&pBlockScanInfo->iter, pBlockScanInfo->delSkyline, pReader);
   }
@@ -2564,9 +2569,8 @@ static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) {
 
     // load the last data block of current table
     STableBlockScanInfo* pScanInfo = *(STableBlockScanInfo**)pStatus->pTableIter;
-    if (pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &pScanInfo->uid, sizeof(pScanInfo->uid))) {
-      // reset the index in last block when handing a new file
-      //      doCleanupTableScanInfo(pScanInfo);
+    if (pScanInfo == NULL) {
+      tsdbError("table Iter is null, invalid pScanInfo, try next table %s", pReader->idStr);
       bool hasNexTable = moveToNextTable(pUidList, pStatus);
       if (!hasNexTable) {
         return TSDB_CODE_SUCCESS;
@@ -2575,8 +2579,15 @@ static int32_t doLoadLastBlockSequentially(STsdbReader* pReader) {
       continue;
     }
 
-    // reset the index in last block when handing a new file
-    //    doCleanupTableScanInfo(pScanInfo);
+    if (pReader->pIgnoreTables && taosHashGet(*pReader->pIgnoreTables, &pScanInfo->uid, sizeof(pScanInfo->uid))) {
+      // reset the index in last block when handing a new file
+      bool hasNexTable = moveToNextTable(pUidList, pStatus);
+      if (!hasNexTable) {
+        return TSDB_CODE_SUCCESS;
+      }
+
+      continue;
+    }
 
     bool hasDataInLastFile = initLastBlockReader(pLastBlockReader, pScanInfo, pReader);
     if (!hasDataInLastFile) {
@@ -2667,16 +2678,32 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
         (ASCENDING_TRAVERSE(pReader->info.order)) ? pBlockInfo->record.firstKey : pBlockInfo->record.lastKey;
     code = buildDataBlockFromBuf(pReader, pScanInfo, endKey);
   } else {
-    if (hasDataInLastBlock(pLastBlockReader) && !ASCENDING_TRAVERSE(pReader->info.order)) {
-      // only return the rows in last block
-      int64_t tsLast = getCurrentKeyInLastBlock(pLastBlockReader);
-      ASSERT(tsLast >= pBlockInfo->record.lastKey);
+    bool bHasDataInLastBlock = hasDataInLastBlock(pLastBlockReader);
+    int64_t tsLast = bHasDataInLastBlock ? getCurrentKeyInLastBlock(pLastBlockReader) : INT64_MIN;
+    if (!bHasDataInLastBlock || ((ASCENDING_TRAVERSE(pReader->info.order) && pBlockInfo->record.lastKey < tsLast) ||
+                                 (!ASCENDING_TRAVERSE(pReader->info.order) && pBlockInfo->record.firstKey > tsLast))) {
+      // whole block is required, return it directly
+      SDataBlockInfo* pInfo = &pReader->resBlockInfo.pResBlock->info;
+      pInfo->rows = pBlockInfo->record.numRow;
+      pInfo->id.uid = pScanInfo->uid;
+      pInfo->dataLoad = 0;
+      pInfo->window = (STimeWindow){.skey = pBlockInfo->record.firstKey, .ekey = pBlockInfo->record.lastKey};
+      setComposedBlockFlag(pReader, false);
+      setBlockAllDumped(&pStatus->fBlockDumpInfo, pBlockInfo->record.lastKey, pReader->info.order);
 
+      // update the last key for the corresponding table
+      pScanInfo->lastKey = ASCENDING_TRAVERSE(pReader->info.order) ? pInfo->window.ekey : pInfo->window.skey;
+      tsdbDebug("%p uid:%" PRIu64
+                " clean file block retrieved from file, global index:%d, "
+                "table index:%d, rows:%d, brange:%" PRId64 "-%" PRId64 ", %s",
+                pReader, pScanInfo->uid, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlockInfo->record.numRow,
+                pBlockInfo->record.firstKey, pBlockInfo->record.lastKey, pReader->idStr);
+    } else {
       SBlockData* pBData = &pReader->status.fileBlockData;
       tBlockDataReset(pBData);
 
       SSDataBlock* pResBlock = pReader->resBlockInfo.pResBlock;
-      tsdbDebug("load data in last block firstly, due to desc scan data, %s", pReader->idStr);
+      tsdbDebug("load data in last block firstly %s", pReader->idStr);
 
       int64_t st = taosGetTimestampUs();
 
@@ -2707,23 +2734,8 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
                   pReader, pResBlock->info.id.uid, pResBlock->info.window.skey, pResBlock->info.window.ekey,
                   pResBlock->info.rows, el, pReader->idStr);
       }
-    } else {  // whole block is required, return it directly
-      SDataBlockInfo* pInfo = &pReader->resBlockInfo.pResBlock->info;
-      pInfo->rows = pBlockInfo->record.numRow;
-      pInfo->id.uid = pScanInfo->uid;
-      pInfo->dataLoad = 0;
-      pInfo->window = (STimeWindow){.skey = pBlockInfo->record.firstKey, .ekey = pBlockInfo->record.lastKey};
-      setComposedBlockFlag(pReader, false);
-      setBlockAllDumped(&pStatus->fBlockDumpInfo, pBlockInfo->record.lastKey, pReader->info.order);
-
-      // update the last key for the corresponding table
-      pScanInfo->lastKey = ASCENDING_TRAVERSE(pReader->info.order) ? pInfo->window.ekey : pInfo->window.skey;
-      tsdbDebug("%p uid:%" PRIu64
-                " clean file block retrieved from file, global index:%d, "
-                "table index:%d, rows:%d, brange:%" PRId64 "-%" PRId64 ", %s",
-                pReader, pScanInfo->uid, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlockInfo->record.numRow,
-                pBlockInfo->record.firstKey, pBlockInfo->record.lastKey, pReader->idStr);
     }
+
   }
 
   return (pReader->code != TSDB_CODE_SUCCESS) ? pReader->code : code;
@@ -4096,12 +4108,10 @@ int32_t tsdbReaderSuspend2(STsdbReader* pReader) {
     }
 
     tsdbDataFileReaderClose(&pReader->pFileReader);
-
     int64_t loadBlocks = 0;
     double  elapse = 0;
     pReader->status.pLDataIterArray = destroySttBlockReader(pReader->status.pLDataIterArray, &loadBlocks, &elapse);
     pReader->status.pLDataIterArray = taosArrayInit(4, POINTER_BYTES);
-
     // resetDataBlockScanInfo excluding lastKey
     STableBlockScanInfo** p = NULL;
     int32_t               iter = 0;

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

@@ -635,7 +635,6 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
       return tqProcessVgCommittedInfoReq(pVnode->pTq, pMsg);
     case TDMT_VND_TMQ_SEEK:
       return tqProcessSeekReq(pVnode->pTq, pMsg);
-
     default:
       vError("unknown msg type:%d in fetch queue", pMsg->msgType);
       return TSDB_CODE_APP_ERROR;

source/libs/executor/inc/tsort.h

@@ -191,6 +191,8 @@ int32_t getProperSortPageSize(size_t rowSize, uint32_t numOfCols);
 bool tsortIsClosed(SSortHandle* pHandle);
 void tsortSetClosed(SSortHandle* pHandle);
 
+void setSingleTableMerge(SSortHandle* pHandle);
+
 #ifdef __cplusplus
 }
 #endif

source/libs/executor/src/scanoperator.c

@@ -2938,17 +2938,22 @@ int32_t startGroupTableMergeScan(SOperatorInfo* pOperator) {
   // one table has one data block
   int32_t numOfTable = tableEndIdx - tableStartIdx + 1;
 
-  STableMergeScanSortSourceParam param = {0};
-  param.pOperator = pOperator;
+  STableMergeScanSortSourceParam *param = taosMemoryCalloc(1, sizeof(STableMergeScanSortSourceParam));
+  param->pOperator = pOperator;
   STableKeyInfo* startKeyInfo = tableListGetInfo(pInfo->base.pTableListInfo, tableStartIdx);
   pAPI->tsdReader.tsdReaderOpen(pHandle->vnode, &pInfo->base.cond, startKeyInfo, numOfTable, pInfo->pReaderBlock, (void**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo), false, NULL);
 
   SSortSource* ps = taosMemoryCalloc(1, sizeof(SSortSource));
-  ps->param = &param;
-  ps->onlyRef = true;
+  ps->param = param;
+  ps->onlyRef = false;
   tsortAddSource(pInfo->pSortHandle, ps);
 
-  int32_t code = tsortOpen(pInfo->pSortHandle);
+  int32_t code = TSDB_CODE_SUCCESS;
+  if (numOfTable == 1) {
+    setSingleTableMerge(pInfo->pSortHandle);
+  } else {
+    code = tsortOpen(pInfo->pSortHandle);
+  }
 
   if (code != TSDB_CODE_SUCCESS) {
     T_LONG_JMP(pTaskInfo->env, terrno);
@@ -3587,4 +3592,4 @@ static void destoryTableCountScanOperator(void* param) {
 
   taosArrayDestroy(pTableCountScanInfo->stbUidList);
   taosMemoryFreeClear(param);
-}
+}

source/libs/executor/src/tsort.c

@@ -69,8 +69,14 @@ struct SSortHandle {
   _sort_fetch_block_fn_t  fetchfp;
   _sort_merge_compar_fn_t comparFn;
   SMultiwayMergeTreeInfo* pMergeTree;
+
+  bool singleTableMerge;
 };
 
+void setSingleTableMerge(SSortHandle* pHandle) {
+  pHandle->singleTableMerge = true;
+}
+
 static int32_t msortComparFn(const void* pLeft, const void* pRight, void* param);
 
 // | offset[0] | offset[1] |....| nullbitmap | data |...|
@@ -1453,6 +1459,26 @@ static STupleHandle* tsortPQSortNextTuple(SSortHandle* pHandle) {
   return &pHandle->tupleHandle;
 }
 
+static STupleHandle* tsortSingleTableMergeNextTuple(SSortHandle* pHandle) {
+  if (1 == pHandle->numOfCompletedSources) return NULL;
+  if (pHandle->tupleHandle.pBlock && pHandle->tupleHandle.rowIndex + 1 < pHandle->tupleHandle.pBlock->info.rows) {
+    pHandle->tupleHandle.rowIndex++;
+  } else {
+    if (pHandle->tupleHandle.rowIndex == -1) return NULL;
+    SSortSource** pSource = taosArrayGet(pHandle->pOrderedSource, 0);
+    SSortSource*  source = *pSource;
+    SSDataBlock*  pBlock = pHandle->fetchfp(source->param);
+    if (!pBlock || pBlock->info.rows == 0) {
+      setCurrentSourceDone(source, pHandle);
+      pHandle->tupleHandle.pBlock = NULL;
+      return NULL;
+    }
+    pHandle->tupleHandle.pBlock = pBlock;
+    pHandle->tupleHandle.rowIndex = 0;
+  }
+  return &pHandle->tupleHandle;
+}
+
 int32_t tsortOpen(SSortHandle* pHandle) {
   if (pHandle->opened) {
     return 0;
@@ -1470,7 +1496,9 @@ int32_t tsortOpen(SSortHandle* pHandle) {
 }
 
 STupleHandle* tsortNextTuple(SSortHandle* pHandle) {
-  if (pHandle->pBoundedQueue)
+  if (pHandle->singleTableMerge)
+    return tsortSingleTableMergeNextTuple(pHandle);
+  else if (pHandle->pBoundedQueue)
     return tsortPQSortNextTuple(pHandle);
   else
     return tsortBufMergeSortNextTuple(pHandle);

source/libs/stream/src/streamDispatch.c

@@ -550,13 +550,27 @@ int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pDat
 
 static void doRetryDispatchData(void* param, void* tmrId) {
   SStreamTask* pTask = param;
+
+  if (streamTaskShouldStop(&pTask->status)) {
+    atomic_sub_fetch_8(&pTask->status.timerActive, 1);
+    qDebug("s-task:%s should stop, abort from timer", pTask->id.idStr);
+    return;
+  }
+
   ASSERT(pTask->outputInfo.status == TASK_OUTPUT_STATUS__WAIT);
 
   int32_t code = streamDispatchAllBlocks(pTask, pTask->msgInfo.pData);
   if (code != TSDB_CODE_SUCCESS) {
-    qDebug("s-task:%s reset the waitRspCnt to be 0 before launch retry dispatch", pTask->id.idStr);
-    atomic_store_32(&pTask->shuffleDispatcher.waitingRspCnt, 0);
-    streamRetryDispatchStreamBlock(pTask, DISPATCH_RETRY_INTERVAL_MS);
+    if (!streamTaskShouldStop(&pTask->status)) {
+      qDebug("s-task:%s reset the waitRspCnt to be 0 before launch retry dispatch", pTask->id.idStr);
+      atomic_store_32(&pTask->shuffleDispatcher.waitingRspCnt, 0);
+      streamRetryDispatchStreamBlock(pTask, DISPATCH_RETRY_INTERVAL_MS);
+    } else {
+      atomic_sub_fetch_8(&pTask->status.timerActive, 1);
+      qDebug("s-task:%s should stop, abort from timer", pTask->id.idStr);
+    }
+  } else {
+    atomic_sub_fetch_8(&pTask->status.timerActive, 1);
   }
 }
 

source/libs/stream/src/streamMeta.c

@@ -137,19 +137,7 @@ void streamMetaClose(SStreamMeta* pMeta) {
     if (pIter == NULL) {
       break;
     }
-
-    SStreamTask* pTask = *(SStreamTask**)pIter;
-    if (pTask->schedTimer) {
-      taosTmrStop(pTask->schedTimer);
-      pTask->schedTimer = NULL;
-    }
-
-    if (pTask->launchTaskTimer) {
-      taosTmrStop(pTask->launchTaskTimer);
-      pTask->launchTaskTimer = NULL;
-    }
-
-    tFreeStreamTask(pTask);
+    tFreeStreamTask(*(SStreamTask**)pIter);
   }
 
   taosHashCleanup(pMeta->pTasks);
@@ -362,11 +350,6 @@ int32_t streamMetaUnregisterTask(SStreamMeta* pMeta, int32_t taskId) {
     int32_t num = taosArrayGetSize(pMeta->pTaskList);
     doRemoveIdFromList(pMeta, num, pTask->id.taskId);
 
-    // remove the ref by timer
-    if (pTask->triggerParam != 0) {
-      taosTmrStop(pTask->schedTimer);
-    }
-
     streamMetaRemoveTask(pMeta, taskId);
     streamMetaReleaseTask(pMeta, pTask);
   } else {

source/libs/stream/src/streamRecover.c

@@ -540,14 +540,14 @@ static void tryLaunchHistoryTask(void* param, void* tmrId) {
   taosWLockLatch(&pMeta->lock);
   SStreamTask** ppTask = (SStreamTask**)taosHashGet(pMeta->pTasks, &pInfo->taskId, sizeof(int32_t));
   if (ppTask) {
-    ASSERT((*ppTask)->status.timerActive == 1);
+    ASSERT((*ppTask)->status.timerActive >= 1);
 
     if (streamTaskShouldStop(&(*ppTask)->status)) {
       const char* pStatus = streamGetTaskStatusStr((*ppTask)->status.taskStatus);
       qDebug("s-task:%s status:%s quit timer task", (*ppTask)->id.idStr, pStatus);
 
       taosMemoryFree(pInfo);
-      (*ppTask)->status.timerActive = 0;
+      atomic_sub_fetch_8(&(*ppTask)->status.timerActive, 1);
       taosWUnLockLatch(&pMeta->lock);
       return;
     }
@@ -556,7 +556,7 @@ static void tryLaunchHistoryTask(void* param, void* tmrId) {
 
   SStreamTask* pTask = streamMetaAcquireTask(pMeta, pInfo->taskId);
   if (pTask != NULL) {
-    ASSERT(pTask->status.timerActive == 1);
+    ASSERT(pTask->status.timerActive >= 1);
 
     // abort the timer if intend to stop task
     SStreamTask* pHTask = streamMetaAcquireTask(pMeta, pTask->historyTaskId.taskId);
@@ -578,7 +578,7 @@ static void tryLaunchHistoryTask(void* param, void* tmrId) {
     }
 
     // not in timer anymore
-    pTask->status.timerActive = 0;
+    atomic_sub_fetch_8(&pTask->status.timerActive, 1);
     streamMetaReleaseTask(pMeta, pTask);
   } else {
     qError("s-task:0x%x failed to load task, it may have been destroyed", pInfo->taskId);
@@ -609,11 +609,11 @@ int32_t streamLaunchFillHistoryTask(SStreamTask* pTask) {
         // todo failed to create timer
         taosMemoryFree(pInfo);
       } else {
-        pTask->status.timerActive = 1;  // timer is active
+        atomic_add_fetch_8(&pTask->status.timerActive, 1);// timer is active
         qDebug("s-task:%s set timer active flag", pTask->id.idStr);
       }
     } else {  // timer exists
-      pTask->status.timerActive = 1;
+      ASSERT(pTask->status.timerActive > 0);
       qDebug("s-task:%s set timer active flag, task timer not null", pTask->id.idStr);
       taosTmrReset(tryLaunchHistoryTask, 100, pInfo, streamEnv.timer, &pTask->launchTaskTimer);
     }

source/libs/stream/src/streamTask.c

@@ -13,11 +13,11 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include <libs/transport/trpc.h>
-#include <streamInt.h>
+#include "streamInt.h"
 #include "executor.h"
 #include "tstream.h"
 #include "wal.h"
+#include "ttimer.h"
 
 static int32_t addToTaskset(SArray* pArray, SStreamTask* pTask) {
   int32_t childId = taosArrayGetSize(pArray);
@@ -213,6 +213,22 @@ static void freeItem(void* p) {
 void tFreeStreamTask(SStreamTask* pTask) {
   qDebug("free s-task:%s, %p", pTask->id.idStr, pTask);
 
+  // remove the ref by timer
+  while(pTask->status.timerActive > 0) {
+    qDebug("s-task:%s wait for task stop timer activities", pTask->id.idStr);
+    taosMsleep(10);
+  }
+
+  if (pTask->schedTimer != NULL) {
+    taosTmrStop(pTask->schedTimer);
+    pTask->schedTimer = NULL;
+  }
+
+  if (pTask->launchTaskTimer != NULL) {
+    taosTmrStop(pTask->launchTaskTimer);
+    pTask->launchTaskTimer = NULL;
+  }
+
   int32_t status = atomic_load_8((int8_t*)&(pTask->status.taskStatus));
   if (pTask->inputQueue) {
     streamQueueClose(pTask->inputQueue);