merge:from 3.0

cmake/cmake.define

@@ -46,7 +46,7 @@ ENDIF ()
 
 IF (TD_WINDOWS)
     MESSAGE("${Yellow} set compiler flag for Windows! ${ColourReset}")
-    SET(COMMON_FLAGS "/w /D_WIN32")
+    SET(COMMON_FLAGS "/w /D_WIN32 /Zi")
     SET(CMAKE_EXE_LINKER_FLAGS "${CMAKE_EXE_LINKER_FLAGS} /MANIFEST:NO")
     # IF (MSVC AND (MSVC_VERSION GREATER_EQUAL 1900))
     #  SET(COMMON_FLAGS "${COMMON_FLAGS} /Wv:18")
@@ -71,8 +71,8 @@ ELSE ()
     ENDIF ()
 
     IF (${SANITIZER} MATCHES "true")
-        SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS}     -Werror -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fsanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=null -fno-sanitize=alignment -static-libasan -g3")
-        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror -Wno-literal-suffix -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fsanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=null -fno-sanitize=alignment -static-libasan -g3")
+        SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS}     -Werror -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fsanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=null -fno-sanitize=alignment -g3")
+        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror -Wno-literal-suffix -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fsanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=null -fno-sanitize=alignment -g3")
 	MESSAGE(STATUS "Will compile with Address Sanitizer!")
     ELSE ()
         SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Werror -Werror=return-type -fPIC -gdwarf-2 -g3")

contrib/CMakeLists.txt

@@ -100,8 +100,10 @@ endif(${BUILD_WITH_NURAFT})
 
 # addr2line
 if(${BUILD_ADDR2LINE})
