taosdata · hjxilinx · Sep 13, 2023 · Sep 1, 2023 · Sep 4, 2023 · Sep 4, 2023
diff --git a/Jenkinsfile2 b/Jenkinsfile2
@@ -315,7 +315,9 @@ def pre_test_build_win() {
         python.exe -m pip install --upgrade pip
         python -m pip uninstall taospy -y 
         python -m pip install taospy==2.7.10
-        xcopy /e/y/i/f %WIN_INTERNAL_ROOT%\\debug\\build\\lib\\taos.dll C:\\Windows\\System32
+        python -m pip uninstall taos-ws-py -y 
+        python -m pip install taos-ws-py==0.2.8
+	xcopy /e/y/i/f %WIN_INTERNAL_ROOT%\\debug\\build\\lib\\taos.dll C:\\Windows\\System32
     '''
     return 1
 }

diff --git a/cmake/cmake.define b/cmake/cmake.define
@@ -1,5 +1,5 @@
 cmake_minimum_required(VERSION 3.0)
-set(CMAKE_VERBOSE_MAKEFILE ON)
+set(CMAKE_VERBOSE_MAKEFILE FALSE)
 set(TD_BUILD_TAOSA_INTERNAL FALSE)
 
 #set output directory

diff --git a/cmake/cmake.version b/cmake/cmake.version
@@ -2,7 +2,7 @@
 IF (DEFINED VERNUMBER)
   SET(TD_VER_NUMBER ${VERNUMBER})
 ELSE ()
-  SET(TD_VER_NUMBER "3.2.0.0.alpha")
+  SET(TD_VER_NUMBER "3.1.2.0.alpha")
 ENDIF ()
 
 IF (DEFINED VERCOMPATIBLE)

diff --git a/docs/en/12-taos-sql/06-select.md b/docs/en/12-taos-sql/06-select.md
@@ -9,7 +9,7 @@ description: This document describes how to query data in TDengine.
 ```sql
 SELECT {DATABASE() | CLIENT_VERSION() | SERVER_VERSION() | SERVER_STATUS() | NOW() | TODAY() | TIMEZONE() | CURRENT_USER() | USER() }
 
-SELECT [hints] [DISTINCT] [TAGS] select_list
+SELECT [hints] [DISTINCT] select_list
     from_clause
     [WHERE condition]
     [partition_by_clause]
@@ -225,14 +225,6 @@ The \_IROWTS pseudocolumn can only be used with INTERP function. This pseudocolu
 select _irowts, interp(current) from meters range('2020-01-01 10:00:00', '2020-01-01 10:30:00') every(1s) fill(linear);
 ```
 
-### TAGS Query
-
-The TAGS keyword returns only tag columns from all child tables when only tag columns are specified. One row containing tag columns is returned for each child table.
-
-```sql
-SELECT TAGS tag_name [, tag_name ...] FROM stb_name
-```
-
 ## Query Objects
 
 `FROM` can be followed by a number of tables or super tables, or can be followed by a sub-query.

diff --git a/docs/en/12-taos-sql/10-function.md b/docs/en/12-taos-sql/10-function.md
@@ -292,11 +292,11 @@ CONCAT_WS(separator_expr, expr1, expr2 [, expr] ...)
 LENGTH(expr)
 ```
 
-**Description**: The length in bytes of a string
+**Description**: The length in bytes
 
 **Return value type**: Bigint
 
-**Applicable data types**: VARCHAR and NCHAR fields or columns
+**Applicable data types**: VARCHAR and NCHAR and VARBINARY 
 
 **Nested query**: It can be used in both the outer query and inner query in a nested query.
 

diff --git a/docs/zh/02-intro.md b/docs/zh/02-intro.md
@@ -10,7 +10,7 @@ TDengine 是一款开源、高性能、云原生的[时序数据库](https://tde
 
 ## 主要产品
 
-TDengine 有三个主要产品：TDengine Pro （即 TDengine 企业版），TDengine Cloud，和 TDengine OSS，关于它们的具体定义请参考 
+TDengine 有三个主要产品：TDengine Enterprise （即 TDengine 企业版），TDengine Cloud，和 TDengine OSS，关于它们的具体定义请参考 
 - [TDengine 企业版](https://www.taosdata.com/tdengine-pro)
 - [TDengine 云服务](https://cloud.taosdata.com/?utm_source=menu&utm_medium=webcn)
 - [TDengine 开源版](https://www.taosdata.com/tdengine-oss)

diff --git a/docs/zh/12-taos-sql/06-select.md b/docs/zh/12-taos-sql/06-select.md
@@ -9,7 +9,7 @@ description: 查询数据的详细语法
 ```sql
 SELECT {DATABASE() | CLIENT_VERSION() | SERVER_VERSION() | SERVER_STATUS() | NOW() | TODAY() | TIMEZONE() | CURRENT_USER() | USER() }
 
-SELECT [hints] [DISTINCT] [TAGS] select_list
+SELECT [hints] [DISTINCT] select_list
     from_clause
     [WHERE condition]
     [partition_by_clause]
@@ -160,16 +160,6 @@ SELECT DISTINCT col_name [, col_name ...] FROM tb_name;
 
 :::
 
-### 标签查询
-
-当查询的列只有标签列时，`TAGS` 关键字可以指定返回所有子表的标签列。每个子表只返回一行标签列。
-
-返回所有子表的标签列：
-
-```sql
-SELECT TAGS tag_name [, tag_name ...] FROM stb_name
-``` 
-
 ### 结果集列名
 
 `SELECT`子句中，如果不指定返回结果集合的列名，结果集列名称默认使用`SELECT`子句中的表达式名称作为列名称。此外，用户可使用`AS`来重命名返回结果集合中列的名称。例如：

diff --git a/docs/zh/12-taos-sql/10-function.md b/docs/zh/12-taos-sql/10-function.md
@@ -292,11 +292,11 @@ CONCAT_WS(separator_expr, expr1, expr2 [, expr] ...)
 LENGTH(expr)
 ```
 
-**功能说明**：以字节计数的字符串长度。
+**功能说明**：以字节计数的长度。
 
 **返回结果类型**：BIGINT。
 
-**适用数据类型**：输入参数是 VARCHAR 类型或者 NCHAR 类型的字符串或者列。
+**适用数据类型**：VARCHAR, NCHAR, VARBINARY。
 
 **嵌套子查询支持**：适用于内层查询和外层查询。
 

diff --git a/include/common/ttime.h b/include/common/ttime.h
@@ -75,7 +75,7 @@ static FORCE_INLINE int64_t taosGetTimestampToday(int32_t precision) {
 int64_t taosTimeAdd(int64_t t, int64_t duration, char unit, int32_t precision);
 
 int64_t taosTimeTruncate(int64_t ts, const SInterval* pInterval);
-int32_t taosTimeCountInterval(int64_t skey, int64_t ekey, int64_t interval, char unit, int32_t precision);
+int32_t taosTimeCountIntervalForFill(int64_t skey, int64_t ekey, int64_t interval, char unit, int32_t precision, int32_t order);
 
 int32_t parseAbsoluteDuration(const char* token, int32_t tokenlen, int64_t* ts, char* unit, int32_t timePrecision);
 int32_t parseNatualDuration(const char* token, int32_t tokenLen, int64_t* duration, char* unit, int32_t timePrecision);

diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h
@@ -111,14 +111,12 @@ typedef int32_t FTaskExpand(void* ahandle, SStreamTask* pTask, int64_t ver);
 typedef struct {
   int8_t      type;
   int64_t     ver;
-  int32_t*    dataRef;
   SPackedData submit;
 } SStreamDataSubmit;
 
 typedef struct {
   int8_t  type;
   int64_t ver;
-  SArray* dataRefs;  // SArray<int32_t*>
   SArray* submits;   // SArray<SPackedSubmit>
 } SStreamMergedSubmit;
 
@@ -260,7 +258,7 @@ typedef struct SStreamTaskId {
 typedef struct SCheckpointInfo {
   int64_t checkpointId;
   int64_t checkpointVer;  // latest checkpointId version
-  int64_t currentVer;     // current offset in WAL, not serialize it
+  int64_t nextProcessVer;     // current offset in WAL, not serialize it
 } SCheckpointInfo;
 
 typedef struct SStreamStatus {
@@ -312,12 +310,27 @@ typedef struct STaskSchedInfo {
   void*   pTimer;
 } STaskSchedInfo;
 
+typedef struct SSinkTaskRecorder {
+  int64_t numOfSubmit;
+  int64_t numOfBlocks;
+  int64_t numOfRows;
+} SSinkTaskRecorder;
+
 typedef struct {
+  int64_t created;
   int64_t init;
   int64_t step1Start;
   int64_t step2Start;
+  int64_t sinkStart;
 } STaskTimestamp;
 
+typedef struct STokenBucket {
+    int32_t capacity;     // total capacity
+    int64_t fillTimestamp;// fill timestamp
+    int32_t numOfToken;   // total available tokens
+    int32_t rate;         // number of token per second
+} STokenBucket;
+
 struct SStreamTask {
   int64_t          ver;
   SStreamTaskId    id;
@@ -345,6 +358,8 @@ struct SStreamTask {
     STaskSinkSma           smaSink;
     STaskSinkFetch         fetchSink;
   };
+  SSinkTaskRecorder sinkRecorder;
+  STokenBucket      tokenBucket;
 
   void*         launchTaskTimer;
   SMsgCb*       pMsgCb;  // msg handle
@@ -419,8 +434,9 @@ int32_t tDecodeStreamTaskChkInfo(SDecoder* pDecoder, SCheckpointInfo* pChkpInfo)
 int32_t tDecodeStreamTaskId(SDecoder* pDecoder, SStreamTaskId* pTaskId);
 
 int32_t streamTaskPutDataIntoInputQ(SStreamTask* pTask, SStreamQueueItem* pItem);
+int32_t streamTaskPutDataIntoOutputQ(SStreamTask* pTask, SStreamDataBlock* pBlock);
 int32_t streamTaskPutTranstateIntoInputQ(SStreamTask* pTask);
-bool    streamQueueIsFull(const STaosQueue* pQueue);
+bool    streamQueueIsFull(const STaosQueue* pQueue, bool inputQ);
 
 typedef struct {
   SMsgHead head;
@@ -630,12 +646,10 @@ SStreamChildEpInfo* streamTaskGetUpstreamTaskEpInfo(SStreamTask* pTask, int32_t
 void    streamTaskInputFail(SStreamTask* pTask);
 int32_t streamTryExec(SStreamTask* pTask);
 int32_t streamSchedExec(SStreamTask* pTask);
-int32_t streamTaskOutputResultBlock(SStreamTask* pTask, SStreamDataBlock* pBlock);
 bool    streamTaskShouldStop(const SStreamStatus* pStatus);
 bool    streamTaskShouldPause(const SStreamStatus* pStatus);
 bool    streamTaskIsIdle(const SStreamTask* pTask);
 
-int32_t streamScanExec(SStreamTask* pTask, int32_t batchSize);
 void    initRpcMsg(SRpcMsg* pMsg, int32_t msgType, void* pCont, int32_t contLen);
 
 char* createStreamTaskIdStr(int64_t streamId, int32_t taskId);
@@ -655,7 +669,7 @@ int32_t streamLaunchFillHistoryTask(SStreamTask* pTask);
 int32_t streamTaskScanHistoryDataComplete(SStreamTask* pTask);
 int32_t streamStartScanHistoryAsync(SStreamTask* pTask, int8_t igUntreated);
 bool    streamHistoryTaskSetVerRangeStep2(SStreamTask* pTask, int64_t latestVer);
-int32_t streamTaskGetInputQItems(const SStreamTask* pTask);
+int32_t streamQueueGetNumOfItems(const SStreamQueue* pQueue);
 
 // common
 int32_t     streamRestoreParam(SStreamTask* pTask);
@@ -679,11 +693,10 @@ void        streamTaskOpenAllUpstreamInput(SStreamTask* pTask);
 // source level
 int32_t streamSetParamForStreamScannerStep1(SStreamTask* pTask, SVersionRange* pVerRange, STimeWindow* pWindow);
 int32_t streamSetParamForStreamScannerStep2(SStreamTask* pTask, SVersionRange* pVerRange, STimeWindow* pWindow);
-int32_t streamSourceScanHistoryData(SStreamTask* pTask);
+int32_t streamScanHistoryData(SStreamTask* pTask);
 int32_t streamDispatchScanHistoryFinishMsg(SStreamTask* pTask);
 
 // agg level
-int32_t streamTaskScanHistoryPrepare(SStreamTask* pTask);
 int32_t streamProcessScanHistoryFinishReq(SStreamTask* pTask, SStreamScanHistoryFinishReq* pReq,
                                           SRpcHandleInfo* pRpcInfo);
 int32_t streamProcessScanHistoryFinishRsp(SStreamTask* pTask);

diff --git a/include/libs/wal/wal.h b/include/libs/wal/wal.h
@@ -192,7 +192,7 @@ int32_t walApplyVer(SWal *, int64_t ver);
 // int32_t  walDataCorrupted(SWal*);
 
 // wal reader
-SWalReader *walOpenReader(SWal *, SWalFilterCond *pCond);
+SWalReader *walOpenReader(SWal *, SWalFilterCond *pCond, int64_t id);
 void        walCloseReader(SWalReader *pRead);
 void        walReadReset(SWalReader *pReader);
 int32_t     walReadVer(SWalReader *pRead, int64_t ver);

diff --git a/include/util/taoserror.h b/include/util/taoserror.h
@@ -789,7 +789,7 @@ int32_t* taosGetErrno();
 
 // stream
 #define TSDB_CODE_STREAM_TASK_NOT_EXIST          TAOS_DEF_ERROR_CODE(0, 0x4100)
-#define TSDB_CODE_STREAM_BACKPRESSURE_OUT_OF_QUEUE          TAOS_DEF_ERROR_CODE(0, 0x4101)
+#define TSDB_CODE_STREAM_EXEC_CANCELLED          TAOS_DEF_ERROR_CODE(0, 0x4102)
 
 // TDLite
 #define TSDB_CODE_TDLITE_IVLD_OPEN_FLAGS         TAOS_DEF_ERROR_CODE(0, 0x5100)

diff --git a/packaging/tools/makepkg.sh b/packaging/tools/makepkg.sh
@@ -338,7 +338,7 @@ if [ "$verMode" == "cluster" ]; then
         tmp_pwd=`pwd`
     	  cd ${install_dir}/connector
     	  if [ ! -d taos-connector-jdbc ];then
-          	git clone -b 3.2.1 --depth=1 https://github.com/taosdata/taos-connector-jdbc.git ||:
+          	git clone -b main --depth=1 https://github.com/taosdata/taos-connector-jdbc.git ||:
     	  fi
     	  cd taos-connector-jdbc
     	  mvn clean package -Dmaven.test.skip=true

diff --git a/source/client/src/clientEnv.c b/source/client/src/clientEnv.c
@@ -380,8 +380,7 @@ void destroySubRequests(SRequestObj *pRequest) {
       pReqList[++reqIdx] = pTmp;
       releaseRequest(tmpRefId);
     } else {
-      tscError("0x%" PRIx64 ", prev req ref 0x%" PRIx64 " is not there, reqId:0x%" PRIx64, pTmp->self, tmpRefId,
-               pTmp->requestId);
+      tscError("prev req ref 0x%" PRIx64 " is not there", tmpRefId);
       break;
     }
   }
@@ -398,7 +397,7 @@ void destroySubRequests(SRequestObj *pRequest) {
       removeRequest(pTmp->self);
       releaseRequest(pTmp->self);
     } else {
-      tscError("0x%" PRIx64 " is not there", tmpRefId);
+      tscError("next req ref 0x%" PRIx64 " is not there", tmpRefId);
       break;
     }
   }
@@ -492,8 +491,7 @@ void stopAllQueries(SRequestObj *pRequest) {
       pReqList[++reqIdx] = pTmp;
       releaseRequest(tmpRefId);
     } else {
-      tscError("0x%" PRIx64 ", prev req ref 0x%" PRIx64 " is not there, reqId:0x%" PRIx64, pTmp->self, tmpRefId,
-               pTmp->requestId);
+      tscError("prev req ref 0x%" PRIx64 " is not there", tmpRefId);
       break;
     }
   }
@@ -512,7 +510,7 @@ void stopAllQueries(SRequestObj *pRequest) {
       taosStopQueryImpl(pTmp);
       releaseRequest(pTmp->self);
     } else {
-      tscError("0x%" PRIx64 " is not there", tmpRefId);
+      tscError("next req ref 0x%" PRIx64 " is not there", tmpRefId);
       break;
     }
   }

diff --git a/source/client/src/clientMain.c b/source/client/src/clientMain.c
@@ -874,8 +874,13 @@ void handleSubQueryFromAnalyse(SSqlCallbackWrapper *pWrapper, SMetaData *pResult
   if (TSDB_CODE_SUCCESS == code) {
     code = cloneCatalogReq(&pNewWrapper->pCatalogReq, pWrapper->pCatalogReq);
   }
-  doAsyncQueryFromAnalyse(pResultMeta, pNewWrapper, code);
-  nodesDestroyNode(pRoot);
+  if (TSDB_CODE_SUCCESS == code) {
+    doAsyncQueryFromAnalyse(pResultMeta, pNewWrapper, code);
+    nodesDestroyNode(pRoot);
+  } else {
+    handleQueryAnslyseRes(pWrapper, pResultMeta, code);
+    return;
+  }
 }
 
 void handleQueryAnslyseRes(SSqlCallbackWrapper *pWrapper, SMetaData *pResultMeta, int32_t code) {
@@ -1148,8 +1153,7 @@ void restartAsyncQuery(SRequestObj *pRequest, int32_t code) {
       pReqList[++reqIdx] = pTmp;
       releaseRequest(tmpRefId);
     } else {
-      tscError("0x%" PRIx64 ", prev req ref 0x%" PRIx64 " is not there, reqId:0x%" PRIx64, pTmp->self, tmpRefId,
-               pTmp->requestId);
+      tscError("prev req ref 0x%" PRIx64 " is not there", tmpRefId);
       break;
     }
   }
@@ -1162,7 +1166,7 @@ void restartAsyncQuery(SRequestObj *pRequest, int32_t code) {
       removeRequest(pTmp->self);
       releaseRequest(pTmp->self);
     } else {
-      tscError("0x%" PRIx64 " is not there", tmpRefId);
+      tscError("next req ref 0x%" PRIx64 " is not there", tmpRefId);
       break;
     }
   }

diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c
@@ -2139,22 +2139,16 @@ int32_t buildCtbNameByGroupIdImpl(const char* stbFullName, uint64_t groupId, cha
     return TSDB_CODE_FAILED;
   }
 
-  SSmlKv pTag = {.key = "group_id",
-                 .keyLen = sizeof("group_id") - 1,
-                 .type = TSDB_DATA_TYPE_UBIGINT,
-                 .u = groupId,
-                 .length = sizeof(uint64_t)};
+  int8_t      type = TSDB_DATA_TYPE_UBIGINT;
+  const char* name = "group_id";
+  int32_t     len = strlen(name);
+  SSmlKv pTag = { .key = name, .keyLen = len, .type = type, .u = groupId, .length = sizeof(uint64_t)};
   taosArrayPush(tags, &pTag);
 
   RandTableName rname = {
-      .tags = tags,
-      .stbFullName = stbFullName,
-      .stbFullNameLen = strlen(stbFullName),
-      .ctbShortName = cname,
-  };
+      .tags = tags, .stbFullName = stbFullName, .stbFullNameLen = strlen(stbFullName), .ctbShortName = cname};
 
   buildChildTableName(&rname);
-
   taosArrayDestroy(tags);
 
   if ((rname.ctbShortName && rname.ctbShortName[0]) == 0) {