+    if(NOT ${TD_WINDOWS})
         cat("${TD_SUPPORT_DIR}/libdwarf_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
         cat("${TD_SUPPORT_DIR}/addr2line_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+    endif(NOT ${TD_WINDOWS})
 endif(${BUILD_ADDR2LINE})
 
 # download dependencies
@@ -335,6 +337,7 @@ endif(${BUILD_WITH_SQLITE})
 
 # addr2line
 if(${BUILD_ADDR2LINE})
+    if(NOT ${TD_WINDOWS})    
         check_include_file( "sys/types.h"     HAVE_SYS_TYPES_H) 
         check_include_file( "sys/stat.h"      HAVE_SYS_STAT_H ) 
         check_include_file( "inttypes.h"      HAVE_INTTYPES_H   ) 
@@ -374,6 +377,7 @@ if(${BUILD_ADDR2LINE})
         add_library(addr2line STATIC "addr2line/addr2line.c")
         target_link_libraries(addr2line PUBLIC libdwarf dl z)
         target_include_directories(addr2line PUBLIC "libdwarf/src/lib/libdwarf" )
+    endif(NOT ${TD_WINDOWS})
 endif(${BUILD_ADDR2LINE})
 
 

docs-cn/14-reference/03-connector/csharp.mdx

@@ -18,7 +18,7 @@ import CSAsyncQuery from "../../07-develop/04-query-data/_cs_async.mdx"
 
 `TDengine.Connector` 是 TDengine 提供的 C# 语言连接器。C# 开发人员可以通过它开发存取 TDengine 集群数据的 C# 应用软件。
 
-`TDengine.Connector` 连接器支持通过 TDengine 客户端驱动（taosc）建立与 TDengine 运行实例的连接，提供数据写入、查询、订阅、schemaless 数据写入、参数绑定接口数据写入等功能 `TDengine.Connector` 目前暂未提供 REST 连接方式，用户可以参考 [RESTful APIs](https://docs.taosdata.com//reference/restful-api/) 文档自行编写。
+`TDengine.Connector` 连接器支持通过 TDengine 客户端驱动（taosc）建立与 TDengine 运行实例的连接，提供数据写入、查询、订阅、schemaless 数据写入、参数绑定接口数据写入等功能 `TDengine.Connector` 目前暂未提供 REST 连接方式，用户可以参考 [REST API](/reference/rest-api/) 文档自行编写。
 
 本文介绍如何在 Linux 或 Windows 环境中安装 `TDengine.Connector`，并通过 `TDengine.Connector` 连接 TDengine 集群，进行数据写入、查询等基本操作。
 

docs-en/14-reference/03-connector/csharp.mdx

@@ -19,7 +19,7 @@ import CSAsyncQuery from "../../07-develop/04-query-data/_cs_async.mdx"
 
 `TDengine.Connector` is a C# language connector provided by TDengine that allows C# developers to develop C# applications that access TDengine cluster data.
 
-The `TDengine.Connector` connector supports connect to TDengine instances via the TDengine client driver (taosc), providing data writing, querying, subscription, schemaless writing, bind interface, etc. The `TDengine.Connector` currently does not provide a REST connection interface. Developers can write their RESTful application by referring to the [RESTful APIs](https://docs.taosdata.com//reference/restful-api/) documentation.
+The `TDengine.Connector` connector supports connect to TDengine instances via the TDengine client driver (taosc), providing data writing, querying, subscription, schemaless writing, bind interface, etc. The `TDengine.Connector` currently does not provide a REST connection interface. Developers can write their RESTful application by referring to the [REST API](/reference/rest-api/) documentation.
 
 This article describes how to install `TDengine.Connector` in a Linux or Windows environment and connect to TDengine clusters via `TDengine.Connector` to perform basic operations such as data writing and querying.
 

example/src/tstream.c

@@ -82,9 +82,7 @@ int32_t create_stream() {
   /*const char* sql = "select sum(k) from tu1 interval(10m)";*/
   /*pRes = tmq_create_stream(pConn, "stream1", "out1", sql);*/
   pRes = taos_query(
-      pConn,
-      "create stream stream1 trigger at_once into outstb as select _wstartts, min(k), max(k), sum(k) as sum_of_k "
-      "from tu1 interval(10m)");
+      pConn, "create stream stream1 trigger at_once into outstb as select _wstartts, sum(k) from tu1 interval(10m)");
   if (taos_errno(pRes) != 0) {
     printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes));
     return -1;

include/common/tmsg.h

@@ -244,12 +244,12 @@ typedef struct {
   const void* pMsg;
 } SSubmitMsgIter;
 
-int32_t tInitSubmitMsgIter(const SSubmitReq* pMsg, SSubmitMsgIter* pIter);
+int32_t tInitSubmitMsgIter(SSubmitReq* pMsg, SSubmitMsgIter* pIter);
 int32_t tGetSubmitMsgNext(SSubmitMsgIter* pIter, SSubmitBlk** pPBlock);
 int32_t tInitSubmitBlkIter(SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, SSubmitBlkIter* pIter);
 STSRow* tGetSubmitBlkNext(SSubmitBlkIter* pIter);
 // for debug
-int32_t tPrintFixedSchemaSubmitReq(const SSubmitReq* pReq, STSchema* pSchema);
+int32_t tPrintFixedSchemaSubmitReq(SSubmitReq* pReq, STSchema* pSchema);
 
 typedef struct {
   int32_t code;
@@ -1697,7 +1697,7 @@ int32_t tDecodeSRSmaParam(SDecoder* pCoder, SRSmaParam* pRSmaParam);
 
 // TDMT_VND_CREATE_STB ==============
 typedef struct SVCreateStbReq {
-  const char*    name;
+  char*          name;
   tb_uid_t       suid;
   int8_t         rollup;
   SSchemaWrapper schema;
@@ -1710,7 +1710,7 @@ int tDecodeSVCreateStbReq(SDecoder* pCoder, SVCreateStbReq* pReq);
 
 // TDMT_VND_DROP_STB ==============
 typedef struct SVDropStbReq {
-  const char* name;
+  char*    name;
   tb_uid_t suid;
 } SVDropStbReq;
 
@@ -1723,13 +1723,13 @@ typedef struct SVCreateTbReq {
   int32_t  flags;
   tb_uid_t uid;
   int64_t  ctime;
-  const char* name;
+  char*    name;
   int32_t  ttl;
   int8_t   type;
   union {
     struct {
       tb_uid_t suid;
-      const uint8_t* pTag;
+      uint8_t* pTag;
     } ctb;
     struct {
       SSchemaWrapper schema;
@@ -1777,7 +1777,7 @@ int32_t tDeserializeSVCreateTbBatchRsp(void* buf, int32_t bufLen, SVCreateTbBatc
 
 // TDMT_VND_DROP_TABLE =================
 typedef struct {
-  const char* name;
+  char*  name;
   int8_t igNotExists;
 } SVDropTbReq;
 
@@ -1809,9 +1809,9 @@ int32_t tDecodeSVDropTbBatchRsp(SDecoder* pCoder, SVDropTbBatchRsp* pRsp);
 
 // TDMT_VND_ALTER_TABLE =====================
 typedef struct {
-  const char* tbName;
+  char*  tbName;
   int8_t action;
-  const char* colName;
+  char*  colName;
   // TSDB_ALTER_TABLE_ADD_COLUMN
   int8_t  type;
   int8_t  flags;
@@ -1820,17 +1820,17 @@ typedef struct {
   // TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES
   int32_t colModBytes;
   // TSDB_ALTER_TABLE_UPDATE_COLUMN_NAME
-  const char* colNewName;
+  char* colNewName;
   // TSDB_ALTER_TABLE_UPDATE_TAG_VAL
-  const char*    tagName;
+  char*    tagName;
   int8_t   isNull;
   uint32_t nTagVal;
-  const uint8_t* pTagVal;
+  uint8_t* pTagVal;
   // TSDB_ALTER_TABLE_UPDATE_OPTIONS
   int8_t  updateTTL;
   int32_t newTTL;
   int8_t  updateComment;
-  const char* newComment;
+  char*   newComment;
 } SVAlterTbReq;
 
 int32_t tEncodeSVAlterTbReq(SEncoder* pEncoder, const SVAlterTbReq* pReq);
@@ -2020,7 +2020,7 @@ static FORCE_INLINE void tFreeClientHbBatchRsp(void* pRsp) {
 
 int32_t tSerializeSClientHbBatchRsp(void* buf, int32_t bufLen, const SClientHbBatchRsp* pBatchRsp);
 int32_t tDeserializeSClientHbBatchRsp(void* buf, int32_t bufLen, SClientHbBatchRsp* pBatchRsp);
-void tFreeSClientHbBatchRsp(SClientHbBatchRsp *pBatchRsp);
+void    tFreeSClientHbBatchRsp(SClientHbBatchRsp* pBatchRsp);
 
 static FORCE_INLINE int32_t tEncodeSKv(SEncoder* pEncoder, const SKv* pKv) {
   if (tEncodeI32(pEncoder, pKv->key) < 0) return -1;
@@ -2267,8 +2267,8 @@ typedef struct {
   int64_t  interval;
   int64_t  offset;  // use unit by precision of DB
   int64_t  sliding;
-  const char* expr;  // sma expression
-  const char* tagsFilter;
+  char*    expr;  // sma expression
+  char*    tagsFilter;
 } STSma;  // Time-range-wise SMA
 
 typedef STSma SVCreateTSmaReq;
@@ -2600,7 +2600,7 @@ typedef struct {
   int64_t       uid;
   int32_t       sver;
   uint32_t      nData;
-  const uint8_t* pData;
+  uint8_t*      pData;
   SVCreateTbReq cTbReq;
 } SVSubmitBlk;
 

include/libs/index/index.h

@@ -16,9 +16,11 @@
 #ifndef _TD_INDEX_H_
 #define _TD_INDEX_H_
 
+#include "nodes.h"
 #include "os.h"
 #include "taoserror.h"
 #include "tarray.h"
+#include "tglobal.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -189,6 +191,12 @@ void        indexTermDestroy(SIndexTerm* p);
  */
 void indexInit();
 
+/* index filter */
+typedef enum { SFLT_NOT_INDEX, SFLT_COARSE_INDEX, SFLT_ACCURATE_INDEX } SIdxFltStatus;
+
+SIdxFltStatus idxGetFltStatus(SNode* pFilterNode);
+
+int32_t doFilterTag(const SNode* pFilterNode, SArray* result);
 /*
  * destory index env
  *

include/libs/nodes/nodes.h

@@ -61,8 +61,8 @@ extern "C" {
 
 #define DESTORY_LIST(list)    \
   do {                        \
-    nodesDestroyList(list); \
-    list = NULL;            \
+    nodesDestroyList((list)); \
+    (list) = NULL;            \
   } while (0)
 
 typedef enum ENodeType {
@@ -96,6 +96,7 @@ typedef enum ENodeType {
   QUERY_NODE_EXPLAIN_OPTIONS,
   QUERY_NODE_STREAM_OPTIONS,
   QUERY_NODE_TOPIC_OPTIONS,
+  QUERY_NODE_LEFT_VALUE,
 
   // Statement nodes are used in parser and planner module.
   QUERY_NODE_SET_OPERATOR,

include/libs/nodes/querynodes.h

@@ -81,6 +81,7 @@ typedef struct SValueNode {
   char*     literal;
   bool      isDuration;
   bool      translate;
+  bool      notReserved;
   int16_t   placeholderNo;
   union {
     bool     b;
@@ -93,6 +94,10 @@ typedef struct SValueNode {
   char    unit;
 } SValueNode;
 
+typedef struct SLeftValueNode {
+  ENodeType type;
+} SLeftValueNode;
+
 typedef struct SOperatorNode {
   SExprNode     node;  // QUERY_NODE_OPERATOR
   EOperatorType opType;

include/libs/stream/tstream.h

@@ -114,17 +114,12 @@ static FORCE_INLINE void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit)
 int32_t streamDataBlockEncode(void** buf, const SStreamDataBlock* pOutput);
 void*   streamDataBlockDecode(const void* buf, SStreamDataBlock* pInput);
 
-typedef struct {
-  void* inputHandle;
-  void* executor;
-} SStreamRunner;
-
 typedef struct {
   int8_t parallelizable;
   char*  qmsg;
   // followings are not applicable to encoder and decoder
-  int8_t         numOfRunners;
-  SStreamRunner* runners;
+  void* inputHandle;
+  void* executor;
 } STaskExec;
 
 typedef struct {
@@ -320,17 +315,15 @@ int32_t streamEnqueueDataSubmit(SStreamTask* pTask, SStreamDataSubmit* input);
 int32_t streamEnqueueDataBlk(SStreamTask* pTask, SStreamDataBlock* input);
 int32_t streamDequeueOutput(SStreamTask* pTask, void** output);
 
-int32_t streamExecTask(SStreamTask* pTask, SMsgCb* pMsgCb, const void* input, int32_t inputType, int32_t workId);
-
 int32_t streamTaskRun(SStreamTask* pTask);
 
 int32_t streamTaskHandleInput(SStreamTask* pTask, void* data);
 
 int32_t streamTaskProcessRunReq(SStreamTask* pTask, SMsgCb* pMsgCb);
-int32_t streamTaskProcessDispatchReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchReq* pReq, SRpcMsg* pMsg);
-int32_t streamTaskProcessDispatchRsp(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchRsp* pRsp);
-int32_t streamTaskProcessRecoverReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamTaskRecoverReq* pReq, SRpcMsg* pMsg);
-int32_t streamTaskProcessRecoverRsp(SStreamTask* pTask, SStreamTaskRecoverRsp* pRsp);
+int32_t streamProcessDispatchReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchReq* pReq, SRpcMsg* pMsg);
+int32_t streamProcessDispatchRsp(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamDispatchRsp* pRsp);
+int32_t streamProcessRecoverReq(SStreamTask* pTask, SMsgCb* pMsgCb, SStreamTaskRecoverReq* pReq, SRpcMsg* pMsg);
+int32_t streamProcessRecoverRsp(SStreamTask* pTask, SStreamTaskRecoverRsp* pRsp);
 
 #ifdef __cplusplus
 }

include/util/tdef.h

@@ -132,6 +132,7 @@ typedef enum EOperatorType {
   OP_TYPE_MOD,
   // unary arithmetic operator
   OP_TYPE_MINUS,
+  OP_TYPE_ASSIGN,
 
   // bit operator
   OP_TYPE_BIT_AND,

include/util/tencode.h

@@ -39,7 +39,7 @@ typedef struct {
 } SEncoder;
 
 typedef struct {
-  const uint8_t* data;
+  uint8_t*      data;
   uint32_t      size;
   uint32_t      pos;
   SCoderMem*    mList;
@@ -120,7 +120,7 @@ static int32_t tEncodeCStrWithLen(SEncoder* pCoder, const char* val, uint32_t le
 static int32_t tEncodeCStr(SEncoder* pCoder, const char* val);
 
 /* ------------------------ DECODE ------------------------ */
-void           tDecoderInit(SDecoder* pCoder, const uint8_t* data, uint32_t size);
+void           tDecoderInit(SDecoder* pCoder, uint8_t* data, uint32_t size);
 void           tDecoderClear(SDecoder* SDecoder);
 int32_t        tStartDecode(SDecoder* pCoder);
 void           tEndDecode(SDecoder* pCoder);
@@ -141,9 +141,9 @@ static int32_t tDecodeU64v(SDecoder* pCoder, uint64_t* val);
 static int32_t tDecodeI64v(SDecoder* pCoder, int64_t* val);
 static int32_t tDecodeFloat(SDecoder* pCoder, float* val);
 static int32_t tDecodeDouble(SDecoder* pCoder, double* val);
-static int32_t tDecodeBinary(SDecoder* pCoder, const uint8_t** val, uint32_t* len);
-static int32_t tDecodeCStrAndLen(SDecoder* pCoder, const char** val, uint32_t* len);
-static int32_t tDecodeCStr(SDecoder* pCoder, const char** val);
+static int32_t tDecodeBinary(SDecoder* pCoder, uint8_t** val, uint32_t* len);
+static int32_t tDecodeCStrAndLen(SDecoder* pCoder, char** val, uint32_t* len);
+static int32_t tDecodeCStr(SDecoder* pCoder, char** val);
 static int32_t tDecodeCStrTo(SDecoder* pCoder, char* val);
 
 /* ------------------------ IMPL ------------------------ */
@@ -317,7 +317,7 @@ static FORCE_INLINE int32_t tDecodeI16v(SDecoder* pCoder, int16_t* val) {
   if (tDecodeU16v(pCoder, &tval) < 0) {
     return -1;
   }
-  *val = ZIGZAGD(int16_t, tval);
+  if (val) *val = ZIGZAGD(int16_t, tval);
   return 0;
 }
 
@@ -331,7 +331,7 @@ static FORCE_INLINE int32_t tDecodeI32v(SDecoder* pCoder, int32_t* val) {
   if (tDecodeU32v(pCoder, &tval) < 0) {
     return -1;
   }
-  *val = ZIGZAGD(int32_t, tval);
+  if (val) *val = ZIGZAGD(int32_t, tval);
   return 0;
 }
 
@@ -345,7 +345,7 @@ static FORCE_INLINE int32_t tDecodeI64v(SDecoder* pCoder, int64_t* val) {
   if (tDecodeU64v(pCoder, &tval) < 0) {
     return -1;
   }
-  *val = ZIGZAGD(int64_t, tval);
+  if (val) *val = ZIGZAGD(int64_t, tval);
   return 0;
 }
 
@@ -377,7 +377,7 @@ static FORCE_INLINE int32_t tDecodeDouble(SDecoder* pCoder, double* val) {
   return 0;
 }
 
-static FORCE_INLINE int32_t tDecodeBinary(SDecoder* pCoder, const uint8_t** val, uint32_t* len) {
+static FORCE_INLINE int32_t tDecodeBinary(SDecoder* pCoder, uint8_t** val, uint32_t* len) {
   if (tDecodeU32v(pCoder, len) < 0) return -1;
 
   if (TD_CODER_CHECK_CAPACITY_FAILED(pCoder, *len)) return -1;
@@ -389,19 +389,19 @@ static FORCE_INLINE int32_t tDecodeBinary(SDecoder* pCoder, const uint8_t** val,
   return 0;
 }
 
-static FORCE_INLINE int32_t tDecodeCStrAndLen(SDecoder* pCoder, const char** val, uint32_t* len) {
-  if (tDecodeBinary(pCoder, (const uint8_t**)val, len) < 0) return -1;
+static FORCE_INLINE int32_t tDecodeCStrAndLen(SDecoder* pCoder, char** val, uint32_t* len) {
+  if (tDecodeBinary(pCoder, (uint8_t**)val, len) < 0) return -1;
   (*len) -= 1;
   return 0;
 }
 
-static FORCE_INLINE int32_t tDecodeCStr(SDecoder* pCoder, const char** val) {
+static FORCE_INLINE int32_t tDecodeCStr(SDecoder* pCoder, char** val) {
   uint32_t len;
   return tDecodeCStrAndLen(pCoder, val, &len);
 }
 
 static int32_t tDecodeCStrTo(SDecoder* pCoder, char* val) {
-  const char* pStr;
+  char*    pStr;
   uint32_t len;
   if (tDecodeCStrAndLen(pCoder, &pStr, &len) < 0) return -1;
 

source/common/src/tmsg.c

@@ -28,7 +28,7 @@
 #undef TD_MSG_SEG_CODE_
 #include "tmsgdef.h"
 
-int32_t tInitSubmitMsgIter(const SSubmitReq *pMsg, SSubmitMsgIter *pIter) {
+int32_t tInitSubmitMsgIter(SSubmitReq *pMsg, SSubmitMsgIter *pIter) {
   if (pMsg == NULL) {
     terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
     return -1;
@@ -102,7 +102,7 @@ STSRow *tGetSubmitBlkNext(SSubmitBlkIter *pIter) {
   }
 }
 
-int32_t tPrintFixedSchemaSubmitReq(const SSubmitReq *pReq, STSchema *pTschema) {
+int32_t tPrintFixedSchemaSubmitReq(SSubmitReq *pReq, STSchema *pTschema) {
   SSubmitMsgIter msgIter = {0};
   if (tInitSubmitMsgIter(pReq, &msgIter) < 0) return -1;
   while (true) {

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

@@ -113,6 +113,8 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
     SRpcMsg *pMsg = *(SRpcMsg **)taosArrayGet(pArray, i);
     SRpcMsg  rsp = {.info = pMsg->info};
 
+    vnodePreprocessReq(pVnode->pImpl, pMsg);
+
     int32_t ret = syncPropose(vnodeGetSyncHandle(pVnode->pImpl), pMsg, false);
     if (ret == TAOS_SYNC_PROPOSE_NOT_LEADER) {
       dTrace("msg:%p, is redirect since not leader, vgId:%d ", pMsg, pVnode->vgId);

source/dnode/snode/src/snode.c

@@ -57,9 +57,7 @@ void sndMetaDelete(SStreamMeta *pMeta) {
 }
 
 int32_t sndMetaDeployTask(SStreamMeta *pMeta, SStreamTask *pTask) {
-  for (int i = 0; i < pTask->exec.numOfRunners; i++) {
-    pTask->exec.runners[i].executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, NULL);
-  }
+  pTask->exec.executor = qCreateStreamExecTaskInfo(pTask->exec.qmsg, NULL);
   return taosHashPut(pMeta->pHash, &pTask->taskId, sizeof(int32_t), pTask, sizeof(void *));
 }
 

source/dnode/vnode/CMakeLists.txt

@@ -48,7 +48,6 @@ target_sources(
     # tq
     "src/tq/tq.c"
     "src/tq/tqCommit.c"
-    "src/tq/tqMetaStore.c"
     "src/tq/tqOffset.c"
     "src/tq/tqPush.c"
     "src/tq/tqRead.c"

source/dnode/vnode/inc/vnode.h

@@ -51,7 +51,7 @@ int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs);
 void    vnodeDestroy(const char *path, STfs *pTfs);
 SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb);
 void    vnodeClose(SVnode *pVnode);
-int32_t vnodePreprocessWriteReqs(SVnode *pVnode, SArray *pMsgs, int64_t *version);
+int32_t vnodePreprocessReq(SVnode *pVnode, SRpcMsg *pMsg);
 int32_t vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg *pRsp);
 int32_t vnodeProcessCMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
 int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
@@ -126,7 +126,7 @@ STqReadHandle *tqInitSubmitMsgScanner(SMeta *pMeta);
 void    tqReadHandleSetColIdList(STqReadHandle *pReadHandle, SArray *pColIdList);
 int32_t tqReadHandleSetTbUidList(STqReadHandle *pHandle, const SArray *tbUidList);
 int32_t tqReadHandleAddTbUidList(STqReadHandle *pHandle, const SArray *tbUidList);
-int32_t tqReadHandleRemoveTbUidList(STqReadHandle* pHandle, const SArray* tbUidList);
+int32_t tqReadHandleRemoveTbUidList(STqReadHandle *pHandle, const SArray *tbUidList);
 
 int32_t tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitReq *pMsg, int64_t ver);
 bool    tqNextDataBlock(STqReadHandle *pHandle);
@@ -177,7 +177,7 @@ struct SMetaEntry {
   int64_t  version;
   int8_t   type;
   tb_uid_t uid;
-  const char *name;
+  char    *name;
   union {
     struct {
       SSchemaWrapper schema;
@@ -187,7 +187,7 @@ struct SMetaEntry {
       int64_t  ctime;
       int32_t  ttlDays;
       tb_uid_t suid;
-      const uint8_t *pTags;
+      uint8_t *pTags;
     } ctbEntry;
     struct {
       int64_t        ctime;

source/dnode/vnode/src/inc/tq.h

@@ -41,45 +41,6 @@ extern "C" {
 #define tqTrace(...) do { if (tqDebugFlag & DEBUG_TRACE) { taosPrintLog("TQ  ", DEBUG_TRACE, tqDebugFlag, __VA_ARGS__); }} while(0)
 // clang-format on
 
-#define TQ_BUFFER_SIZE 4
-
-#define TQ_BUCKET_MASK 0xFF
-#define TQ_BUCKET_SIZE 256
-
-#define TQ_PAGE_SIZE 4096
-// key + offset + size
-#define TQ_IDX_SIZE 24
-// 4096 / 24
-#define TQ_MAX_IDX_ONE_PAGE 170
-// 24 * 170
-#define TQ_IDX_PAGE_BODY_SIZE 4080
-// 4096 - 4080
-#define TQ_IDX_PAGE_HEAD_SIZE 16
-
-#define TQ_ACTION_CONST      0
-#define TQ_ACTION_INUSE      1
-#define TQ_ACTION_INUSE_CONT 2
-#define TQ_ACTION_INTXN      3
-
-#define TQ_SVER 0
-
-// TODO: inplace mode is not implemented
-#define TQ_UPDATE_INPLACE 0
-#define TQ_UPDATE_APPEND  1
-
-#define TQ_DUP_INTXN_REWRITE 0
-#define TQ_DUP_INTXN_REJECT  2
-
-static inline bool tqUpdateAppend(int32_t tqConfigFlag) { return tqConfigFlag & TQ_UPDATE_APPEND; }
-
-static inline bool tqDupIntxnReject(int32_t tqConfigFlag) { return tqConfigFlag & TQ_DUP_INTXN_REJECT; }
-
-static const int8_t TQ_CONST_DELETE = TQ_ACTION_CONST;
-
-#define TQ_DELETE_TOKEN (void*)&TQ_CONST_DELETE
-
-typedef enum { TQ_ITEM_READY, TQ_ITEM_PROCESS, TQ_ITEM_EMPTY } STqItemStatus;
-
 typedef struct STqOffsetCfg   STqOffsetCfg;
 typedef struct STqOffsetStore STqOffsetStore;
 
@@ -98,53 +59,6 @@ struct STqReadHandle {
   STSchema*         pSchema;
 };
 
-typedef struct {
-  int16_t ver;
-  int16_t action;
-  int32_t checksum;
-  int64_t ssize;
-  char    content[];
-} STqSerializedHead;
-
-typedef int32_t (*FTqSerialize)(const void* pObj, STqSerializedHead** ppHead);
-typedef int32_t (*FTqDeserialize)(void* self, const STqSerializedHead* pHead, void** ppObj);
-typedef void (*FTqDelete)(void*);
-
-typedef struct {
-  int64_t key;
-  int64_t offset;
-  int64_t serializedSize;
-  void*   valueInUse;
-  void*   valueInTxn;
-} STqMetaHandle;
-
-typedef struct STqMetaList {
-  STqMetaHandle       handle;
-  struct STqMetaList* next;
-  // struct STqMetaList* inTxnPrev;
-  // struct STqMetaList* inTxnNext;
-  struct STqMetaList* unpersistPrev;
-  struct STqMetaList* unpersistNext;
-} STqMetaList;
-
-typedef struct {
-  STQ*         pTq;
-  STqMetaList* bucket[TQ_BUCKET_SIZE];
-  // a table head
-  STqMetaList* unpersistHead;
-  // topics that are not connectted
-  STqMetaList* unconnectTopic;
-
-  TdFilePtr pFile;
-  TdFilePtr pIdxFile;
-
-  char*          dirPath;
-  int32_t        tqConfigFlag;
-  FTqSerialize   pSerializer;
-  FTqDeserialize pDeserializer;
-  FTqDelete      pDeleter;
-} STqMetaStore;
-
 typedef struct {
   int64_t  consumerId;
   int32_t  epoch;
@@ -174,7 +88,6 @@ typedef struct {
 
 struct STQ {
   char*     path;
-  // STqMetaStore* tqMeta;
   SHashObj* pushMgr;  // consumerId -> STqExec*
   SHashObj* execs;    // subKey -> STqExec
   SHashObj* pStreamTasks;
@@ -190,87 +103,10 @@ typedef struct {
 
 static STqMgmt tqMgmt;
 
-typedef struct {
-  int8_t         status;
-  int64_t        offset;
-  qTaskInfo_t    task;
-  STqReadHandle* pReadHandle;
-} STqTaskItem;
-
-// new version
-typedef struct {
-  int64_t     firstOffset;
-  int64_t     lastOffset;
-  STqTaskItem output[TQ_BUFFER_SIZE];
-} STqBuffer;
-
-typedef struct {
-  char            topicName[TSDB_TOPIC_FNAME_LEN];
-  char*           sql;
-  char*           logicalPlan;
-  char*           physicalPlan;
-  char*           qmsg;
-  STqBuffer       buffer;
-  SWalReadHandle* pReadhandle;
-} STqTopic;
-
-typedef struct {
-  int64_t consumerId;
-  int32_t epoch;
-  char    cgroup[TSDB_TOPIC_FNAME_LEN];
-  SArray* topics;  // SArray<STqTopic>
-} STqConsumer;
-
-typedef struct {
-  int8_t      type;
-  int8_t      nodeType;
-  int8_t      reserved[6];
-  int64_t     streamId;
-  qTaskInfo_t task;
-  // TODO sync function
-} STqStreamPusher;
-
-typedef struct {
-  int8_t inited;
-  tmr_h  timer;
-} STqPushMgmt;
-
-static STqPushMgmt tqPushMgmt;
-
 // init once
 int  tqInit();
 void tqCleanUp();
 
-// open in each vnode
-// required by vnode
-
-int32_t tqSerializeConsumer(const STqConsumer*, STqSerializedHead**);
-int32_t tqDeserializeConsumer(STQ*, const STqSerializedHead*, STqConsumer**);
-
-static int FORCE_INLINE tqQueryExecuting(int32_t status) { return status; }
-
-// tqMetaStore.h
-STqMetaStore* tqStoreOpen(STQ* pTq, const char* path, FTqSerialize pSerializer, FTqDeserialize pDeserializer,
-                          FTqDelete pDeleter, int32_t tqConfigFlag);
-int32_t       tqStoreClose(STqMetaStore*);
-// int32_t       tqStoreDelete(TqMetaStore*);
-// int32_t       tqStoreCommitAll(TqMetaStore*);
-int32_t tqStorePersist(STqMetaStore*);
-// clean deleted idx and data from persistent file
-int32_t tqStoreCompact(STqMetaStore*);
-
-void* tqHandleGet(STqMetaStore*, int64_t key);
-// make it unpersist
-void*   tqHandleTouchGet(STqMetaStore*, int64_t key);
-int32_t tqHandleMovePut(STqMetaStore*, int64_t key, void* value);
-int32_t tqHandleCopyPut(STqMetaStore*, int64_t key, void* value, size_t vsize);
-// delete committed kv pair
-// notice that a delete action still needs to be committed
-int32_t tqHandleDel(STqMetaStore*, int64_t key);
-int32_t tqHandlePurge(STqMetaStore*, int64_t key);
-int32_t tqHandleCommit(STqMetaStore*, int64_t key);
-int32_t tqHandleAbort(STqMetaStore*, int64_t key);
-
 // tqOffset
 STqOffsetStore* STqOffsetOpen(STqOffsetCfg*);
 void            STqOffsetClose(STqOffsetStore*);

source/dnode/vnode/src/inc/vnodeInt.h

@@ -104,7 +104,7 @@ int          tsdbOpen(SVnode* pVnode, STsdb** ppTsdb, const char* dir, STsdbKeep
 int          tsdbClose(STsdb** pTsdb);
 int          tsdbBegin(STsdb* pTsdb);
 int          tsdbCommit(STsdb* pTsdb);
-int          tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, const SSubmitReq* pMsg);
+int          tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, SSubmitReq* pMsg);
 int          tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSubmitRsp* pRsp);
 int          tsdbInsertTableData(STsdb* pTsdb, SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, SSubmitBlkRsp* pRsp);
 tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
@@ -123,11 +123,7 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen);
 int32_t tqProcessVgDeleteReq(STQ* pTq, char* msg, int32_t msgLen);
 int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId);
 int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen);
-#if 0
-int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen, int32_t workerId);
-int32_t tqProcessStreamTrigger(STQ* pTq, void* data, int32_t dataLen, int32_t workerId);
-#endif
-int32_t tqProcessStreamTriggerNew(STQ* pTq, SSubmitReq* data);
+int32_t tqProcessStreamTrigger(STQ* pTq, SSubmitReq* data);
 int32_t tqProcessTaskRunReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskDispatchReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRecoverReq(STQ* pTq, SRpcMsg* pMsg);

source/dnode/vnode/src/meta/metaTable.c

@@ -30,9 +30,9 @@ int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
   int         vLen = 0;
   const void *pKey = NULL;
   const void *pVal = NULL;
-  void *      pBuf = NULL;
+  void       *pBuf = NULL;
   int32_t     szBuf = 0;
-  void *      p = NULL;
+  void       *p = NULL;
   SMetaReader mr = {0};
 
   // validate req
@@ -71,9 +71,9 @@ _err:
 }
 
 int metaDropSTable(SMeta *pMeta, int64_t verison, SVDropStbReq *pReq) {
-  TBC *       pNameIdxc = NULL;
-  TBC *       pUidIdxc = NULL;
-  TBC *       pCtbIdxc = NULL;
+  TBC        *pNameIdxc = NULL;
+  TBC        *pUidIdxc = NULL;
+  TBC        *pCtbIdxc = NULL;
   SCtbIdxKey *pCtbIdxKey;
   const void *pKey = NULL;
   int         nKey;
@@ -134,8 +134,8 @@ _err:
 int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
   SMetaEntry  oStbEntry = {0};
   SMetaEntry  nStbEntry = {0};
-  TBC *       pUidIdxc = NULL;
-  TBC *       pTbDbc = NULL;
+  TBC        *pUidIdxc = NULL;
+  TBC        *pTbDbc = NULL;
   const void *pData;
   int         nData;
   int64_t     oversion;
@@ -165,7 +165,9 @@ int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
   ret = tdbTbcGet(pTbDbc, NULL, NULL, &pData, &nData);
   ASSERT(ret == 0);
 
-  tDecoderInit(&dc, pData, nData);
+  oStbEntry.pBuf = taosMemoryMalloc(nData);
+  memcpy(oStbEntry.pBuf, pData, nData);
+  tDecoderInit(&dc, oStbEntry.pBuf, nData);
   metaDecodeEntry(&dc, &oStbEntry);
 
   nStbEntry.version = version;
@@ -193,6 +195,7 @@ int metaAlterSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {
   // update uid index
   tdbTbcUpsert(pUidIdxc, &pReq->suid, sizeof(tb_uid_t), &version, sizeof(version), 0);
 
+  if (oStbEntry.pBuf) taosMemoryFree(oStbEntry.pBuf);
   metaULock(pMeta);
   tDecoderClear(&dc);
   tdbTbcClose(pTbDbc);
@@ -220,9 +223,6 @@ int metaCreateTable(SMeta *pMeta, int64_t version, SVCreateTbReq *pReq) {
     terrno = TSDB_CODE_TDB_TABLE_ALREADY_EXIST;
     metaReaderClear(&mr);
     return -1;
-  } else {
-    pReq->uid = tGenIdPI64();
-    pReq->ctime = taosGetTimestampMs();
   }
   metaReaderClear(&mr);
 
@@ -256,9 +256,9 @@ _err:
 }
 
 int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUids) {
-  TBC *       pTbDbc = NULL;
-  TBC *       pUidIdxc = NULL;
-  TBC *       pNameIdxc = NULL;
+  TBC        *pTbDbc = NULL;
+  TBC        *pUidIdxc = NULL;
+  TBC        *pNameIdxc = NULL;
   const void *pData;
   int         nData;
   tb_uid_t    uid;
@@ -377,14 +377,14 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUi
 }
 
 static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq) {
-  void *          pVal = NULL;
+  void           *pVal = NULL;
   int             nVal = 0;
-  const void *    pData = NULL;
+  const void     *pData = NULL;
   int             nData = 0;
   int             ret = 0;
   tb_uid_t        uid;
   int64_t         oversion;
-  SSchema *       pColumn = NULL;
+  SSchema        *pColumn = NULL;
   SMetaEntry      entry = {0};
   SSchemaWrapper *pSchema;
   int             c;
@@ -420,7 +420,9 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl
 
   // get table entry
   SDecoder dc = {0};
-  tDecoderInit(&dc, pData, nData);
+  entry.pBuf = taosMemoryMalloc(nData);
+  memcpy(entry.pBuf, pData, nData);
+  tDecoderInit(&dc, entry.pBuf, nData);
   ret = metaDecodeEntry(&dc, &entry);
   ASSERT(ret == 0);
 
@@ -530,7 +532,7 @@ _err:
 static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq) {
   SMetaEntry  ctbEntry = {0};
   SMetaEntry  stbEntry = {0};
-  void *      pVal = NULL;
+  void       *pVal = NULL;
   int         nVal = 0;
   int         ret;
   int         c;
@@ -561,7 +563,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
   oversion = *(int64_t *)pData;
 
   // search table.db
-  TBC *    pTbDbc = NULL;
+  TBC     *pTbDbc = NULL;
   SDecoder dc1 = {0};
   SDecoder dc2 = {0};
 
@@ -585,7 +587,7 @@ static int metaUpdateTableTagVal(SMeta *pMeta, int64_t version, SVAlterTbReq *pA
   metaDecodeEntry(&dc2, &stbEntry);
 
   SSchemaWrapper *pTagSchema = &stbEntry.stbEntry.schemaTag;
-  SSchema *       pColumn = NULL;
+  SSchema        *pColumn = NULL;
   int32_t         iCol = 0;
   for (;;) {
     pColumn = NULL;
@@ -690,8 +692,8 @@ int metaAlterTable(SMeta *pMeta, int64_t version, SVAlterTbReq *pReq) {
 
 static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
   STbDbKey tbDbKey;
-  void *   pKey = NULL;
-  void *   pVal = NULL;
+  void    *pKey = NULL;
+  void    *pVal = NULL;
   int      kLen = 0;
   int      vLen = 0;
   SEncoder coder = {0};
@@ -806,14 +808,14 @@ static void metaDestroyTagIdxKey(STagIdxKey *pTagIdxKey) {
 }
 
 static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
-  void *         pData = NULL;
+  void          *pData = NULL;
   int            nData = 0;
   STbDbKey       tbDbKey = {0};
   SMetaEntry     stbEntry = {0};
-  STagIdxKey *   pTagIdxKey = NULL;
+  STagIdxKey    *pTagIdxKey = NULL;
   int32_t        nTagIdxKey;
   const SSchema *pTagColumn;       // = &stbEntry.stbEntry.schema.pSchema[0];
-  const void *   pTagData = NULL;  //
+  const void    *pTagData = NULL;  //
   SDecoder       dc = {0};
 
   // get super table
@@ -855,7 +857,7 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) {
 
 static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
   SEncoder              coder = {0};
-  void *                pVal = NULL;
+  void                 *pVal = NULL;
   int                   vLen = 0;
   int                   rcode = 0;
   SSkmDbKey             skmDbKey = {0};

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

@@ -217,7 +217,7 @@ int32_t tsdbInsertData2(SMemTable *pMemTb, int64_t version, const SVSubmitBlk *p
     if (tDecodeIsEnd(&dc)) break;
 
     // decode row
-    if (tDecodeBinary(&dc, (const uint8_t **)&tRow.pRow, &tRow.szRow) < 0) {
+    if (tDecodeBinary(&dc, (uint8_t **)&tRow.pRow, &tRow.szRow) < 0) {
       terrno = TSDB_CODE_INVALID_MSG;
       return -1;
     }
@@ -273,7 +273,7 @@ static FORCE_INLINE int32_t tsdbEncodeRow(SEncoder *pEncoder, const STsdbRow *pR
 
 static FORCE_INLINE int32_t tsdbDecodeRow(SDecoder *pDecoder, STsdbRow *pRow) {
   if (tDecodeI64(pDecoder, &pRow->version) < 0) return -1;
-  if (tDecodeBinary(pDecoder, (const uint8_t **)&pRow->pRow, &pRow->szRow) < 0) return -1;
+  if (tDecodeBinary(pDecoder, (uint8_t **)&pRow->pRow, &pRow->szRow) < 0) return -1;
   return 0;
 }
 

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

@@ -85,7 +85,7 @@ static FORCE_INLINE int tsdbCheckRowRange(STsdb *pTsdb, tb_uid_t uid, STSRow *ro
   return 0;
 }
 
-int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, const SSubmitReq *pMsg) {
+int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg) {
   ASSERT(pMsg != NULL);
   // STsdbMeta *    pMeta = pTsdb->tsdbMeta;
   SSubmitMsgIter msgIter = {0};
@@ -150,7 +150,6 @@ int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, const SSubmitReq *pMsg) {
         return -1;
       }
     }
-
   }
 
   if (terrno != TSDB_CODE_SUCCESS) return -1;

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

@@ -24,26 +24,66 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in
 static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp);
 static int vnodeProcessCreateTSmaReq(SVnode *pVnode, int64_t version, void *pReq, int len, SRpcMsg *pRsp);
 
-int vnodePreprocessWriteReqs(SVnode *pVnode, SArray *pMsgs, int64_t *version) {
-#if 0
-  SRpcMsg *pMsg;
-  SRpcMsg  *pRpc;
-
-  *version = pVnode->state.processed;
-  for (int i = 0; i < taosArrayGetSize(pMsgs); i++) {
-    pMsg = *(SRpcMsg **)taosArrayGet(pMsgs, i);
-    pRpc = pMsg;
-
-    // set request version
-    if (walWrite(pVnode->pWal, pVnode->state.processed++, pRpc->msgType, pRpc->pCont, pRpc->contLen) < 0) {
-      vError("vnode:%d  write wal error since %s", TD_VID(pVnode), terrstr());
-      return -1;
+int32_t vnodePreprocessReq(SVnode *pVnode, SRpcMsg *pMsg) {
+  SDecoder dc = {0};
+
+  switch (pMsg->msgType) {
+    case TDMT_VND_CREATE_TABLE: {
+      int64_t ctime = taosGetTimestampMs();
+      int32_t nReqs;
+
+      tDecoderInit(&dc, (uint8_t *)pMsg->pCont + sizeof(SMsgHead), pMsg->contLen - sizeof(SMsgHead));
+      tStartDecode(&dc);
+
+      tDecodeI32v(&dc, &nReqs);
+      for (int32_t iReq = 0; iReq < nReqs; iReq++) {
+        tb_uid_t uid = tGenIdPI64();
+        tStartDecode(&dc);
+
+        tDecodeI32v(&dc, NULL);
+        *(int64_t *)(dc.data + dc.pos) = uid;
+        *(int64_t *)(dc.data + dc.pos + 8) = ctime;
+
+        tEndDecode(&dc);
+      }
+
+      tEndDecode(&dc);
+      tDecoderClear(&dc);
+    } break;
+    case TDMT_VND_SUBMIT: {
+      SSubmitMsgIter msgIter = {0};
+      SSubmitReq    *pSubmitReq = (SSubmitReq *)pMsg->pCont;
+      SSubmitBlk    *pBlock = NULL;
+      int64_t        ctime = taosGetTimestampMs();
+      tb_uid_t       uid;
+
+      tInitSubmitMsgIter(pSubmitReq, &msgIter);
+
+      for (;;) {
+        tGetSubmitMsgNext(&msgIter, &pBlock);
+        if (pBlock == NULL) break;
+
+        if (msgIter.schemaLen > 0) {
+          uid = tGenIdPI64();
+
+          tDecoderInit(&dc, pBlock->data, msgIter.schemaLen);
+          tStartDecode(&dc);
+
+          tDecodeI32v(&dc, NULL);
+          *(int64_t *)(dc.data + dc.pos) = uid;
+          *(int64_t *)(dc.data + dc.pos + 8) = ctime;
+          pBlock->uid = htobe64(uid);
+
+          tEndDecode(&dc);
+          tDecoderClear(&dc);
         }
       }
 
-  walFsync(pVnode->pWal, false);
+    } break;
+    default:
+      break;
+  }
 
-#endif
   return 0;
 }
 
@@ -106,13 +146,6 @@ int vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg
                               pMsg->contLen - sizeof(SMsgHead)) < 0) {
       }
     } break;
-#if 0
-    case TDMT_VND_TASK_WRITE_EXEC: {
-      if (tqProcessTaskExec(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), pMsg->contLen - sizeof(SMsgHead),
-                            0) < 0) {
-      }
-    } break;
-#endif
     case TDMT_VND_ALTER_VNODE:
       break;
     default:
@@ -195,17 +228,6 @@ int vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
     case TDMT_VND_TASK_RECOVER_RSP:
       return tqProcessTaskRecoverRsp(pVnode->pTq, pMsg);
 
-#if 0
-    case TDMT_VND_TASK_PIPE_EXEC:
-    case TDMT_VND_TASK_MERGE_EXEC:
-      return tqProcessTaskExec(pVnode->pTq, msgstr, msgLen, 0);
-    case TDMT_VND_STREAM_TRIGGER:{
-      // refactor, avoid double free
-      int code = tqProcessStreamTrigger(pVnode->pTq, pMsg->pCont, pMsg->contLen, 0);
-      pMsg->pCont = NULL;
-      return code;
-    }
-#endif
     case TDMT_VND_QUERY_HEARTBEAT:
       return qWorkerProcessHbMsg(pVnode, pVnode->pQuery, pMsg);
     default:
@@ -675,7 +697,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
     goto _exit;
   }
 
-  for (int i = 0;;) {
+  for (;;) {
     tGetSubmitMsgNext(&msgIter, &pBlock);
     if (pBlock == NULL) break;
 

source/dnode/vnode/test/tqMetaTest.cpp

-#include <gtest/gtest.h>
-#include <cstring>
-#include <iostream>
-#include <queue>
-
-#include "tqMetaStore.h"
-
-struct Foo {
-  int32_t a;
-};
-
-int FooSerializer(const void* pObj, STqSerializedHead** ppHead) {
-  Foo* foo = (Foo*)pObj;
-  if ((*ppHead) == NULL || (*ppHead)->ssize < sizeof(STqSerializedHead) + sizeof(int32_t)) {
-    *ppHead = (STqSerializedHead*)taosMemoryRealloc(*ppHead, sizeof(STqSerializedHead) + sizeof(int32_t));
-    (*ppHead)->ssize = sizeof(STqSerializedHead) + sizeof(int32_t);
-  }
-  *(int32_t*)(*ppHead)->content = foo->a;
-  return (*ppHead)->ssize;
-}
-
-const void* FooDeserializer(const STqSerializedHead* pHead, void** ppObj) {
-  if (*ppObj == NULL) {
-    *ppObj = taosMemoryRealloc(*ppObj, sizeof(int32_t));
-  }
-  Foo* pFoo = *(Foo**)ppObj;
-  pFoo->a = *(int32_t*)pHead->content;
-  return NULL;
-}
-
-void FooDeleter(void* pObj) { taosMemoryFree(pObj); }
-
-class TqMetaUpdateAppendTest : public ::testing::Test {
- protected:
-  void SetUp() override {
-    taosRemoveDir(pathName);
-    pMeta = tqStoreOpen(pathName, FooSerializer, FooDeserializer, FooDeleter, TQ_UPDATE_APPEND);
-    ASSERT(pMeta);
-  }
-
-  void TearDown() override { tqStoreClose(pMeta); }
-
-  STqMetaStore* pMeta;
-  const char*   pathName = TD_TMP_DIR_PATH "tq_test";
-};
-
-TEST_F(TqMetaUpdateAppendTest, copyPutTest) {
-  Foo foo;
-  foo.a = 3;
-  tqHandleCopyPut(pMeta, 1, &foo, sizeof(Foo));
-
-  Foo* pFoo = (Foo*)tqHandleGet(pMeta, 1);
-  EXPECT_EQ(pFoo == NULL, true);
-
-  tqHandleCommit(pMeta, 1);
-  pFoo = (Foo*)tqHandleGet(pMeta, 1);
-  EXPECT_EQ(pFoo->a, 3);
-}
-
-TEST_F(TqMetaUpdateAppendTest, persistTest) {
-  Foo* pFoo = (Foo*)taosMemoryMalloc(sizeof(Foo));
-  pFoo->a = 2;
-  tqHandleMovePut(pMeta, 1, pFoo);
-  Foo* pBar = (Foo*)tqHandleGet(pMeta, 1);
-  EXPECT_EQ(pBar == NULL, true);
-  tqHandleCommit(pMeta, 1);
-  pBar = (Foo*)tqHandleGet(pMeta, 1);
-  EXPECT_EQ(pBar->a, pFoo->a);
-  pBar = (Foo*)tqHandleGet(pMeta, 2);
-  EXPECT_EQ(pBar == NULL, true);
-
-  tqStoreClose(pMeta);
-  pMeta = tqStoreOpen(pathName, FooSerializer, FooDeserializer, FooDeleter, TQ_UPDATE_APPEND);
-  ASSERT(pMeta);
-
-  pBar = (Foo*)tqHandleGet(pMeta, 1);
-  ASSERT_EQ(pBar != NULL, true);
-  EXPECT_EQ(pBar->a, 2);
-
-  pBar = (Foo*)tqHandleGet(pMeta, 2);
-  EXPECT_EQ(pBar == NULL, true);
-}
-
-TEST_F(TqMetaUpdateAppendTest, uncommittedTest) {
-  Foo* pFoo = (Foo*)taosMemoryMalloc(sizeof(Foo));
-  pFoo->a = 3;
-  tqHandleMovePut(pMeta, 1, pFoo);
-
-  pFoo = (Foo*)tqHandleGet(pMeta, 1);
-  EXPECT_EQ(pFoo == NULL, true);
-}
-
-TEST_F(TqMetaUpdateAppendTest, abortTest) {
-  Foo* pFoo = (Foo*)taosMemoryMalloc(sizeof(Foo));
-  pFoo->a = 3;
-  tqHandleMovePut(pMeta, 1, pFoo);
-
-  pFoo = (Foo*)tqHandleGet(pMeta, 1);
-  EXPECT_EQ(pFoo == NULL, true);
-
-  tqHandleAbort(pMeta, 1);
-  pFoo = (Foo*)tqHandleGet(pMeta, 1);
-  EXPECT_EQ(pFoo == NULL, true);
-}
-
-TEST_F(TqMetaUpdateAppendTest, deleteTest) {
-  Foo* pFoo = (Foo*)taosMemoryMalloc(sizeof(Foo));
-  pFoo->a = 3;
-  tqHandleMovePut(pMeta, 1, pFoo);
-
-  pFoo = (Foo*)tqHandleGet(pMeta, 1);
-  EXPECT_EQ(pFoo == NULL, true);
-
-  tqHandleCommit(pMeta, 1);
-
-  pFoo = (Foo*)tqHandleGet(pMeta, 1);
-  ASSERT_EQ(pFoo != NULL, true);
-  EXPECT_EQ(pFoo->a, 3);
-
-  tqHandleDel(pMeta, 1);
-  pFoo = (Foo*)tqHandleGet(pMeta, 1);
-  ASSERT_EQ(pFoo != NULL, true);
-  EXPECT_EQ(pFoo->a, 3);
-
-  tqHandleCommit(pMeta, 1);
-  pFoo = (Foo*)tqHandleGet(pMeta, 1);
-  EXPECT_EQ(pFoo == NULL, true);
-
-  tqStoreClose(pMeta);
-  pMeta = tqStoreOpen(pathName, FooSerializer, FooDeserializer, FooDeleter, TQ_UPDATE_APPEND);
-  ASSERT(pMeta);
-
-  pFoo = (Foo*)tqHandleGet(pMeta, 1);
-  EXPECT_EQ(pFoo == NULL, true);
-}
-
-TEST_F(TqMetaUpdateAppendTest, intxnPersist) {
-  Foo* pFoo = (Foo*)taosMemoryMalloc(sizeof(Foo));
-  pFoo->a = 3;
-  tqHandleMovePut(pMeta, 1, pFoo);
-  tqHandleCommit(pMeta, 1);
-
-  Foo* pBar = (Foo*)taosMemoryMalloc(sizeof(Foo));
-  pBar->a = 4;
-  tqHandleMovePut(pMeta, 1, pBar);
-
-  Foo* pFoo1 = (Foo*)tqHandleGet(pMeta, 1);
-  EXPECT_EQ(pFoo1->a, 3);
-
-  tqStoreClose(pMeta);
-  pMeta = tqStoreOpen(pathName, FooSerializer, FooDeserializer, FooDeleter, TQ_UPDATE_APPEND);
-  ASSERT(pMeta);
-
-  pFoo1 = (Foo*)tqHandleGet(pMeta, 1);
-  EXPECT_EQ(pFoo1->a, 3);
-
-  tqHandleCommit(pMeta, 1);
-
-  pFoo1 = (Foo*)tqHandleGet(pMeta, 1);
-  EXPECT_EQ(pFoo1->a, 4);
-
-  tqStoreClose(pMeta);
-  pMeta = tqStoreOpen(pathName, FooSerializer, FooDeserializer, FooDeleter, TQ_UPDATE_APPEND);
-  ASSERT(pMeta);
-
-  pFoo1 = (Foo*)tqHandleGet(pMeta, 1);
-  EXPECT_EQ(pFoo1->a, 4);
-}
-
-TEST_F(TqMetaUpdateAppendTest, multiplePage) {
-  taosSeedRand(0);
-  std::vector<int> v;
-  for (int i = 0; i < 1000; i++) {
-    v.push_back(taosRand());
-    Foo foo;
-    foo.a = v[i];
-    tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
-  }
-  for (int i = 0; i < 500; i++) {
-    tqHandleCommit(pMeta, i);
-    Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
-    ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
-    EXPECT_EQ(pFoo->a, v[i]);
-  }
-
-  tqStoreClose(pMeta);
-  pMeta = tqStoreOpen(pathName, FooSerializer, FooDeserializer, FooDeleter, TQ_UPDATE_APPEND);
-  ASSERT(pMeta);
-
-  for (int i = 500; i < 1000; i++) {
-    tqHandleCommit(pMeta, i);
-    Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
-    ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
-    EXPECT_EQ(pFoo->a, v[i]);
-  }
-
-  for (int i = 0; i < 1000; i++) {
-    Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
-    ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
-    EXPECT_EQ(pFoo->a, v[i]);
-  }
-}
-
-TEST_F(TqMetaUpdateAppendTest, multipleRewrite) {
-  taosSeedRand(0);
-  std::vector<int> v;
-  for (int i = 0; i < 1000; i++) {
-    v.push_back(taosRand());
-    Foo foo;
-    foo.a = v[i];
-    tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
-  }
-
-  for (int i = 0; i < 500; i++) {
-    tqHandleCommit(pMeta, i);
-    v[i] = taosRand();
-    Foo foo;
-    foo.a = v[i];
-    tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
-  }
-
-  for (int i = 500; i < 1000; i++) {
-    v[i] = taosRand();
-    Foo foo;
-    foo.a = v[i];
-    tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
-  }
-
-  for (int i = 0; i < 1000; i++) {
-    tqHandleCommit(pMeta, i);
-  }
-
-  tqStoreClose(pMeta);
-  pMeta = tqStoreOpen(pathName, FooSerializer, FooDeserializer, FooDeleter, TQ_UPDATE_APPEND);
-  ASSERT(pMeta);
-
-  for (int i = 500; i < 1000; i++) {
-    v[i] = taosRand();
-    Foo foo;
-    foo.a = v[i];
-    tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
-    tqHandleCommit(pMeta, i);
-  }
-
-  for (int i = 0; i < 1000; i++) {
-    Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
-    ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
-    EXPECT_EQ(pFoo->a, v[i]);
-  }
-}
-
-TEST_F(TqMetaUpdateAppendTest, dupCommit) {
-  taosSeedRand(0);
-  std::vector<int> v;
-  for (int i = 0; i < 1000; i++) {
-    v.push_back(taosRand());
-    Foo foo;
-    foo.a = v[i];
-    tqHandleCopyPut(pMeta, i, &foo, sizeof(Foo));
-  }
-
-  for (int i = 0; i < 1000; i++) {
-    int ret = tqHandleCommit(pMeta, i);
-    EXPECT_EQ(ret, 0);
-    ret = tqHandleCommit(pMeta, i);
-    EXPECT_EQ(ret, -1);
-  }
-
-  for (int i = 0; i < 1000; i++) {
-    int ret = tqHandleCommit(pMeta, i);
-    EXPECT_EQ(ret, -1);
-  }
-
-  for (int i = 0; i < 1000; i++) {
-    Foo* pFoo = (Foo*)tqHandleGet(pMeta, i);
-    ASSERT_EQ(pFoo != NULL, true) << " at idx " << i << "\n";
-    EXPECT_EQ(pFoo->a, v[i]);
-  }
-}

source/libs/executor/inc/indexoperator.h

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#ifndef _INDEX_OPERATOR_H
-#define _INDEX_OPERATOR_H
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-#include "nodes.h"
-#include "tglobal.h"
-
-typedef enum { SFLT_NOT_INDEX, SFLT_COARSE_INDEX, SFLT_ACCURATE_INDEX } SIdxFltStatus;
-
-SIdxFltStatus idxGetFltStatus(SNode *pFilterNode);
-// construct tag filter operator later
-int32_t doFilterTag(const SNode *pFilterNode, SArray *result);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /*INDEX_OPERATOR_*/

source/libs/function/src/builtins.c

@@ -14,8 +14,8 @@
  */
 
 #include "builtins.h"
-#include "querynodes.h"
 #include "builtinsimpl.h"
+#include "querynodes.h"
 #include "scalar.h"
 #include "taoserror.h"
 #include "tdatablock.h"
@@ -185,6 +185,19 @@ static int32_t translateApercentile(SFunctionNode* pFunc, char* pErrBuf, int32_t
   if (!IS_NUMERIC_TYPE(para1Type) || !IS_INTEGER_TYPE(para2Type)) {
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }
+
+  SNode* pParamNode = nodesListGetNode(pFunc->pParameterList, 1);
+  if (nodeType(pParamNode) != QUERY_NODE_VALUE) {
+    return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
+  }
+  
+  SValueNode* pValue = (SValueNode*)pParamNode;
+  if (pValue->datum.i < 0 || pValue->datum.i > 100) {
+    return invaildFuncParaValueErrMsg(pErrBuf, len, pFunc->functionName);
+  }
+
+  pValue->notReserved = true;
+  
   if (3 == paraNum) {
     SNode* pPara3 = nodesListGetNode(pFunc->pParameterList, 2);
     if (QUERY_NODE_VALUE != nodeType(pPara3) || !validAperventileAlgo((SValueNode*)pPara3)) {
@@ -215,7 +228,7 @@ static int32_t translateTop(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
-  SValueNode* pValue = (SValueNode*) pParamNode;
+  SValueNode* pValue = (SValueNode*)pParamNode;
   if (pValue->node.resType.type != TSDB_DATA_TYPE_BIGINT) {
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }
@@ -224,6 +237,8 @@ static int32_t translateTop(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
     return invaildFuncParaValueErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
+  pValue->notReserved = true;
+
   SDataType* pType = &((SExprNode*)nodesListGetNode(pFunc->pParameterList, 0))->resType;
   pFunc->node.resType = (SDataType){.bytes = pType->bytes, .type = pType->type};
   return TSDB_CODE_SUCCESS;
@@ -336,7 +351,7 @@ static int32_t translateStateCount(SFunctionNode* pFunc, char* pErrBuf, int32_t
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
-  pFunc->node.resType = (SDataType) { .bytes = tDataTypes[TSDB_DATA_TYPE_BIGINT].bytes, .type = TSDB_DATA_TYPE_BIGINT };
+  pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_BIGINT].bytes, .type = TSDB_DATA_TYPE_BIGINT};
   return TSDB_CODE_SUCCESS;
 }
 
@@ -361,7 +376,7 @@ static int32_t translateStateDuration(SFunctionNode* pFunc, char* pErrBuf, int32
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
-  pFunc->node.resType = (SDataType) { .bytes = tDataTypes[TSDB_DATA_TYPE_BIGINT].bytes, .type = TSDB_DATA_TYPE_BIGINT };
+  pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_BIGINT].bytes, .type = TSDB_DATA_TYPE_BIGINT};
   return TSDB_CODE_SUCCESS;
 }
 
@@ -392,7 +407,7 @@ static int32_t translateCsum(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
     }
   }
 
-  pFunc->node.resType = (SDataType) { .bytes = tDataTypes[resType].bytes, .type = resType};
+  pFunc->node.resType = (SDataType){.bytes = tDataTypes[resType].bytes, .type = resType};
   return TSDB_CODE_SUCCESS;
 }
 
@@ -500,8 +515,7 @@ static int32_t translateUnique(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
 
   SNode* pPara = nodesListGetNode(pFunc->pParameterList, 0);
   if (QUERY_NODE_COLUMN != nodeType(pPara)) {
-    return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
-                           "The parameters of UNIQUE can only be columns");
+    return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR, "The parameters of UNIQUE can only be columns");
   }
 
   pFunc->node.resType = ((SExprNode*)pPara)->resType;

source/libs/function/src/tudf.c

@@ -72,12 +72,20 @@ static int32_t udfSpawnUdfd(SUdfdData* pData) {
   char path[PATH_MAX] = {0};
   if (tsProcPath == NULL) {
     path[0] = '.';
+  #ifdef WINDOWS
+    GetModuleFileName(NULL, path, PATH_MAX);
+    taosDirName(path);
+  #endif
   } else {
     strncpy(path, tsProcPath, strlen(tsProcPath));
     taosDirName(path);
   }
 #ifdef WINDOWS
+  if (strlen(path)==0) {
     strcat(path, "udfd.exe");
+  } else {
+    strcat(path, "\\udfd.exe");
+  }
 #else
   strcat(path, "/udfd");
 #endif

source/libs/index/CMakeLists.txt

@@ -12,6 +12,9 @@ target_link_libraries(
     PUBLIC os
     PUBLIC util
     PUBLIC common
+    PUBLIC nodes
+    PUBLIC scalar 
+    PUBLIC function 
 )
 
 if (${BUILD_WITH_LUCENE})

source/libs/executor/src/indexoperator.c → source/libs/index/src/indexFilter.c

@@ -13,10 +13,11 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "indexoperator.h"
-#include "executorimpl.h"
 #include "index.h"
+#include "indexInt.h"
 #include "nodes.h"
+#include "querynodes.h"
+#include "scalar.h"
 #include "tdatablock.h"
 
 // clang-format off
@@ -69,9 +70,9 @@ typedef int32_t (*sif_func_t)(SIFParam *left, SIFParam *rigth, SIFParam *output)
 static sif_func_t sifNullFunc = NULL;
 // typedef struct SIFWalkParm
 // construct tag filter operator later
-static void destroyTagFilterOperatorInfo(void *param) {
-  STagFilterOperatorInfo *pInfo = (STagFilterOperatorInfo *)param;
-}
+// static void destroyTagFilterOperatorInfo(void *param) {
+//  STagFilterOperatorInfo *pInfo = (STagFilterOperatorInfo *)param;
+//}
 
 static void sifFreeParam(SIFParam *param) {
   if (param == NULL) return;
@@ -178,13 +179,13 @@ static int32_t sifInitParam(SNode *node, SIFParam *param, SIFCtx *ctx) {
     case QUERY_NODE_NODE_LIST: {
       SNodeListNode *nl = (SNodeListNode *)node;
       if (LIST_LENGTH(nl->pNodeList) <= 0) {
-        qError("invalid length for node:%p, length: %d", node, LIST_LENGTH(nl->pNodeList));
+        indexError("invalid length for node:%p, length: %d", node, LIST_LENGTH(nl->pNodeList));
         SIF_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
       }
       SIF_ERR_RET(scalarGenerateSetFromList((void **)&param->pFilter, node, nl->dataType.type));
       if (taosHashPut(ctx->pRes, &node, POINTER_BYTES, param, sizeof(*param))) {
         taosHashCleanup(param->pFilter);
-        qError("taosHashPut nodeList failed, size:%d", (int32_t)sizeof(*param));
+        indexError("taosHashPut nodeList failed, size:%d", (int32_t)sizeof(*param));
         SIF_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
       }
       break;
@@ -194,7 +195,7 @@ static int32_t sifInitParam(SNode *node, SIFParam *param, SIFCtx *ctx) {
     case QUERY_NODE_LOGIC_CONDITION: {
       SIFParam *res = (SIFParam *)taosHashGet(ctx->pRes, &node, POINTER_BYTES);
       if (NULL == res) {
-        qError("no result for node, type:%d, node:%p", nodeType(node), node);
+        indexError("no result for node, type:%d, node:%p", nodeType(node), node);
         SIF_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
       }
       *param = *res;
@@ -210,7 +211,7 @@ static int32_t sifInitOperParams(SIFParam **params, SOperatorNode *node, SIFCtx
   int32_t code = 0;
   int32_t nParam = sifGetOperParamNum(node->opType);
   if (NULL == node->pLeft || (nParam == 2 && NULL == node->pRight)) {
-    qError("invalid operation node, left: %p, rigth: %p", node->pLeft, node->pRight);
+    indexError("invalid operation node, left: %p, rigth: %p", node->pLeft, node->pRight);
     SIF_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }
   SIFParam *paramList = taosMemoryCalloc(nParam, sizeof(SIFParam));
@@ -232,7 +233,7 @@ static int32_t sifInitParamList(SIFParam **params, SNodeList *nodeList, SIFCtx *
   int32_t   code = 0;
   SIFParam *tParams = taosMemoryCalloc(nodeList->length, sizeof(SIFParam));
   if (tParams == NULL) {
-    qError("failed to calloc, nodeList: %p", nodeList);
+    indexError("failed to calloc, nodeList: %p", nodeList);
     SIF_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
   }
 
@@ -252,7 +253,7 @@ _return:
   SIF_RET(code);
 }
 static int32_t sifExecFunction(SFunctionNode *node, SIFCtx *ctx, SIFParam *output) {
-  qError("index-filter not support buildin function");
+  indexError("index-filter not support buildin function");
   return TSDB_CODE_QRY_INVALID_INPUT;
 }
 static int32_t sifDoIndex(SIFParam *left, SIFParam *right, int8_t operType, SIFParam *output) {
@@ -390,7 +391,7 @@ _return:
 
 static int32_t sifExecLogic(SLogicConditionNode *node, SIFCtx *ctx, SIFParam *output) {
   if (NULL == node->pParameterList || node->pParameterList->length <= 0) {
-    qError("invalid logic parameter list, list:%p, paramNum:%d", node->pParameterList,
+    indexError("invalid logic parameter list, list:%p, paramNum:%d", node->pParameterList,
                node->pParameterList ? node->pParameterList->length : 0);
     return TSDB_CODE_QRY_INVALID_INPUT;
   }
@@ -485,7 +486,7 @@ EDealRes sifCalcWalker(SNode *node, void *context) {
     return sifWalkOper(node, ctx);
   }
 
-  qError("invalid node type for index filter calculating, type:%d", nodeType(node));
+  indexError("invalid node type for index filter calculating, type:%d", nodeType(node));
   ctx->code = TSDB_CODE_QRY_INVALID_INPUT;
   return DEAL_RES_ERROR;
 }
@@ -509,7 +510,7 @@ static int32_t sifCalculate(SNode *pNode, SIFParam *pDst) {
   SIFCtx  ctx = {.code = 0, .noExec = false};
   ctx.pRes = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
   if (NULL == ctx.pRes) {
-    qError("index-filter failed to taosHashInit");
+    indexError("index-filter failed to taosHashInit");
     return TSDB_CODE_QRY_OUT_OF_MEMORY;
   }
 
@@ -519,7 +520,7 @@ static int32_t sifCalculate(SNode *pNode, SIFParam *pDst) {
   if (pDst) {
     SIFParam *res = (SIFParam *)taosHashGet(ctx.pRes, (void *)&pNode, POINTER_BYTES);
     if (res == NULL) {
-      qError("no valid res in hash, node:(%p), type(%d)", (void *)&pNode, nodeType(pNode));
+      indexError("no valid res in hash, node:(%p), type(%d)", (void *)&pNode, nodeType(pNode));
       SIF_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
     }
     taosArrayAddAll(pDst->result, res->result);
@@ -539,7 +540,7 @@ static int32_t sifGetFltHint(SNode *pNode, SIdxFltStatus *status) {
   SIFCtx ctx = {.code = 0, .noExec = true};
   ctx.pRes = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
   if (NULL == ctx.pRes) {
-    qError("index-filter failed to taosHashInit");
+    indexError("index-filter failed to taosHashInit");
     return TSDB_CODE_QRY_OUT_OF_MEMORY;
   }
 
@@ -549,7 +550,7 @@ static int32_t sifGetFltHint(SNode *pNode, SIdxFltStatus *status) {
 
   SIFParam *res = (SIFParam *)taosHashGet(ctx.pRes, (void *)&pNode, POINTER_BYTES);
   if (res == NULL) {
-    qError("no valid res in hash, node:(%p), type(%d)", (void *)&pNode, nodeType(pNode));
+    indexError("no valid res in hash, node:(%p), type(%d)", (void *)&pNode, nodeType(pNode));
     SIF_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
   }
   *status = res->status;

source/libs/nodes/src/nodesCloneFuncs.c

@@ -19,6 +19,21 @@
 #include "taos.h"
 #include "taoserror.h"
 
+#define COPY_SCALAR_FIELD(fldname)     \
+  do {                                 \
+    (pDst)->fldname = (pSrc)->fldname; \
+  } while (0)
+
+#define COPY_CHAR_ARRAY_FIELD(fldname)        \
+  do {                                        \
+    strcpy((pDst)->fldname, (pSrc)->fldname); \
+  } while (0)
+
+#define COPY_OBJECT_FIELD(fldname, size)                  \
+  do {                                                    \
+    memcpy(&((pDst)->fldname), &((pSrc)->fldname), size); \
+  } while (0)
+
 #define COPY_CHAR_POINT_FIELD(fldname)         \
   do {                                         \
     if (NULL == (pSrc)->fldname) {             \
@@ -70,27 +85,61 @@
     }                                                               \
   } while (0)
 
-static void dataTypeCopy(const SDataType* pSrc, SDataType* pDst) {}
-
 static SNode* exprNodeCopy(const SExprNode* pSrc, SExprNode* pDst) {
-  dataTypeCopy(&pSrc->resType, &pDst->resType);
-  pDst->pAssociation = NULL;
+  COPY_OBJECT_FIELD(resType, sizeof(SDataType));
+  COPY_CHAR_ARRAY_FIELD(aliasName);
+  COPY_CHAR_ARRAY_FIELD(userAlias);
   return (SNode*)pDst;
 }
 
 static SNode* columnNodeCopy(const SColumnNode* pSrc, SColumnNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, exprNodeCopy);
-  pDst->pProjectRef = NULL;
+  COPY_SCALAR_FIELD(tableId);
+  COPY_SCALAR_FIELD(tableType);
+  COPY_SCALAR_FIELD(colId);
+  COPY_SCALAR_FIELD(colType);
+  COPY_CHAR_ARRAY_FIELD(dbName);
+  COPY_CHAR_ARRAY_FIELD(tableName);
+  COPY_CHAR_ARRAY_FIELD(tableAlias);
+  COPY_CHAR_ARRAY_FIELD(colName);
+  COPY_SCALAR_FIELD(dataBlockId);
+  COPY_SCALAR_FIELD(slotId);
   return (SNode*)pDst;
 }
 
 static SNode* valueNodeCopy(const SValueNode* pSrc, SValueNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, exprNodeCopy);
   COPY_CHAR_POINT_FIELD(literal);
+  COPY_SCALAR_FIELD(isDuration);
+  COPY_SCALAR_FIELD(translate);
+  COPY_SCALAR_FIELD(notReserved);
+  COPY_SCALAR_FIELD(placeholderNo);
+  COPY_SCALAR_FIELD(typeData);
+  COPY_SCALAR_FIELD(unit);
   if (!pSrc->translate) {
     return (SNode*)pDst;
   }
   switch (pSrc->node.resType.type) {
+    case TSDB_DATA_TYPE_BOOL:
+      COPY_SCALAR_FIELD(datum.b);
+      break;
+    case TSDB_DATA_TYPE_TINYINT:
+    case TSDB_DATA_TYPE_SMALLINT:
+    case TSDB_DATA_TYPE_INT:
+    case TSDB_DATA_TYPE_BIGINT:
+    case TSDB_DATA_TYPE_TIMESTAMP:
+      COPY_SCALAR_FIELD(datum.i);
+      break;
+    case TSDB_DATA_TYPE_FLOAT:
+    case TSDB_DATA_TYPE_DOUBLE:
+      COPY_SCALAR_FIELD(datum.d);
+      break;
+    case TSDB_DATA_TYPE_UTINYINT:
+    case TSDB_DATA_TYPE_USMALLINT:
+    case TSDB_DATA_TYPE_UINT:
+    case TSDB_DATA_TYPE_UBIGINT:
+      COPY_SCALAR_FIELD(datum.u);
+      break;
     case TSDB_DATA_TYPE_NCHAR:
     case TSDB_DATA_TYPE_VARCHAR:
     case TSDB_DATA_TYPE_VARBINARY:
@@ -104,7 +153,7 @@ static SNode* valueNodeCopy(const SValueNode* pSrc, SValueNode* pDst) {
     case TSDB_DATA_TYPE_JSON:
     case TSDB_DATA_TYPE_DECIMAL:
     case TSDB_DATA_TYPE_BLOB:
-      // todo
+    case TSDB_DATA_TYPE_MEDIUMBLOB:
     default:
       break;
   }
@@ -113,6 +162,7 @@ static SNode* valueNodeCopy(const SValueNode* pSrc, SValueNode* pDst) {
 
 static SNode* operatorNodeCopy(const SOperatorNode* pSrc, SOperatorNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, exprNodeCopy);
+  COPY_SCALAR_FIELD(opType);
   CLONE_NODE_FIELD(pLeft);
   CLONE_NODE_FIELD(pRight);
   return (SNode*)pDst;
@@ -120,18 +170,27 @@ static SNode* operatorNodeCopy(const SOperatorNode* pSrc, SOperatorNode* pDst) {
 
 static SNode* logicConditionNodeCopy(const SLogicConditionNode* pSrc, SLogicConditionNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, exprNodeCopy);
+  COPY_SCALAR_FIELD(condType);
   CLONE_NODE_LIST_FIELD(pParameterList);
   return (SNode*)pDst;
 }
 
 static SNode* functionNodeCopy(const SFunctionNode* pSrc, SFunctionNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, exprNodeCopy);
+  COPY_CHAR_ARRAY_FIELD(functionName);
+  COPY_SCALAR_FIELD(funcId);
+  COPY_SCALAR_FIELD(funcType);
   CLONE_NODE_LIST_FIELD(pParameterList);
+  COPY_SCALAR_FIELD(udfBufSize);
   return (SNode*)pDst;
 }
 
 static SNode* tableNodeCopy(const STableNode* pSrc, STableNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, exprNodeCopy);
+  COPY_CHAR_ARRAY_FIELD(dbName);
+  COPY_CHAR_ARRAY_FIELD(tableName);
+  COPY_CHAR_ARRAY_FIELD(tableAlias);
+  COPY_SCALAR_FIELD(precision);
   return (SNode*)pDst;
 }
 
@@ -159,6 +218,8 @@ static SNode* realTableNodeCopy(const SRealTableNode* pSrc, SRealTableNode* pDst
   COPY_BASE_OBJECT_FIELD(table, tableNodeCopy);
   CLONE_OBJECT_FIELD(pMeta, tableMetaClone);
   CLONE_OBJECT_FIELD(pVgroupList, vgroupsInfoClone);
+  COPY_CHAR_ARRAY_FIELD(qualDbName);
+  COPY_SCALAR_FIELD(ratio);
   return (SNode*)pDst;
 }
 
@@ -170,6 +231,7 @@ static SNode* tempTableNodeCopy(const STempTableNode* pSrc, STempTableNode* pDst
 
 static SNode* joinTableNodeCopy(const SJoinTableNode* pSrc, SJoinTableNode* pDst) {
   COPY_BASE_OBJECT_FIELD(table, tableNodeCopy);
+  COPY_SCALAR_FIELD(joinType);
   CLONE_NODE_FIELD(pLeft);
   CLONE_NODE_FIELD(pRight);
   CLONE_NODE_FIELD(pOnCond);
@@ -177,21 +239,30 @@ static SNode* joinTableNodeCopy(const SJoinTableNode* pSrc, SJoinTableNode* pDst
 }
 
 static SNode* targetNodeCopy(const STargetNode* pSrc, STargetNode* pDst) {
+  COPY_SCALAR_FIELD(dataBlockId);
+  COPY_SCALAR_FIELD(slotId);
   CLONE_NODE_FIELD(pExpr);
   return (SNode*)pDst;
 }
 
 static SNode* groupingSetNodeCopy(const SGroupingSetNode* pSrc, SGroupingSetNode* pDst) {
+  COPY_SCALAR_FIELD(groupingSetType);
   CLONE_NODE_LIST_FIELD(pParameterList);
   return (SNode*)pDst;
 }
 
 static SNode* orderByExprNodeCopy(const SOrderByExprNode* pSrc, SOrderByExprNode* pDst) {
   CLONE_NODE_FIELD(pExpr);
+  COPY_SCALAR_FIELD(order);
+  COPY_SCALAR_FIELD(nullOrder);
   return (SNode*)pDst;
 }
 
-static SNode* limitNodeCopy(const SLimitNode* pSrc, SLimitNode* pDst) { return (SNode*)pDst; }
+static SNode* limitNodeCopy(const SLimitNode* pSrc, SLimitNode* pDst) {
+  COPY_SCALAR_FIELD(limit);
+  COPY_SCALAR_FIELD(offset);
+  return (SNode*)pDst;
+}
 
 static SNode* stateWindowNodeCopy(const SStateWindowNode* pSrc, SStateWindowNode* pDst) {
   CLONE_NODE_FIELD(pCol);
@@ -215,13 +286,16 @@ static SNode* intervalWindowNodeCopy(const SIntervalWindowNode* pSrc, SIntervalW
 }
 
 static SNode* nodeListNodeCopy(const SNodeListNode* pSrc, SNodeListNode* pDst) {
+  COPY_OBJECT_FIELD(dataType, sizeof(SDataType));
   CLONE_NODE_LIST_FIELD(pNodeList);
   return (SNode*)pDst;
 }
 
 static SNode* fillNodeCopy(const SFillNode* pSrc, SFillNode* pDst) {
+  COPY_SCALAR_FIELD(mode);
   CLONE_NODE_FIELD(pValues);
   CLONE_NODE_FIELD(pWStartTs);
+  COPY_OBJECT_FIELD(timeRange, sizeof(STimeWindow));
   return (SNode*)pDst;
 }
 
@@ -229,7 +303,7 @@ static SNode* logicNodeCopy(const SLogicNode* pSrc, SLogicNode* pDst) {
   CLONE_NODE_LIST_FIELD(pTargets);
   CLONE_NODE_FIELD(pConditions);
   CLONE_NODE_LIST_FIELD(pChildren);
-  pDst->pParent = NULL;
+  COPY_SCALAR_FIELD(optimizedFlag);
   return (SNode*)pDst;
 }
 
@@ -239,12 +313,25 @@ static SNode* logicScanCopy(const SScanLogicNode* pSrc, SScanLogicNode* pDst) {
   CLONE_NODE_LIST_FIELD(pScanPseudoCols);
   CLONE_OBJECT_FIELD(pMeta, tableMetaClone);
   CLONE_OBJECT_FIELD(pVgroupList, vgroupsInfoClone);
+  COPY_SCALAR_FIELD(scanType);
+  COPY_OBJECT_FIELD(scanSeq[0], sizeof(uint8_t) * 2);
+  COPY_OBJECT_FIELD(scanRange, sizeof(STimeWindow));
+  COPY_OBJECT_FIELD(tableName, sizeof(SName));
+  COPY_SCALAR_FIELD(showRewrite);
+  COPY_SCALAR_FIELD(ratio);
   CLONE_NODE_LIST_FIELD(pDynamicScanFuncs);
+  COPY_SCALAR_FIELD(dataRequired);
+  COPY_SCALAR_FIELD(interval);
+  COPY_SCALAR_FIELD(offset);
+  COPY_SCALAR_FIELD(sliding);
+  COPY_SCALAR_FIELD(intervalUnit);
+  COPY_SCALAR_FIELD(slidingUnit);
   return (SNode*)pDst;
 }
 
 static SNode* logicJoinCopy(const SJoinLogicNode* pSrc, SJoinLogicNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
+  COPY_SCALAR_FIELD(joinType);
   CLONE_NODE_FIELD(pOnConditions);
   return (SNode*)pDst;
 }
@@ -259,32 +346,50 @@ static SNode* logicAggCopy(const SAggLogicNode* pSrc, SAggLogicNode* pDst) {
 static SNode* logicProjectCopy(const SProjectLogicNode* pSrc, SProjectLogicNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
   CLONE_NODE_LIST_FIELD(pProjections);
+  COPY_CHAR_ARRAY_FIELD(stmtName);
+  COPY_SCALAR_FIELD(limit);
+  COPY_SCALAR_FIELD(offset);
+  COPY_SCALAR_FIELD(slimit);
+  COPY_SCALAR_FIELD(soffset);
   return (SNode*)pDst;
 }
 
 static SNode* logicVnodeModifCopy(const SVnodeModifLogicNode* pSrc, SVnodeModifLogicNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
-  pDst->pDataBlocks = NULL;
-  pDst->pVgDataBlocks = NULL;
+  COPY_SCALAR_FIELD(msgType);
   return (SNode*)pDst;
 }
 
 static SNode* logicExchangeCopy(const SExchangeLogicNode* pSrc, SExchangeLogicNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
+  COPY_SCALAR_FIELD(srcGroupId);
+  COPY_SCALAR_FIELD(precision);
   return (SNode*)pDst;
 }
 
 static SNode* logicWindowCopy(const SWindowLogicNode* pSrc, SWindowLogicNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
+  COPY_SCALAR_FIELD(winType);
   CLONE_NODE_LIST_FIELD(pFuncs);
+  COPY_SCALAR_FIELD(interval);
+  COPY_SCALAR_FIELD(offset);
+  COPY_SCALAR_FIELD(sliding);
+  COPY_SCALAR_FIELD(intervalUnit);
+  COPY_SCALAR_FIELD(slidingUnit);
+  COPY_SCALAR_FIELD(sessionGap);
   CLONE_NODE_FIELD(pTspk);
+  CLONE_NODE_FIELD(pStateExpr);
+  COPY_SCALAR_FIELD(triggerType);
+  COPY_SCALAR_FIELD(watermark);
   return (SNode*)pDst;
 }
 
 static SNode* logicFillCopy(const SFillLogicNode* pSrc, SFillLogicNode* pDst) {
   COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
+  COPY_SCALAR_FIELD(mode);
   CLONE_NODE_FIELD(pWStartTs);
   CLONE_NODE_FIELD(pValues);
+  COPY_OBJECT_FIELD(timeRange, sizeof(STimeWindow));
   return (SNode*)pDst;
 }
 
@@ -301,28 +406,41 @@ static SNode* logicPartitionCopy(const SPartitionLogicNode* pSrc, SPartitionLogi
 }
 
 static SNode* logicSubplanCopy(const SLogicSubplan* pSrc, SLogicSubplan* pDst) {
+  COPY_OBJECT_FIELD(id, sizeof(SSubplanId));
   CLONE_NODE_FIELD(pNode);
-  pDst->pChildren = NULL;
-  pDst->pParents = NULL;
-  pDst->pVgroupList = NULL;
+  COPY_SCALAR_FIELD(subplanType);
+  COPY_SCALAR_FIELD(level);
+  COPY_SCALAR_FIELD(splitFlag);
   return (SNode*)pDst;
 }
 
 static SNode* dataBlockDescCopy(const SDataBlockDescNode* pSrc, SDataBlockDescNode* pDst) {
+  COPY_SCALAR_FIELD(dataBlockId);
   CLONE_NODE_LIST_FIELD(pSlots);
+  COPY_SCALAR_FIELD(totalRowSize);
+  COPY_SCALAR_FIELD(outputRowSize);
+  COPY_SCALAR_FIELD(precision);
   return (SNode*)pDst;
 }
 
 static SNode* slotDescCopy(const SSlotDescNode* pSrc, SSlotDescNode* pDst) {
-  dataTypeCopy(&pSrc->dataType, &pDst->dataType);
+  COPY_SCALAR_FIELD(slotId);
+  COPY_OBJECT_FIELD(dataType, sizeof(SDataType));
+  COPY_SCALAR_FIELD(reserve);
+  COPY_SCALAR_FIELD(output);
+  COPY_SCALAR_FIELD(tag);
   return (SNode*)pDst;
 }
 
 static SNode* downstreamSourceCopy(const SDownstreamSourceNode* pSrc, SDownstreamSourceNode* pDst) {
+  COPY_OBJECT_FIELD(addr, sizeof(SQueryNodeAddr));
+  COPY_SCALAR_FIELD(taskId);
+  COPY_SCALAR_FIELD(schedId);
   return (SNode*)pDst;
 }
 
 static SNode* selectStmtCopy(const SSelectStmt* pSrc, SSelectStmt* pDst) {
+  COPY_SCALAR_FIELD(isDistinct);
   CLONE_NODE_LIST_FIELD(pProjectionList);
   CLONE_NODE_FIELD(pFromTable);
   CLONE_NODE_FIELD(pWhere);
@@ -333,6 +451,12 @@ static SNode* selectStmtCopy(const SSelectStmt* pSrc, SSelectStmt* pDst) {
   CLONE_NODE_LIST_FIELD(pOrderByList);
   CLONE_NODE_FIELD(pLimit);
   CLONE_NODE_FIELD(pLimit);
+  COPY_CHAR_ARRAY_FIELD(stmtName);
+  COPY_SCALAR_FIELD(precision);
+  COPY_SCALAR_FIELD(isEmptyResult);
+  COPY_SCALAR_FIELD(isTimeOrderQuery);
+  COPY_SCALAR_FIELD(hasAggFuncs);
+  COPY_SCALAR_FIELD(hasRepeatScanFuncs);
   return (SNode*)pDst;
 }
 
@@ -345,7 +469,6 @@ SNodeptr nodesCloneNode(const SNodeptr pNode) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return NULL;
   }
-  memcpy(pDst, pNode, nodesNodeSize(nodeType(pNode)));
   switch (nodeType(pNode)) {
     case QUERY_NODE_COLUMN:
       return columnNodeCopy((const SColumnNode*)pNode, (SColumnNode*)pDst);
@@ -387,6 +510,8 @@ SNodeptr nodesCloneNode(const SNodeptr pNode) {
       return slotDescCopy((const SSlotDescNode*)pNode, (SSlotDescNode*)pDst);
     case QUERY_NODE_DOWNSTREAM_SOURCE:
       return downstreamSourceCopy((const SDownstreamSourceNode*)pNode, (SDownstreamSourceNode*)pDst);
+    case QUERY_NODE_LEFT_VALUE:
+      return pDst;
     case QUERY_NODE_SELECT_STMT:
       return selectStmtCopy((const SSelectStmt*)pNode, (SSelectStmt*)pDst);
     case QUERY_NODE_LOGIC_PLAN_SCAN:

source/libs/nodes/src/nodesCodeFuncs.c

@@ -78,6 +78,8 @@ const char* nodesNodeName(ENodeType type) {
       return "TableOptions";
     case QUERY_NODE_INDEX_OPTIONS:
       return "IndexOptions";
+    case QUERY_NODE_LEFT_VALUE:
+      return "LeftValue";
     case QUERY_NODE_SET_OPERATOR:
       return "SetOperator";
     case QUERY_NODE_SELECT_STMT:
@@ -2175,7 +2177,7 @@ static int32_t jsonToDatum(const SJson* pJson, void* pObj) {
         code = TSDB_CODE_OUT_OF_MEMORY;
         break;
       }
-      varDataSetLen(pNode->datum.p, pNode->node.resType.bytes);
+      varDataSetLen(pNode->datum.p, pNode->node.resType.bytes - VARSTR_HEADER_SIZE);
       if (TSDB_DATA_TYPE_NCHAR == pNode->node.resType.type) {
         char* buf = taosMemoryCalloc(1, pNode->node.resType.bytes * 2 + VARSTR_HEADER_SIZE + 1);
         if (NULL == buf) {
@@ -3019,6 +3021,8 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
       break;
     case QUERY_NODE_DOWNSTREAM_SOURCE:
       return downstreamSourceNodeToJson(pObj, pJson);
+    case QUERY_NODE_LEFT_VALUE:
+      return TSDB_CODE_SUCCESS;  // SLeftValueNode has no fields to serialize.
     case QUERY_NODE_SET_OPERATOR:
       return setOperatorToJson(pObj, pJson);
     case QUERY_NODE_SELECT_STMT:
@@ -3130,6 +3134,8 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
       return jsonToSlotDescNode(pJson, pObj);
     case QUERY_NODE_DOWNSTREAM_SOURCE:
       return jsonToDownstreamSourceNode(pJson, pObj);
+    case QUERY_NODE_LEFT_VALUE:
+      return TSDB_CODE_SUCCESS;  // SLeftValueNode has no fields to deserialize.
     case QUERY_NODE_SET_OPERATOR:
       return jsonToSetOperator(pJson, pObj);
     case QUERY_NODE_SELECT_STMT:

source/libs/nodes/src/nodesUtilFuncs.c

@@ -79,6 +79,8 @@ int32_t nodesNodeSize(ENodeType type) {
       return sizeof(SStreamOptions);
     case QUERY_NODE_TOPIC_OPTIONS:
       return sizeof(STopicOptions);
+    case QUERY_NODE_LEFT_VALUE:
+      return sizeof(SLeftValueNode);
     case QUERY_NODE_SET_OPERATOR:
       return sizeof(SSetOperator);
     case QUERY_NODE_SELECT_STMT:

source/libs/parser/src/parTranslater.c

@@ -646,12 +646,13 @@ static EDealRes translateValueImpl(STranslateContext* pCxt, SValueNode* pVal, SD
       }
       case TSDB_DATA_TYPE_VARCHAR:
       case TSDB_DATA_TYPE_VARBINARY: {
-       pVal->datum.p = taosMemoryCalloc(1, targetDt.bytes + VARSTR_HEADER_SIZE + 1);
+        pVal->datum.p = taosMemoryCalloc(1, targetDt.bytes + 1);
         if (NULL == pVal->datum.p) {
           return generateDealNodeErrMsg(pCxt, TSDB_CODE_OUT_OF_MEMORY);
         }
-       varDataSetLen(pVal->datum.p, targetDt.bytes);
-       strncpy(varDataVal(pVal->datum.p), pVal->literal, targetDt.bytes);
+        int32_t len = TMIN(targetDt.bytes - VARSTR_HEADER_SIZE, pVal->node.resType.bytes);
+        varDataSetLen(pVal->datum.p, len);
+        strncpy(varDataVal(pVal->datum.p), pVal->literal, len);
         break;
       }
       case TSDB_DATA_TYPE_TIMESTAMP: {
@@ -662,19 +663,18 @@ static EDealRes translateValueImpl(STranslateContext* pCxt, SValueNode* pVal, SD
         break;
       }
       case TSDB_DATA_TYPE_NCHAR: {
-       int32_t bytes = targetDt.bytes * TSDB_NCHAR_SIZE;
-       pVal->datum.p = taosMemoryCalloc(1, bytes + VARSTR_HEADER_SIZE + 1);
+        pVal->datum.p = taosMemoryCalloc(1, targetDt.bytes + 1);
         if (NULL == pVal->datum.p) {
           return generateDealNodeErrMsg(pCxt, TSDB_CODE_OUT_OF_MEMORY);
           ;
         }
 
-       int32_t output = 0;
-       if (!taosMbsToUcs4(pVal->literal, pVal->node.resType.bytes, (TdUcs4*)varDataVal(pVal->datum.p), bytes,
-                          &output)) {
+        int32_t len = 0;
+        if (!taosMbsToUcs4(pVal->literal, pVal->node.resType.bytes, (TdUcs4*)varDataVal(pVal->datum.p),
+                           targetDt.bytes - VARSTR_HEADER_SIZE, &len)) {
           return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_WRONG_VALUE_TYPE, pVal->literal);
         }
-       varDataSetLen(pVal->datum.p, output);
+        varDataSetLen(pVal->datum.p, len);
         break;
       }
       case TSDB_DATA_TYPE_DECIMAL:
@@ -689,8 +689,20 @@ static EDealRes translateValueImpl(STranslateContext* pCxt, SValueNode* pVal, SD
   return DEAL_RES_CONTINUE;
 }
 
+static int32_t calcTypeBytes(SDataType dt) {
+  if (TSDB_DATA_TYPE_BINARY == dt.type) {
+    return dt.bytes + VARSTR_HEADER_SIZE;
+  } else if (TSDB_DATA_TYPE_NCHAR == dt.type) {
+    return dt.bytes * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;
+  } else {
+    return dt.bytes;
+  }
+}
+
 static EDealRes translateValue(STranslateContext* pCxt, SValueNode* pVal) {
- return translateValueImpl(pCxt, pVal, pVal->node.resType);
+  SDataType dt = pVal->node.resType;
+  dt.bytes = calcTypeBytes(dt);
+  return translateValueImpl(pCxt, pVal, dt);
 }
 
 static bool isMultiResFunc(SNode* pNode) {
@@ -2341,16 +2353,6 @@ static int32_t translateAlterDatabase(STranslateContext* pCxt, SAlterDatabaseStm
  return buildCmdMsg(pCxt, TDMT_MND_ALTER_DB, (FSerializeFunc)tSerializeSAlterDbReq, &alterReq);
 }
 
-static int32_t calcTypeBytes(SDataType dt) {
- if (TSDB_DATA_TYPE_BINARY == dt.type) {
-   return dt.bytes + VARSTR_HEADER_SIZE;
- } else if (TSDB_DATA_TYPE_NCHAR == dt.type) {
-   return dt.bytes * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;
- } else {
-   return dt.bytes;
- }
-}
-
 static int32_t columnDefNodeToField(SNodeList* pList, SArray** pArray) {
  *pArray = taosArrayInit(LIST_LENGTH(pList), sizeof(SField));
  SNode* pNode;
@@ -4089,18 +4091,8 @@ static int32_t createValueFromFunction(STranslateContext* pCxt, SFunctionNode* p
  return scalarCalculateConstants((SNode*)pFunc, (SNode**)pVal);
 }
 
-static int32_t colDataBytesToValueDataBytes(uint8_t type, int32_t bytes) {
- if (TSDB_DATA_TYPE_VARCHAR == type || TSDB_DATA_TYPE_BINARY == type || TSDB_DATA_TYPE_VARBINARY == type) {
-   return bytes - VARSTR_HEADER_SIZE;
- } else if (TSDB_DATA_TYPE_NCHAR == type) {
-   return (bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
- }
- return bytes;
-}
-
 static SDataType schemaToDataType(SSchema* pSchema) {
   SDataType dt = {.type = pSchema->type, .bytes = pSchema->bytes, .precision = 0, .scale = 0};
- dt.bytes = colDataBytesToValueDataBytes(pSchema->type, pSchema->bytes);
   return dt;
 }
 

source/libs/parser/test/parInitialATest.cpp

@@ -204,7 +204,7 @@ TEST_F(ParserInitialATest, alterTable) {
     }
   };
 
-  auto setAlterTagFunc = [&](const char* pTbname, const char* pTagName, const uint8_t* pNewVal, uint32_t bytes) {
+  auto setAlterTagFunc = [&](const char* pTbname, const char* pTagName, uint8_t* pNewVal, uint32_t bytes) {
     memset(&expect, 0, sizeof(SVAlterTbReq));
     expect.tbName = strdup(pTbname);
     expect.action = TSDB_ALTER_TABLE_UPDATE_TAG_VAL;
@@ -215,7 +215,7 @@ TEST_F(ParserInitialATest, alterTable) {
     expect.pTagVal = pNewVal;
   };
 
-  auto setAlterOptionsFunc = [&](const char* pTbname, int32_t ttl, const char* pComment = nullptr) {
+  auto setAlterOptionsFunc = [&](const char* pTbname, int32_t ttl, char* pComment = nullptr) {
     memset(&expect, 0, sizeof(SVAlterTbReq));
     expect.tbName = strdup(pTbname);
     expect.action = TSDB_ALTER_TABLE_UPDATE_OPTIONS;
@@ -240,7 +240,7 @@ TEST_F(ParserInitialATest, alterTable) {
     void*          pBuf = POINTER_SHIFT(pVgData->pData, sizeof(SMsgHead));
     SVAlterTbReq   req = {0};
     SDecoder       coder = {0};
-    tDecoderInit(&coder, (const uint8_t*)pBuf, pVgData->size);
+    tDecoderInit(&coder, (uint8_t*)pBuf, pVgData->size);
     ASSERT_EQ(tDecodeSVAlterTbReq(&coder, &req), TSDB_CODE_SUCCESS);
 
     ASSERT_EQ(std::string(req.tbName), std::string(expect.tbName));
@@ -274,7 +274,7 @@ TEST_F(ParserInitialATest, alterTable) {
   setAlterOptionsFunc("t1", 10, nullptr);
   run("ALTER TABLE t1 TTL 10");
 
-  setAlterOptionsFunc("t1", -1, "test");
+  setAlterOptionsFunc("t1", -1, (char*)"test");
   run("ALTER TABLE t1 COMMENT 'test'");
 
   setAlterColFunc("t1", TSDB_ALTER_TABLE_ADD_COLUMN, "cc1", TSDB_DATA_TYPE_BIGINT);
@@ -290,7 +290,7 @@ TEST_F(ParserInitialATest, alterTable) {
   run("ALTER TABLE t1 RENAME COLUMN c1 cc1");
 
   int32_t val = 10;
-  setAlterTagFunc("st1s1", "tag1", (const uint8_t*)&val, sizeof(val));
+  setAlterTagFunc("st1s1", "tag1", (uint8_t*)&val, sizeof(val));
   run("ALTER TABLE st1s1 SET TAG tag1=10");
 
   // todo

source/libs/planner/src/planPhysiCreater.c

@@ -599,14 +599,17 @@ typedef struct SRewritePrecalcExprsCxt {
 static EDealRes collectAndRewrite(SRewritePrecalcExprsCxt* pCxt, SNode** pNode) {
   SNode* pExpr = nodesCloneNode(*pNode);
   if (NULL == pExpr) {
+    pCxt->errCode = TSDB_CODE_OUT_OF_MEMORY;
     return DEAL_RES_ERROR;
   }
   if (nodesListAppend(pCxt->pPrecalcExprs, pExpr)) {
+    pCxt->errCode = TSDB_CODE_OUT_OF_MEMORY;
     nodesDestroyNode(pExpr);
     return DEAL_RES_ERROR;
   }
   SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
   if (NULL == pCol) {
+    pCxt->errCode = TSDB_CODE_OUT_OF_MEMORY;
     nodesDestroyNode(pExpr);
     return DEAL_RES_ERROR;
   }
@@ -624,16 +627,45 @@ static EDealRes collectAndRewrite(SRewritePrecalcExprsCxt* pCxt, SNode** pNode)
   return DEAL_RES_IGNORE_CHILD;
 }
 
+static int32_t rewriteValueToOperator(SRewritePrecalcExprsCxt* pCxt, SNode** pNode) {
+  SOperatorNode* pOper = (SOperatorNode*)nodesMakeNode(QUERY_NODE_OPERATOR);
+  if (NULL == pOper) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  pOper->pLeft = nodesMakeNode(QUERY_NODE_LEFT_VALUE);
+  if (NULL == pOper->pLeft) {
+    nodesDestroyNode(pOper);
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  SValueNode* pVal = (SValueNode*)*pNode;
+  pOper->node.resType = pVal->node.resType;
+  strcpy(pOper->node.aliasName, pVal->node.aliasName);
+  pOper->opType = OP_TYPE_ASSIGN;
+  pOper->pRight = *pNode;
+  *pNode = (SNode*)pOper;
+  return TSDB_CODE_SUCCESS;
+}
+
 static EDealRes doRewritePrecalcExprs(SNode** pNode, void* pContext) {
   SRewritePrecalcExprsCxt* pCxt = (SRewritePrecalcExprsCxt*)pContext;
   switch (nodeType(*pNode)) {
+    case QUERY_NODE_VALUE: {
+      if (((SValueNode*)*pNode)->notReserved) {
+        break;
+      }
+      pCxt->errCode = rewriteValueToOperator(pCxt, pNode);
+      if (TSDB_CODE_SUCCESS != pCxt->errCode) {
+        return DEAL_RES_ERROR;
+      }
+      return collectAndRewrite(pCxt, pNode);
+    }
     case QUERY_NODE_OPERATOR:
     case QUERY_NODE_LOGIC_CONDITION: {
-      return collectAndRewrite(pContext, pNode);
+      return collectAndRewrite(pCxt, pNode);
     }
     case QUERY_NODE_FUNCTION: {
       if (fmIsScalarFunc(((SFunctionNode*)(*pNode))->funcId)) {
-        return collectAndRewrite(pContext, pNode);
+        return collectAndRewrite(pCxt, pNode);
       }
     }
     default:
@@ -677,9 +709,8 @@ static int32_t rewritePrecalcExprs(SPhysiPlanContext* pCxt, SNodeList* pList, SN
   }
   SRewritePrecalcExprsCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pPrecalcExprs = *pPrecalcExprs};
   nodesRewriteExprs(*pRewrittenList, doRewritePrecalcExprs, &cxt);
-  if (0 == LIST_LENGTH(cxt.pPrecalcExprs)) {
-    nodesDestroyList(cxt.pPrecalcExprs);
-    *pPrecalcExprs = NULL;
+  if (0 == LIST_LENGTH(cxt.pPrecalcExprs) || TSDB_CODE_SUCCESS != cxt.errCode) {
+    DESTORY_LIST(*pPrecalcExprs);
   }
   return cxt.errCode;
 }

source/libs/planner/test/planBasicTest.cpp

@@ -50,4 +50,6 @@ TEST_F(PlanBasicTest, func) {
   run("SELECT DIFF(c1) FROM t1");
 
   run("SELECT PERCENTILE(c1, 60) FROM t1");
+
+  run("SELECT TOP(c1, 60) FROM t1");
 }

source/libs/planner/test/planGroupByTest.cpp

@@ -49,6 +49,8 @@ TEST_F(PlanGroupByTest, aggFunc) {
   run("SELECT LAST(*), FIRST(*) FROM t1");
 
   run("SELECT LAST(*), FIRST(*) FROM t1 GROUP BY c1");
+
+  run("SELECT SUM(10), COUNT(c1) FROM t1 GROUP BY c2");
 }
 
 TEST_F(PlanGroupByTest, selectFunc) {

source/libs/planner/test/planTestUtil.cpp

@@ -14,6 +14,7 @@
  */
 
 #include "planTestUtil.h"
+#include <getopt.h>
 
 #include <algorithm>
 #include <array>

source/libs/scalar/src/scalar.c

@@ -182,6 +182,11 @@ int32_t sclCopyValueNodeValue(SValueNode *pNode, void **res) {
 
 int32_t sclInitParam(SNode* node, SScalarParam *param, SScalarCtx *ctx, int32_t *rowNum) {
   switch (nodeType(node)) {
+    case QUERY_NODE_LEFT_VALUE: {
+      SSDataBlock* pb = taosArrayGetP(ctx->pBlockList, 0);
+      param->numOfRows = pb->info.rows;
+      break;
+    }
     case QUERY_NODE_VALUE: {
       SValueNode *valueNode = (SValueNode *)node;
 
@@ -845,7 +850,7 @@ EDealRes sclWalkTarget(SNode* pNode, SScalarCtx *ctx) {
 }
 
 EDealRes sclCalcWalker(SNode* pNode, void* pContext) {
-  if (QUERY_NODE_VALUE == nodeType(pNode) || QUERY_NODE_NODE_LIST == nodeType(pNode) || QUERY_NODE_COLUMN == nodeType(pNode)) {
+  if (QUERY_NODE_VALUE == nodeType(pNode) || QUERY_NODE_NODE_LIST == nodeType(pNode) || QUERY_NODE_COLUMN == nodeType(pNode)|| QUERY_NODE_LEFT_VALUE == nodeType(pNode)) {
     return DEAL_RES_CONTINUE;
   }
 

source/libs/scalar/src/sclfunc.c

@@ -824,7 +824,7 @@ int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutp
         }
         //for constant conversion, need to set proper length of pOutput description
         if (len < outputLen) {
-          pOutput->columnData->info.bytes = len;
+          pOutput->columnData->info.bytes = len + VARSTR_HEADER_SIZE;
         }
         break;
       }

source/libs/scalar/src/sclvector.c

@@ -1333,6 +1333,22 @@ void vectorMathMinus(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pO
   doReleaseVec(pLeftCol,  leftConvert);
 }
 
+void vectorAssign(SScalarParam* pLeft, SScalarParam* pRight, SScalarParam *pOut, int32_t _ord) {
+  SColumnInfoData *pOutputCol = pOut->columnData;
+
+  pOut->numOfRows = pLeft->numOfRows;
+
+  if (colDataIsNull_s(pRight->columnData, 0)) {
+    for (int32_t i = 0; i < pOut->numOfRows; ++i) {
+      colDataAppend(pOutputCol, i, NULL, true);
+    }
+  } else {
+    for (int32_t i = 0; i < pOut->numOfRows; ++i) {
+      colDataAppend(pOutputCol, i, colDataGetData(pRight->columnData, 0), false);
+    }
+  }
+}
+
 void vectorConcat(SScalarParam* pLeft, SScalarParam* pRight, void *out, int32_t _ord) {
 #if 0
   int32_t len = pLeft->bytes + pRight->bytes;
@@ -1691,6 +1707,8 @@ _bin_scalar_fn_t getBinScalarOperatorFn(int32_t binFunctionId) {
       return vectorMathRemainder;
     case OP_TYPE_MINUS:
       return vectorMathMinus;
+    case OP_TYPE_ASSIGN:
+      return vectorAssign;
     case OP_TYPE_GREATER_THAN:
       return vectorGreater;
     case OP_TYPE_GREATER_EQUAL:

source/libs/scheduler/inc/schedulerInt.h

@@ -132,7 +132,7 @@ typedef struct SSchLevel {
   int32_t         taskSucceed;
   int32_t         taskNum;
   int32_t         taskLaunchedNum;
-  SHashObj       *flowCtrl;      // key is ep, element is SSchFlowControl
+  int32_t         taskDoneNum;
   SArray         *subTasks;      // Element is SQueryTask
 } SSchLevel;
 
@@ -175,11 +175,13 @@ typedef struct SSchJob {
   SArray          *levels;    // starting from 0. SArray<SSchLevel>
   SNodeList       *subPlans;  // subplan pointer copied from DAG, no need to free it in scheduler
 
+  SArray          *dataSrcTasks; // SArray<SQueryTask*>
   int32_t          levelIdx;
   SEpSet           dataSrcEps;
   SHashObj        *execTasks; // executing tasks, key:taskid, value:SQueryTask*
   SHashObj        *succTasks; // succeed tasks, key:taskid, value:SQueryTask*
   SHashObj        *failTasks; // failed tasks, key:taskid, value:SQueryTask*
+  SHashObj        *flowCtrl;  // key is ep, element is SSchFlowControl
 
   SExplainCtx     *explainCtx;
   int8_t           status;  
@@ -200,7 +202,7 @@ typedef struct SSchJob {
 
 extern SSchedulerMgmt schMgmt;
 
-#define SCH_TASK_READY_TO_LUNCH(readyNum, task) ((readyNum) >= taosArrayGetSize((task)->children))
+#define SCH_TASK_READY_FOR_LAUNCH(readyNum, task) ((readyNum) >= taosArrayGetSize((task)->children))
 
 #define SCH_TASK_ID(_task) ((_task) ? (_task)->taskId : -1)
 #define SCH_SET_TASK_LASTMSG_TYPE(_task, _type) do { if(_task) { atomic_store_32(&(_task)->lastMsgType, _type); } } while (0)
@@ -223,7 +225,7 @@ extern SSchedulerMgmt schMgmt;
 
 #define SCH_SET_JOB_NEED_FLOW_CTRL(_job) (_job)->attr.needFlowCtrl = true
 #define SCH_JOB_NEED_FLOW_CTRL(_job) ((_job)->attr.needFlowCtrl)
-#define SCH_TASK_NEED_FLOW_CTRL(_job, _task) (SCH_IS_DATA_SRC_QRY_TASK(_task) && SCH_JOB_NEED_FLOW_CTRL(_job) && SCH_IS_LEAF_TASK(_job, _task) && SCH_IS_LEVEL_UNFINISHED((_task)->level))
+#define SCH_TASK_NEED_FLOW_CTRL(_job, _task) (SCH_IS_DATA_SRC_QRY_TASK(_task) && SCH_JOB_NEED_FLOW_CTRL(_job) && SCH_IS_LEVEL_UNFINISHED((_task)->level))
 
 #define SCH_SET_JOB_TYPE(_job, type) (_job)->attr.queryJob = ((type) != SUBPLAN_TYPE_MODIFY)
 #define SCH_IS_QUERY_JOB(_job) ((_job)->attr.queryJob) 
@@ -261,7 +263,7 @@ int32_t schLaunchTask(SSchJob *job, SSchTask *task);
 int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, SQueryNodeAddr *addr, int32_t msgType);
 SSchJob *schAcquireJob(int64_t refId);
 int32_t schReleaseJob(int64_t refId);
-void schFreeFlowCtrl(SSchLevel *pLevel);
+void schFreeFlowCtrl(SSchJob *pJob);
 int32_t schCheckJobNeedFlowCtrl(SSchJob *pJob, SSchLevel *pLevel);
 int32_t schDecTaskFlowQuota(SSchJob *pJob, SSchTask *pTask);
 int32_t schCheckIncTaskFlowQuota(SSchJob *pJob, SSchTask *pTask, bool *enough);

source/libs/scheduler/src/schFlowCtrl.c

@@ -19,13 +19,13 @@
 #include "catalog.h"
 #include "tref.h"
 
-void schFreeFlowCtrl(SSchLevel *pLevel) {
-  if (NULL == pLevel->flowCtrl) {
+void schFreeFlowCtrl(SSchJob *pJob) {
+  if (NULL == pJob->flowCtrl) {
     return;
   }
 
   SSchFlowControl *ctrl = NULL;
-  void *pIter = taosHashIterate(pLevel->flowCtrl, NULL);
+  void *pIter = taosHashIterate(pJob->flowCtrl, NULL);
   while (pIter) {
     ctrl = (SSchFlowControl *)pIter;
 
@@ -33,11 +33,11 @@ void schFreeFlowCtrl(SSchLevel *pLevel) {
       taosArrayDestroy(ctrl->taskList);
     }
     
-    pIter = taosHashIterate(pLevel->flowCtrl, pIter);
+    pIter = taosHashIterate(pJob->flowCtrl, pIter);
   }
 
-  taosHashCleanup(pLevel->flowCtrl);
-  pLevel->flowCtrl = NULL;
+  taosHashCleanup(pJob->flowCtrl);
+  pJob->flowCtrl = NULL;
 }
 
 int32_t schCheckJobNeedFlowCtrl(SSchJob *pJob, SSchLevel *pLevel) {
@@ -47,9 +47,9 @@ int32_t schCheckJobNeedFlowCtrl(SSchJob *pJob, SSchLevel *pLevel) {
   }
 
   int32_t sum = 0;
-  
-  for (int32_t i = 0; i < pLevel->taskNum; ++i) {
-    SSchTask *pTask = taosArrayGet(pLevel->subTasks, i);
+  int32_t taskNum = taosArrayGetSize(pJob->dataSrcTasks);
+  for (int32_t i = 0; i < taskNum; ++i) {
+    SSchTask *pTask = *(SSchTask **)taosArrayGet(pJob->dataSrcTasks, i);
 
     sum += pTask->plan->execNodeStat.tableNum;
   }
@@ -59,9 +59,9 @@ int32_t schCheckJobNeedFlowCtrl(SSchJob *pJob, SSchLevel *pLevel) {
     return TSDB_CODE_SUCCESS;
   }
 
-  pLevel->flowCtrl = taosHashInit(pLevel->taskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
-  if (NULL == pLevel->flowCtrl) {
-    SCH_JOB_ELOG("taosHashInit %d flowCtrl failed", pLevel->taskNum);
+  pJob->flowCtrl = taosHashInit(pJob->taskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
+  if (NULL == pJob->flowCtrl) {
+    SCH_JOB_ELOG("taosHashInit %d flowCtrl failed", pJob->taskNum);
     SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
   }
 
@@ -78,7 +78,7 @@ int32_t schDecTaskFlowQuota(SSchJob *pJob, SSchTask *pTask) {
   int32_t code = 0;
   SEp *ep = SCH_GET_CUR_EP(&pTask->plan->execNode);
   
-  ctrl = (SSchFlowControl *)taosHashGet(pLevel->flowCtrl, ep, sizeof(SEp));
+  ctrl = (SSchFlowControl *)taosHashGet(pJob->flowCtrl, ep, sizeof(SEp));
   if (NULL == ctrl) {
     SCH_TASK_ELOG("taosHashGet node from flowCtrl failed, fqdn:%s, port:%d", ep->fqdn, ep->port);
     SCH_ERR_RET(TSDB_CODE_SCH_INTERNAL_ERROR);
@@ -110,11 +110,11 @@ int32_t schCheckIncTaskFlowQuota(SSchJob *pJob, SSchTask *pTask, bool *enough) {
   SEp *ep = SCH_GET_CUR_EP(&pTask->plan->execNode);
   
   do {
-    ctrl = (SSchFlowControl *)taosHashGet(pLevel->flowCtrl, ep, sizeof(SEp));
+    ctrl = (SSchFlowControl *)taosHashGet(pJob->flowCtrl, ep, sizeof(SEp));
     if (NULL == ctrl) {
       SSchFlowControl nctrl = {.tableNumSum = pTask->plan->execNodeStat.tableNum, .execTaskNum = 1};
 
-      code = taosHashPut(pLevel->flowCtrl, ep, sizeof(SEp), &nctrl, sizeof(nctrl));
+      code = taosHashPut(pJob->flowCtrl, ep, sizeof(SEp), &nctrl, sizeof(nctrl));
       if (code) {
         if (HASH_NODE_EXIST(code)) {
           continue;
@@ -273,10 +273,9 @@ int32_t schLaunchTasksInFlowCtrlList(SSchJob *pJob, SSchTask *pTask) {
 
   SCH_ERR_RET(schDecTaskFlowQuota(pJob, pTask));
 
-  SSchLevel *pLevel = pTask->level;
   SEp *ep = SCH_GET_CUR_EP(&pTask->plan->execNode);
   
-  SSchFlowControl *ctrl = (SSchFlowControl *)taosHashGet(pLevel->flowCtrl, ep, sizeof(SEp));
+  SSchFlowControl *ctrl = (SSchFlowControl *)taosHashGet(pJob->flowCtrl, ep, sizeof(SEp));
   if (NULL == ctrl) {
     SCH_TASK_ELOG("taosHashGet node from flowCtrl failed, fqdn:%s, port:%d", ep->fqdn, ep->port);
     SCH_ERR_RET(TSDB_CODE_SCH_INTERNAL_ERROR);

source/libs/scheduler/src/scheduler.c

@@ -391,6 +391,8 @@ int32_t schBuildTaskRalation(SSchJob *pJob, SHashObj *planToTask) {
           SCH_TASK_ELOG("taosArrayPush childTask failed, level:%d, taskIdx:%d, childIdx:%d", i, m, n);
           SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
         }
+
+        SCH_TASK_DLOG("children info, the %d child TID %" PRIx64, n, (*childTask)->taskId);
       }
 
       if (parentNum > 0) {
@@ -423,6 +425,8 @@ int32_t schBuildTaskRalation(SSchJob *pJob, SHashObj *planToTask) {
           SCH_TASK_ELOG("taosArrayPush parentTask failed, level:%d, taskIdx:%d, childIdx:%d", i, m, n);
           SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
         }
+
+        SCH_TASK_DLOG("parents info, the %d parent TID %" PRIx64, n, (*parentTask)->taskId);        
       }
 
       SCH_TASK_DLOG("level:%d, parentNum:%d, childNum:%d", i, parentNum, childNum);
@@ -464,6 +468,17 @@ int32_t schRecordTaskExecNode(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *ad
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t schRecordQueryDataSrc(SSchJob *pJob, SSchTask *pTask) {
+  if (!SCH_IS_DATA_SRC_QRY_TASK(pTask)) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  taosArrayPush(pJob->dataSrcTasks, &pTask);
+
+  return TSDB_CODE_SUCCESS;
+}
+
+
 int32_t schValidateAndBuildJob(SQueryPlan *pDag, SSchJob *pJob) {
   int32_t code = 0;
   pJob->queryId = pDag->queryId;
@@ -473,6 +488,11 @@ int32_t schValidateAndBuildJob(SQueryPlan *pDag, SSchJob *pJob) {
     SCH_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }
 
+  pJob->dataSrcTasks = taosArrayInit(pDag->numOfSubplans, POINTER_BYTES);
+  if (NULL == pJob->dataSrcTasks) {
+    SCH_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
+  }
+
   int32_t levelNum = (int32_t)LIST_LENGTH(pDag->pSubplans);
   if (levelNum <= 0) {
     SCH_JOB_ELOG("invalid level num:%d", levelNum);
@@ -551,6 +571,8 @@ int32_t schValidateAndBuildJob(SQueryPlan *pDag, SSchJob *pJob) {
         SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
       }
 
+      SCH_ERR_JRET(schRecordQueryDataSrc(pJob, p));
+
       if (0 != taosHashPut(planToTask, &plan, POINTER_BYTES, &p, POINTER_BYTES)) {
         SCH_TASK_ELOG("taosHashPut to planToTaks failed, taskIdx:%d", n);
         SCH_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
@@ -629,6 +651,17 @@ int32_t schSetTaskCandidateAddrs(SSchJob *pJob, SSchTask *pTask) {
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t schRemoveTaskFromExecList(SSchJob *pJob, SSchTask *pTask) {
+  int32_t code = taosHashRemove(pJob->execTasks, &pTask->taskId, sizeof(pTask->taskId));
+  if (code) {
+    SCH_TASK_ELOG("task failed to rm from execTask list, code:%x", code);
+    SCH_ERR_RET(TSDB_CODE_SCH_INTERNAL_ERROR);
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+
 int32_t schPushTaskToExecList(SSchJob *pJob, SSchTask *pTask) {
   int32_t code = taosHashPut(pJob->execTasks, &pTask->taskId, sizeof(pTask->taskId), &pTask, POINTER_BYTES);
   if (0 != code) {
@@ -774,6 +807,9 @@ int32_t schTaskCheckSetRetry(SSchJob *pJob, SSchTask *pTask, int32_t errCode, bo
 int32_t schHandleTaskRetry(SSchJob *pJob, SSchTask *pTask) {
   atomic_sub_fetch_32(&pTask->level->taskLaunchedNum, 1);
 
+  SCH_ERR_RET(schRemoveTaskFromExecList(pJob, pTask));
+  SCH_SET_TASK_STATUS(pTask, JOB_TASK_STATUS_NOT_START);
+  
   if (SCH_TASK_NEED_FLOW_CTRL(pJob, pTask)) {
     SCH_ERR_RET(schDecTaskFlowQuota(pJob, pTask));
     SCH_ERR_RET(schLaunchTasksInFlowCtrlList(pJob, pTask));
@@ -947,6 +983,32 @@ _return:
   SCH_RET(schProcessOnJobFailure(pJob, errCode));
 }
 
+int32_t schLaunchNextLevelTasks(SSchJob *pJob, SSchTask *pTask) {
+  if (!SCH_IS_QUERY_JOB(pJob)) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SSchLevel *pLevel = pTask->level;
+  int32_t doneNum = atomic_add_fetch_32(&pLevel->taskDoneNum, 1);
+  if (doneNum == pLevel->taskNum) {
+    pJob->levelIdx--;
+
+    pLevel = taosArrayGet(pJob->levels, pJob->levelIdx);
+    for (int32_t i = 0; i < pLevel->taskNum; ++i) {
+      SSchTask *pTask = taosArrayGet(pLevel->subTasks, i);
+
+      if (pTask->children && taosArrayGetSize(pTask->children) > 0) {
+        continue;
+      }
+      
+      SCH_ERR_RET(schLaunchTask(pJob, pTask));
+    }
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+
 // Note: no more task error processing, handled in function internal
 int32_t schProcessOnTaskSuccess(SSchJob *pJob, SSchTask *pTask) {
   bool    moved = false;
@@ -1015,11 +1077,13 @@ int32_t schProcessOnTaskSuccess(SSchJob *pJob, SSchTask *pTask) {
     qSetSubplanExecutionNode(par->plan, pTask->plan->id.groupId, &source);
     SCH_UNLOCK(SCH_WRITE, &par->lock);
 
-    if (SCH_TASK_READY_TO_LUNCH(readyNum, par)) {
-      SCH_ERR_RET(schLaunchTaskImpl(pJob, par));
+    if (SCH_TASK_READY_FOR_LAUNCH(readyNum, par)) {
+      SCH_ERR_RET(schLaunchTask(pJob, par));
     }
   }
 
+  SCH_ERR_RET(schLaunchNextLevelTasks(pJob, pTask));
+
   return TSDB_CODE_SUCCESS;
 
 _return:
@@ -2400,8 +2464,6 @@ void schFreeJobImpl(void *job) {
   for (int32_t i = 0; i < numOfLevels; ++i) {
     SSchLevel *pLevel = taosArrayGet(pJob->levels, i);
 
-    schFreeFlowCtrl(pLevel);
-
     int32_t numOfTasks = taosArrayGetSize(pLevel->subTasks);
     for (int32_t j = 0; j < numOfTasks; ++j) {
       SSchTask *pTask = taosArrayGet(pLevel->subTasks, j);
@@ -2411,12 +2473,15 @@ void schFreeJobImpl(void *job) {
     taosArrayDestroy(pLevel->subTasks);
   }
 
+  schFreeFlowCtrl(pJob);
+
   taosHashCleanup(pJob->execTasks);
   taosHashCleanup(pJob->failTasks);
   taosHashCleanup(pJob->succTasks);
 
   taosArrayDestroy(pJob->levels);
   taosArrayDestroy(pJob->nodeList);
+  taosArrayDestroy(pJob->dataSrcTasks);
 
   qExplainFreeCtx(pJob->explainCtx);
 

source/libs/sync/src/syncMessage.c

@@ -411,7 +411,7 @@ SyncPing* syncPingDeserialize3(void* buf, int32_t bufLen) {
   }
   uint32_t len;
   char*    data = NULL;
-  if (tDecodeBinary(&decoder, (const uint8_t**)(&data), &len) < 0) {
+  if (tDecodeBinary(&decoder, (uint8_t**)(&data), &len) < 0) {
     return NULL;
   }
   assert(len = pMsg->dataLen);
@@ -670,7 +670,7 @@ SyncPingReply* syncPingReplyDeserialize3(void* buf, int32_t bufLen) {
   }
   uint32_t len;
   char*    data = NULL;
-  if (tDecodeBinary(&decoder, (const uint8_t**)(&data), &len) < 0) {
+  if (tDecodeBinary(&decoder, (uint8_t**)(&data), &len) < 0) {
     return NULL;
   }
   assert(len = pMsg->dataLen);

source/os/CMakeLists.txt

@@ -18,14 +18,16 @@ if(USE_TD_MEMORY)
     add_definitions(-DUSE_TD_MEMORY) 
 endif ()
 if(BUILD_ADDR2LINE)
+    if(NOT TD_WINDOWS)
         target_include_directories(
             os
             PUBLIC "${TD_SOURCE_DIR}/contrib/libdwarf/src/lib/libdwarf"
         )
-    add_definitions(-DUSE_ADDR2LINE)
         target_link_libraries(
             os PUBLIC addr2line dl z
         )
+    endif()
+    add_definitions(-DUSE_ADDR2LINE)
 endif ()
 if(CHECK_STR2INT_ERROR)
     add_definitions(-DTD_CHECK_STR_TO_INT_ERROR)

source/os/src/osDir.c

@@ -91,7 +91,12 @@ void taosRemoveDir(const char *dirname) {
 bool taosDirExist(const char *dirname) { return taosCheckExistFile(dirname); }
 
 int32_t taosMkDir(const char *dirname) {
+  if (taosDirExist(dirname)) return 0;
+#ifdef WINDOWS
+  int32_t code = _mkdir(dirname, 0755);
+#else
   int32_t code = mkdir(dirname, 0755);
+#endif
   if (code < 0 && errno == EEXIST) {
     return 0;
   }
@@ -101,36 +106,48 @@ int32_t taosMkDir(const char *dirname) {
 
 int32_t taosMulMkDir(const char *dirname) {
   if (dirname == NULL) return -1;
-  char *  temp = strdup(dirname);
+  char temp[1024];
+#ifdef WINDOWS
+  taosRealPath(dirname, temp, sizeof(temp));
+#else
+  strcpy(temp, dirname);
+#endif
   char *  pos = temp;
   int32_t code = 0;
 
-  if (strncmp(temp, "/", 1) == 0) {
+  if (taosDirExist(temp)) return code;
+
+  if (strncmp(temp, TD_DIRSEP, 1) == 0) {
     pos += 1;
-  } else if (strncmp(temp, "./", 2) == 0) {
+  } else if (strncmp(temp, "." TD_DIRSEP, 2) == 0) {
     pos += 2;
   }
 
   for (; *pos != '\0'; pos++) {
-    if (*pos == '/') {
+    if (*pos == TD_DIRSEP[0]) {
       *pos = '\0';
+    #ifdef WINDOWS
+      code = _mkdir(temp, 0755);
+    #else
       code = mkdir(temp, 0755);
+    #endif
       if (code < 0 && errno != EEXIST) {
-        free(temp);
         return code;
       }
-      *pos = '/';
+      *pos = TD_DIRSEP[0];
     }
   }
 
-  if (*(pos - 1) != '/') {
+  if (*(pos - 1) != TD_DIRSEP[0]) {
+  #ifdef WINDOWS
+    code = _mkdir(temp, 0755);
+  #else
     code = mkdir(temp, 0755);
+  #endif
     if (code < 0 && errno != EEXIST) {
-      free(temp);
       return code;
     }
   }
-  free(temp);
 
   // int32_t code = mkdir(dirname, 0755);
   if (code < 0 && errno == EEXIST) {
@@ -233,8 +250,14 @@ char *taosDirName(char *name) {
   _splitpath(name, Drive1, Dir1, NULL, NULL);
   size_t dirNameLen = strlen(Drive1) + strlen(Dir1);
   if (dirNameLen > 0) {
+    if (name[dirNameLen - 1] == '/' || name[dirNameLen - 1] == '\\') {
+      name[dirNameLen - 1] = 0;
+    } else {
       name[dirNameLen] = 0;
     }
+  } else {
+    name[0] = 0;
+  }
   return name;
 #else
   return dirname(name);

source/util/src/tencode.c

@@ -30,7 +30,7 @@ struct SEncoderNode {
 
 struct SDecoderNode {
   SDecoderNode* pNext;
-  const uint8_t* data;
+  uint8_t*      data;
   uint32_t      size;
   uint32_t      pos;
 };
@@ -52,7 +52,7 @@ void tEncoderClear(SEncoder* pCoder) {
   memset(pCoder, 0, sizeof(*pCoder));
 }
 
-void tDecoderInit(SDecoder* pDecoder, const uint8_t* data, uint32_t size) {
+void tDecoderInit(SDecoder* pDecoder, uint8_t* data, uint32_t size) {
   pDecoder->data = data;
   pDecoder->size = size;
   pDecoder->pos = 0;

taos-tools @ 2c4a1c83

-Subproject commit 772aef458fec13804c654e59c7a248b55ba3807b
+Subproject commit 2c4a1c83322b983881aea93ec2b51e7df826125a