Merge branch 3.0 to FIX/TD-19334-3.0

cmake/cmake.platform

@@ -82,10 +82,12 @@ IF ("${CPUTYPE}" STREQUAL "")
     MESSAGE(STATUS "Current platform is amd64")
     SET(PLATFORM_ARCH_STR "amd64")
     SET(TD_INTEL_64 TRUE)
+    ADD_DEFINITIONS("-D_TD_X86_")
   ELSEIF (CMAKE_SYSTEM_PROCESSOR MATCHES "(x86)|(X86)")
     MESSAGE(STATUS "Current platform is x86")
     SET(PLATFORM_ARCH_STR "i386")
     SET(TD_INTEL_32 TRUE)
+    ADD_DEFINITIONS("-D_TD_X86_")
   ELSEIF (CMAKE_SYSTEM_PROCESSOR MATCHES "armv7l")
     MESSAGE(STATUS "Current platform is aarch32")
     SET(PLATFORM_ARCH_STR "arm")

cmake/taostools_CMakeLists.txt.in

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

docs/zh/02-intro.md

@@ -69,7 +69,7 @@ TDengine 的主要功能如下：
 
 - **[分析能力](https://www.taosdata.com/tdengine/easy_data_analytics)**：通过超级表、存储计算分离、分区分片、预计算和其它技术，TDengine 能够高效地浏览、格式化和访问数据。
 
-- **[核心开源](https://www.taosdata.com/tdengine/open_source_time-series_database)**：TDengine 的核心代码包括集群功能全部在开源协议下公开。全球超过 140k 个运行实例，GitHub Star 19k，且拥有一个活跃的开发者社区。
+- **[核心开源](https://www.taosdata.com/tdengine/open_source_time-series_database)**：TDengine 的核心代码包括集群功能全部在开源协议下公开。全球超过 140k 个运行实例，GitHub Star 20k，且拥有一个活跃的开发者社区。
 
 采用 TDengine，可将典型的物联网、车联网、工业互联网大数据平台的总拥有成本大幅降低。表现在几个方面：
 

include/client/taos.h

@@ -218,6 +218,9 @@ DLL_EXPORT int taos_get_db_route_info(TAOS *taos, const char *db, TAOS_DB_ROUTE_
 DLL_EXPORT int taos_get_table_vgId(TAOS *taos, const char *db, const char *table, int *vgId);
 
 DLL_EXPORT int       taos_load_table_info(TAOS *taos, const char *tableNameList);
+
+/*  --------------------------schemaless INTERFACE------------------------------- */
+
 DLL_EXPORT TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision);
 DLL_EXPORT TAOS_RES *taos_schemaless_insert_with_reqid(TAOS *taos, char *lines[], int numLines, int protocol,
                                                        int precision, int64_t reqid);
@@ -225,6 +228,13 @@ DLL_EXPORT TAOS_RES *taos_schemaless_insert_raw(TAOS *taos, char *lines, int len
                                                 int precision);
 DLL_EXPORT TAOS_RES *taos_schemaless_insert_raw_with_reqid(TAOS *taos, char *lines, int len, int32_t *totalRows,
                                                            int protocol, int precision, int64_t reqid);
+DLL_EXPORT TAOS_RES *taos_schemaless_insert_ttl(TAOS *taos, char *lines[], int numLines, int protocol, int precision, int32_t ttl);
+DLL_EXPORT TAOS_RES *taos_schemaless_insert_ttl_with_reqid(TAOS *taos, char *lines[], int numLines, int protocol,
+                                                       int precision, int32_t ttl, int64_t reqid);
+DLL_EXPORT TAOS_RES *taos_schemaless_insert_raw_ttl(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol,
+                                                int precision, int32_t ttl);
+DLL_EXPORT TAOS_RES *taos_schemaless_insert_raw_ttl_with_reqid(TAOS *taos, char *lines, int len, int32_t *totalRows,
+                                                           int protocol, int precision, int32_t ttl, int64_t reqid);
 
 /* --------------------------TMQ INTERFACE------------------------------- */
 
@@ -297,6 +307,7 @@ DLL_EXPORT tmq_res_t   tmq_get_res_type(TAOS_RES *res);
 DLL_EXPORT int32_t     tmq_get_raw(TAOS_RES *res, tmq_raw_data *raw);
 DLL_EXPORT int32_t     tmq_write_raw(TAOS *taos, tmq_raw_data raw);
 DLL_EXPORT int         taos_write_raw_block(TAOS *taos, int numOfRows, char *pData, const char *tbname);
+DLL_EXPORT int         taos_write_raw_block_with_fields(TAOS* taos, int rows, char* pData, const char* tbname, TAOS_FIELD *fields, int numFields);
 DLL_EXPORT void        tmq_free_raw(tmq_raw_data raw);
 // Returning null means error. Returned result need to be freed by tmq_free_json_meta
 DLL_EXPORT char *tmq_get_json_meta(TAOS_RES *res);

include/common/tglobal.h

@@ -92,6 +92,10 @@ extern int32_t tsQueryNodeChunkSize;
 extern bool    tsQueryUseNodeAllocator;
 extern bool    tsKeepColumnName;
 extern bool    tsEnableQueryHb;
+extern int32_t tsRedirectPeriod;
+extern int32_t tsRedirectFactor;
+extern int32_t tsRedirectMaxPeriod;
+extern int32_t tsMaxRetryWaitTime;
 
 // client
 extern int32_t tsMinSlidingTime;

include/common/tmsg.h

@@ -1592,14 +1592,14 @@ typedef struct SSubQueryMsg {
   int8_t   explain;
   int8_t   needFetch;
   uint32_t sqlLen;
-  char    *sql;
+  char*    sql;
   uint32_t msgLen;
-  char    *msg;
+  char*    msg;
 } SSubQueryMsg;
 
-int32_t tSerializeSSubQueryMsg(void *buf, int32_t bufLen, SSubQueryMsg *pReq);
-int32_t tDeserializeSSubQueryMsg(void *buf, int32_t bufLen, SSubQueryMsg *pReq);
-void tFreeSSubQueryMsg(SSubQueryMsg *pReq);
+int32_t tSerializeSSubQueryMsg(void* buf, int32_t bufLen, SSubQueryMsg* pReq);
+int32_t tDeserializeSSubQueryMsg(void* buf, int32_t bufLen, SSubQueryMsg* pReq);
+void    tFreeSSubQueryMsg(SSubQueryMsg* pReq);
 
 typedef struct {
   SMsgHead header;
@@ -1638,9 +1638,8 @@ typedef struct {
   int32_t  execId;
 } SResFetchReq;
 
-int32_t tSerializeSResFetchReq(void *buf, int32_t bufLen, SResFetchReq *pReq);
-int32_t tDeserializeSResFetchReq(void *buf, int32_t bufLen, SResFetchReq *pReq);
-
+int32_t tSerializeSResFetchReq(void* buf, int32_t bufLen, SResFetchReq* pReq);
+int32_t tDeserializeSResFetchReq(void* buf, int32_t bufLen, SResFetchReq* pReq);
 
 typedef struct {
   SMsgHead header;
@@ -1713,12 +1712,11 @@ typedef struct {
   int32_t  execId;
 } STaskDropReq;
 
-int32_t tSerializeSTaskDropReq(void *buf, int32_t bufLen, STaskDropReq *pReq);
-int32_t tDeserializeSTaskDropReq(void *buf, int32_t bufLen, STaskDropReq *pReq);
-
-int32_t tSerializeSQueryTableRsp(void *buf, int32_t bufLen, SQueryTableRsp *pRsp);
-int32_t tDeserializeSQueryTableRsp(void *buf, int32_t bufLen, SQueryTableRsp *pRsp);
+int32_t tSerializeSTaskDropReq(void* buf, int32_t bufLen, STaskDropReq* pReq);
+int32_t tDeserializeSTaskDropReq(void* buf, int32_t bufLen, STaskDropReq* pReq);
 
+int32_t tSerializeSQueryTableRsp(void* buf, int32_t bufLen, SQueryTableRsp* pRsp);
+int32_t tDeserializeSQueryTableRsp(void* buf, int32_t bufLen, SQueryTableRsp* pRsp);
 
 typedef struct {
   int32_t code;
@@ -2923,9 +2921,8 @@ typedef struct {
   STqOffsetVal reqOffset;
 } SMqPollReq;
 
-int32_t tSerializeSMqPollReq(void *buf, int32_t bufLen, SMqPollReq *pReq);
-int32_t tDeserializeSMqPollReq(void *buf, int32_t bufLen, SMqPollReq *pReq);
-
+int32_t tSerializeSMqPollReq(void* buf, int32_t bufLen, SMqPollReq* pReq);
+int32_t tDeserializeSMqPollReq(void* buf, int32_t bufLen, SMqPollReq* pReq);
 
 typedef struct {
   int32_t vgId;
@@ -3138,7 +3135,8 @@ int32_t tDecodeDeleteRes(SDecoder* pCoder, SDeleteRes* pRes);
 typedef struct {
   // int64_t uid;
   char    tbname[TSDB_TABLE_NAME_LEN];
-  int64_t ts;
+  int64_t startTs;
+  int64_t endTs;
 } SSingleDeleteReq;
 
 int32_t tEncodeSSingleDeleteReq(SEncoder* pCoder, const SSingleDeleteReq* pReq);
@@ -3161,7 +3159,7 @@ typedef struct {
 
 typedef struct {
   SMsgHead header;
-  SArray*   pMsgs; //SArray<SBatchMsg>
+  SArray*  pMsgs;  // SArray<SBatchMsg>
 } SBatchReq;
 
 typedef struct {
@@ -3173,11 +3171,11 @@ typedef struct {
 } SBatchRspMsg;
 
 typedef struct {
-  SArray* pRsps; //SArray<SBatchRspMsg>
+  SArray* pRsps;  // SArray<SBatchRspMsg>
 } SBatchRsp;
 
-int32_t tSerializeSBatchReq(void *buf, int32_t bufLen, SBatchReq *pReq);
-int32_t tDeserializeSBatchReq(void *buf, int32_t bufLen, SBatchReq *pReq);
+int32_t                  tSerializeSBatchReq(void* buf, int32_t bufLen, SBatchReq* pReq);
+int32_t                  tDeserializeSBatchReq(void* buf, int32_t bufLen, SBatchReq* pReq);
 static FORCE_INLINE void tFreeSBatchReqMsg(void* msg) {
   if (NULL == msg) {
     return;
@@ -3186,8 +3184,8 @@ static FORCE_INLINE void tFreeSBatchReqMsg(void* msg) {
   taosMemoryFree(pMsg->msg);
 }
 
-int32_t tSerializeSBatchRsp(void *buf, int32_t bufLen, SBatchRsp *pRsp);
-int32_t tDeserializeSBatchRsp(void *buf, int32_t bufLen, SBatchRsp *pRsp);
+int32_t tSerializeSBatchRsp(void* buf, int32_t bufLen, SBatchRsp* pRsp);
+int32_t tDeserializeSBatchRsp(void* buf, int32_t bufLen, SBatchRsp* pRsp);
 
 static FORCE_INLINE void tFreeSBatchRspMsg(void* p) {
   if (NULL == p) {
@@ -3198,11 +3196,10 @@ static FORCE_INLINE void tFreeSBatchRspMsg(void* p) {
   taosMemoryFree(pRsp->msg);
 }
 
-int32_t tSerializeSMqAskEpReq(void *buf, int32_t bufLen, SMqAskEpReq *pReq);
-int32_t tDeserializeSMqAskEpReq(void *buf, int32_t bufLen, SMqAskEpReq *pReq);
-int32_t tSerializeSMqHbReq(void *buf, int32_t bufLen, SMqHbReq *pReq);
-int32_t tDeserializeSMqHbReq(void *buf, int32_t bufLen, SMqHbReq *pReq);
-
+int32_t tSerializeSMqAskEpReq(void* buf, int32_t bufLen, SMqAskEpReq* pReq);
+int32_t tDeserializeSMqAskEpReq(void* buf, int32_t bufLen, SMqAskEpReq* pReq);
+int32_t tSerializeSMqHbReq(void* buf, int32_t bufLen, SMqHbReq* pReq);
+int32_t tDeserializeSMqHbReq(void* buf, int32_t bufLen, SMqHbReq* pReq);
 
 #pragma pack(pop)
 

include/common/tmsgdef.h

@@ -280,8 +280,8 @@ enum {
   
   TD_NEW_MSG_SEG(TDMT_VND_STREAM_MSG)
   TD_DEF_MSG_TYPE(TDMT_VND_STREAM_TRIGGER, "vnode-stream-trigger", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_RECOVER_STEP1, "vnode-stream-recover1", NULL, NULL)
-  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_RECOVER_STEP2, "vnode-stream-recover2", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_RECOVER_NONBLOCKING_STAGE, "vnode-stream-recover1", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_STREAM_RECOVER_BLOCKING_STAGE, "vnode-stream-recover2", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_STREAM_MAX_MSG, "vnd-stream-max", NULL, NULL)
 
   TD_NEW_MSG_SEG(TDMT_VND_TMQ_MSG)

include/libs/executor/executor.h

@@ -160,13 +160,6 @@ int32_t qAsyncKillTask(qTaskInfo_t tinfo);
  */
 void qDestroyTask(qTaskInfo_t tinfo);
 
-/**
- * Get the queried table uid
- * @param qHandle
- * @return
- */
-int64_t qGetQueriedTableUid(qTaskInfo_t tinfo);
-
 /**
  * Extract the qualified table id list, and than pass them to the TSDB driver to load the required table data blocks.
  *

include/libs/function/function.h

@@ -57,7 +57,7 @@ typedef struct SFuncExecFuncs {
 #define MAX_INTERVAL_TIME_WINDOW 10000000  // maximum allowed time windows in final results
 
 #define TOP_BOTTOM_QUERY_LIMIT    100
-#define FUNCTIONS_NAME_MAX_LENGTH 16
+#define FUNCTIONS_NAME_MAX_LENGTH 32
 
 typedef struct SResultRowEntryInfo {
   bool     initialized : 1;  // output buffer has been initialized

include/libs/nodes/querynodes.h

@@ -106,7 +106,7 @@ typedef struct SValueNode {
     char*    p;
   } datum;
   int64_t typeData;
-  char    unit;
+  int8_t  unit;
 } SValueNode;
 
 typedef struct SLeftValueNode {

include/libs/parser/parser.h

@@ -108,7 +108,7 @@ int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char*
 void*   smlInitHandle(SQuery* pQuery);
 void    smlDestroyHandle(void* pHandle);
 int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols, bool format, STableMeta* pTableMeta,
-                    char* tableName, const char* sTableName, int32_t sTableNameLen, char* msgBuf, int16_t msgBufLen);
+                    char* tableName, const char* sTableName, int32_t sTableNameLen, int32_t ttl, char* msgBuf, int16_t msgBufLen);
 int32_t smlBuildOutput(void* handle, SHashObj* pVgHash);
 
 int32_t rewriteToVnodeModifyOpStmt(SQuery* pQuery, SArray* pBufArray);

include/libs/qcom/query.h

@@ -259,9 +259,15 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
 #define NEED_CLIENT_HANDLE_ERROR(_code)                                          \
   (NEED_CLIENT_RM_TBLMETA_ERROR(_code) || NEED_CLIENT_REFRESH_VG_ERROR(_code) || \
    NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code))
+
+#define SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_SYN_NOT_LEADER || (_code) == TSDB_CODE_SYN_INTERNAL_ERROR)
+#define SYNC_SELF_LEADER_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_SYN_NOT_LEADER || (_code) == TSDB_CODE_SYN_INTERNAL_ERROR)
+#define SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) (false)  // used later
+
 #define NEED_REDIRECT_ERROR(_code)                                                  \
   ((_code) == TSDB_CODE_RPC_REDIRECT || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL || \
-   (_code) == TSDB_CODE_NODE_NOT_DEPLOYED || (_code) == TSDB_CODE_SYN_NOT_LEADER || \
+   (_code) == TSDB_CODE_NODE_NOT_DEPLOYED || SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || \
+   SYNC_SELF_LEADER_REDIRECT_ERROR(_code) || SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) ||   \
    (_code) == TSDB_CODE_APP_NOT_READY || (_code) == TSDB_CODE_RPC_BROKEN_LINK)
 
 #define NEED_CLIENT_RM_TBLMETA_REQ(_type)                                                                  \
@@ -270,7 +276,8 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
 
 #define NEED_SCHEDULER_REDIRECT_ERROR(_code)                                      \
   ((_code) == TSDB_CODE_RPC_REDIRECT || (_code) == TSDB_CODE_NODE_NOT_DEPLOYED || \
-   (_code) == TSDB_CODE_SYN_NOT_LEADER || (_code) == TSDB_CODE_APP_NOT_READY)
+   SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || SYNC_SELF_LEADER_REDIRECT_ERROR(_code) || \
+   SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) || (_code) == TSDB_CODE_APP_NOT_READY)
 
 #define REQUEST_TOTAL_EXEC_TIMES 2
 

include/libs/sync/sync.h

@@ -45,6 +45,9 @@ extern "C" {
 #define SYNC_APPEND_ENTRIES_TIMEOUT_MS 10000
 #define SYNC_HEART_TIMEOUT_MS          1000 * 8
 
+#define SYNC_HEARTBEAT_SLOW_MS       1500
+#define SYNC_HEARTBEAT_REPLY_SLOW_MS 1500
+
 #define SYNC_MAX_BATCH_SIZE 1
 #define SYNC_INDEX_BEGIN    0
 #define SYNC_INDEX_INVALID  -1
@@ -66,6 +69,7 @@ typedef struct SWal           SWal;
 typedef struct SSyncRaftEntry SSyncRaftEntry;
 
 typedef enum {
+  TAOS_SYNC_STATE_OFFLINE = 0,
   TAOS_SYNC_STATE_FOLLOWER = 100,
   TAOS_SYNC_STATE_CANDIDATE = 101,
   TAOS_SYNC_STATE_LEADER = 102,
@@ -228,6 +232,8 @@ int32_t syncEndSnapshot(int64_t rid);
 int32_t syncLeaderTransfer(int64_t rid);
 int32_t syncStepDown(int64_t rid, SyncTerm newTerm);
 bool    syncIsReadyForRead(int64_t rid);
+bool    syncSnapshotSending(int64_t rid);
+bool    syncSnapshotRecving(int64_t rid);
 
 SSyncState  syncGetState(int64_t rid);
 void        syncGetRetryEpSet(int64_t rid, SEpSet* pEpSet);

include/libs/transport/trpc.h

@@ -85,6 +85,11 @@ typedef struct SRpcInit {
   int32_t  retryLimit;     // retry limit
   int32_t  retryInterval;  // retry interval ms
 
+  int32_t retryMinInterval;  // retry init interval
+  int32_t retryStepFactor;   // retry interval factor
+  int32_t retryMaxInterval;  // retry max interval
+  int64_t retryMaxTimouet;
+
   int32_t compressSize;  // -1: no compress, 0 : all data compressed, size: compress data if larger than size
   int8_t  encryption;    // encrypt or not
 

include/os/os.h

@@ -48,8 +48,10 @@ extern "C" {
 #else
 #include <argp.h>
 #include <sys/prctl.h>
+#if defined(_TD_X86_)
 #include <cpuid.h>
 #endif
+#endif
 #else
 
 #ifndef __func__

include/os/osSocket.h

@@ -169,6 +169,9 @@ void        taosSetMaskSIGPIPE();
 uint32_t    taosInetAddr(const char *ipAddr);
 const char *taosInetNtoa(struct in_addr ipInt, char *dstStr, int32_t len);
 
+uint64_t taosHton64(uint64_t val);
+uint64_t taosNtoh64(uint64_t val);
+
 #ifdef __cplusplus
 }
 #endif

include/util/taoserror.h

@@ -92,6 +92,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_NO_AVAIL_DISK                 TAOS_DEF_ERROR_CODE(0, 0x0129)
 #define TSDB_CODE_NOT_FOUND                     TAOS_DEF_ERROR_CODE(0, 0x012A)
 #define TSDB_CODE_NO_DISKSPACE                  TAOS_DEF_ERROR_CODE(0, 0x012B)
+#define TSDB_CODE_TIMEOUT_ERROR                 TAOS_DEF_ERROR_CODE(0, 0x012C)
 
 //client
 #define TSDB_CODE_TSC_INVALID_OPERATION         TAOS_DEF_ERROR_CODE(0, 0x0200)
@@ -415,6 +416,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_SYN_PROPOSE_NOT_READY         TAOS_DEF_ERROR_CODE(0, 0x0911)
 #define TSDB_CODE_SYN_STANDBY_NOT_READY         TAOS_DEF_ERROR_CODE(0, 0x0912)
 #define TSDB_CODE_SYN_BATCH_ERROR               TAOS_DEF_ERROR_CODE(0, 0x0913)
+#define TSDB_CODE_SYN_RESTORING                 TAOS_DEF_ERROR_CODE(0, 0x0914)
 #define TSDB_CODE_SYN_INTERNAL_ERROR            TAOS_DEF_ERROR_CODE(0, 0x09FF)
 
 // tq

include/util/tarray.h

@@ -220,15 +220,10 @@ void taosArrayClear(SArray* pArray);
  */
 void taosArrayClearEx(SArray* pArray, void (*fp)(void*));
 
-/**
- * clear the array (remove all element)
- * @param pArray
- * @param fp
- */
-void taosArrayClearP(SArray* pArray, FDelete fp);
-
 void* taosArrayDestroy(SArray* pArray);
+
 void  taosArrayDestroyP(SArray* pArray, FDelete fp);
+
 void  taosArrayDestroyEx(SArray* pArray, FDelete fp);
 
 /**
@@ -238,12 +233,6 @@ void  taosArrayDestroyEx(SArray* pArray, FDelete fp);
  */
 void taosArraySort(SArray* pArray, __compar_fn_t comparFn);
 
-/**
- * sort string array
- * @param pArray
- */
-void taosArraySortString(SArray* pArray, __compar_fn_t comparFn);
-
 /**
  * search the array
  * @param pArray

source/client/src/clientEnv.c

@@ -148,6 +148,10 @@ void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
   rpcInit.dfp = destroyAhandle;
   rpcInit.retryLimit = tsRpcRetryLimit;
   rpcInit.retryInterval = tsRpcRetryInterval;
+  rpcInit.retryMinInterval = tsRedirectPeriod;
+  rpcInit.retryStepFactor = tsRedirectFactor;
+  rpcInit.retryMaxInterval = tsRedirectMaxPeriod;
+  rpcInit.retryMaxTimouet = tsMaxRetryWaitTime;
 
   void *pDnodeConn = rpcOpen(&rpcInit);
   if (pDnodeConn == NULL) {

source/client/src/clientImpl.c

@@ -1645,7 +1645,7 @@ int32_t getVersion1BlockMetaSize(const char* p, int32_t numOfCols) {
 static int32_t estimateJsonLen(SReqResultInfo* pResultInfo, int32_t numOfCols, int32_t numOfRows) {
   char* p = (char*)pResultInfo->pData;
 
-  // version + length + numOfRows + numOfCol + groupId + flag_segment + column_info
+  // | version | total length | total rows | total columns | flag seg| block group id | column schema | each column length |
   int32_t  len = getVersion1BlockMetaSize(p, numOfCols);
   int32_t* colLength = (int32_t*)(p + len);
   len += sizeof(int32_t) * numOfCols;

source/client/src/clientRawBlockWrite.c

@@ -1211,6 +1211,208 @@ static void destroyVgHash(void* data) {
   taosMemoryFreeClear(vgData->data);
 }
 
+int taos_write_raw_block_with_fields(TAOS* taos, int rows, char* pData, const char* tbname, TAOS_FIELD *fields, int numFields){
+  int32_t     code = TSDB_CODE_SUCCESS;
+  STableMeta* pTableMeta = NULL;
+  SQuery*     pQuery = NULL;
+  SSubmitReq* subReq = NULL;
+
+  SRequestObj* pRequest = (SRequestObj*)createRequest(*(int64_t*)taos, TSDB_SQL_INSERT, 0);
+  if (!pRequest) {
+    uError("WriteRaw:createRequest error request is null");
+    code = terrno;
+    goto end;
+  }
+
+  pRequest->syncQuery = true;
+  if (!pRequest->pDb) {
+    uError("WriteRaw:not use db");
+    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
+    goto end;
+  }
+
+  SName pName = {TSDB_TABLE_NAME_T, pRequest->pTscObj->acctId, {0}, {0}};
+  tstrncpy(pName.dbname, pRequest->pDb, sizeof(pName.dbname));
+  tstrncpy(pName.tname, tbname, sizeof(pName.tname));
+
+  struct SCatalog* pCatalog = NULL;
+  code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);
+  if (code != TSDB_CODE_SUCCESS) {
+    uError("WriteRaw: get gatlog error");
+    goto end;
+  }
+
+  SRequestConnInfo conn = {0};
+  conn.pTrans = pRequest->pTscObj->pAppInfo->pTransporter;
+  conn.requestId = pRequest->requestId;
+  conn.requestObjRefId = pRequest->self;
+  conn.mgmtEps = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp);
+
+  SVgroupInfo vgData = {0};
+  code = catalogGetTableHashVgroup(pCatalog, &conn, &pName, &vgData);
+  if (code != TSDB_CODE_SUCCESS) {
+    uError("WriteRaw:catalogGetTableHashVgroup failed. table name: %s", tbname);
+    goto end;
+  }
+
+  code = catalogGetTableMeta(pCatalog, &conn, &pName, &pTableMeta);
+  if (code != TSDB_CODE_SUCCESS) {
+    uError("WriteRaw:catalogGetTableMeta failed. table name: %s", tbname);
+    goto end;
+  }
+  uint64_t suid = (TSDB_NORMAL_TABLE == pTableMeta->tableType ? 0 : pTableMeta->suid);
+  uint64_t uid = pTableMeta->uid;
+  int32_t  numOfCols = pTableMeta->tableInfo.numOfColumns;
+
+  uint16_t fLen = 0;
+  int32_t  rowSize = 0;
+  int16_t  nVar = 0;
+    for (int i = 0; i < pTableMeta->tableInfo.numOfColumns; i++) {
+      SSchema* schema = pTableMeta->schema + i;
+      fLen += TYPE_BYTES[schema->type];
+      rowSize += schema->bytes;
+      if (IS_VAR_DATA_TYPE(schema->type)) {
+        nVar++;
+      }
+    }
+
+  fLen -= sizeof(TSKEY);
+
+  int32_t extendedRowSize = rowSize + TD_ROW_HEAD_LEN - sizeof(TSKEY) + nVar * sizeof(VarDataOffsetT) +
+                            (int32_t)TD_BITMAP_BYTES(numOfCols - 1);
+  int32_t schemaLen = 0;
+  int32_t submitLen = sizeof(SSubmitBlk) + schemaLen + rows * extendedRowSize;
+
+  int32_t totalLen = sizeof(SSubmitReq) + submitLen;
+  subReq = taosMemoryCalloc(1, totalLen);
+  SSubmitBlk* blk = POINTER_SHIFT(subReq, sizeof(SSubmitReq));
+  void*       blkSchema = POINTER_SHIFT(blk, sizeof(SSubmitBlk));
+  STSRow*     rowData = POINTER_SHIFT(blkSchema, schemaLen);
+
+  SRowBuilder rb = {0};
+  tdSRowInit(&rb, pTableMeta->sversion);
+  tdSRowSetTpInfo(&rb, numOfCols, fLen);
+  int32_t dataLen = 0;
+
+  // | version | total length | total rows | total columns | flag seg| block group id | column schema | each column length |
+  char*    pStart = pData + getVersion1BlockMetaSize(pData, numFields);
+  int32_t* colLength = (int32_t*)pStart;
+  pStart += sizeof(int32_t) * numFields;
+
+  SResultColumn* pCol = taosMemoryCalloc(numFields, sizeof(SResultColumn));
+
+  for (int32_t i = 0; i < numFields; ++i) {
+    if (IS_VAR_DATA_TYPE(fields[i].type)) {
+      pCol[i].offset = (int32_t*)pStart;
+      pStart += rows * sizeof(int32_t);
+    } else {
+      pCol[i].nullbitmap = pStart;
+      pStart += BitmapLen(rows);
+    }
+
+    pCol[i].pData = pStart;
+    pStart += colLength[i];
+  }
+
+  SHashObj* schemaHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
+  for (int i = 0; i < numFields; i++) {
+    TAOS_FIELD* schema = &fields[i];
+    taosHashPut(schemaHash, schema->name, strlen(schema->name), &i, sizeof(int32_t));
+  }
+
+  for (int32_t j = 0; j < rows; j++) {
+    tdSRowResetBuf(&rb, rowData);
+    int32_t offset = 0;
+    for (int32_t k = 0; k < numOfCols; k++) {
+      const SSchema* pColumn = &pTableMeta->schema[k];
+      int32_t*       index = taosHashGet(schemaHash, pColumn->name, strlen(pColumn->name));
+      if (!index) {   // add none
+        tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NONE, NULL, false, offset, k);
+      }else{
+        if (IS_VAR_DATA_TYPE(pColumn->type)) {
+          if (pCol[*index].offset[j] != -1) {
+            char* data = pCol[*index].pData + pCol[*index].offset[j];
+            tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NORM, data, true, offset, k);
+          } else {
+            tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NULL, NULL, false, offset, k);
+          }
+        } else {
+          if (!colDataIsNull_f(pCol[*index].nullbitmap, j)) {
+            char* data = pCol[*index].pData + pColumn->bytes * j;
+            tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NORM, data, true, offset, k);
+          } else {
+            tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NULL, NULL, false, offset, k);
+          }
+        }
+      }
+
+      if (pColumn->colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
+        offset += TYPE_BYTES[pColumn->type];
+      }
+    }
+    tdSRowEnd(&rb);
+    int32_t rowLen = TD_ROW_LEN(rowData);
+    rowData = POINTER_SHIFT(rowData, rowLen);
+    dataLen += rowLen;
+  }
+
+  taosHashCleanup(schemaHash);
+  taosMemoryFree(pCol);
+
+  blk->uid = htobe64(uid);
+  blk->suid = htobe64(suid);
+  blk->sversion = htonl(pTableMeta->sversion);
+  blk->schemaLen = htonl(schemaLen);
+  blk->numOfRows = htonl(rows);
+  blk->dataLen = htonl(dataLen);
+  subReq->length = sizeof(SSubmitReq) + sizeof(SSubmitBlk) + schemaLen + dataLen;
+  subReq->numOfBlocks = 1;
+
+  pQuery = (SQuery*)nodesMakeNode(QUERY_NODE_QUERY);
+  if (NULL == pQuery) {
+    uError("create SQuery error");
+    code = TSDB_CODE_OUT_OF_MEMORY;
+    goto end;
+  }
+  pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;
+  pQuery->haveResultSet = false;
+  pQuery->msgType = TDMT_VND_SUBMIT;
+  pQuery->pRoot = (SNode*)nodesMakeNode(QUERY_NODE_VNODE_MODIF_STMT);
+  if (NULL == pQuery->pRoot) {
+    uError("create pQuery->pRoot error");
+    code = TSDB_CODE_OUT_OF_MEMORY;
+    goto end;
+  }
+  SVnodeModifOpStmt* nodeStmt = (SVnodeModifOpStmt*)(pQuery->pRoot);
+  nodeStmt->pDataBlocks = taosArrayInit(1, POINTER_BYTES);
+
+  SVgDataBlocks* dst = taosMemoryCalloc(1, sizeof(SVgDataBlocks));
+  if (NULL == dst) {
+    code = TSDB_CODE_TSC_OUT_OF_MEMORY;
+    goto end;
+  }
+  dst->vg = vgData;
+  dst->numOfTables = subReq->numOfBlocks;
+  dst->size = subReq->length;
+  dst->pData = (char*)subReq;
+  subReq->header.vgId = htonl(dst->vg.vgId);
+  subReq->version = htonl(1);
+  subReq->header.contLen = htonl(subReq->length);
+  subReq->length = htonl(subReq->length);
+  subReq->numOfBlocks = htonl(subReq->numOfBlocks);
+  subReq = NULL;  // no need free
+  taosArrayPush(nodeStmt->pDataBlocks, &dst);
+
+  launchQueryImpl(pRequest, pQuery, true, NULL);
+  code = pRequest->code;
+
+  end:
+  taosMemoryFreeClear(pTableMeta);
+  qDestroyQuery(pQuery);
+  taosMemoryFree(subReq);
+  return code;
+}
+
 int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname) {
   int32_t     code = TSDB_CODE_SUCCESS;
   STableMeta* pTableMeta = NULL;
@@ -1293,6 +1495,7 @@ int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname)
   tdSRowSetTpInfo(&rb, numOfCols, fLen);
   int32_t dataLen = 0;
 
+  // | version | total length | total rows | total columns | flag seg| block group id | column schema | each column length |
   char*    pStart = pData + getVersion1BlockMetaSize(pData, numOfCols);
   int32_t* colLength = (int32_t*)pStart;
   pStart += sizeof(int32_t) * numOfCols;
@@ -1577,7 +1780,7 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) {
         const SSchema* pColumn = &pTableMeta->schema[k];
         int32_t*       index = taosHashGet(schemaHash, pColumn->name, strlen(pColumn->name));
         if (!index) {
-          tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NULL, NULL, false, offset, k);
+          tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NONE, NULL, false, offset, k);
         } else {
           char* colData = rspObj.resInfo.row[*index];
           if (!colData) {
@@ -1670,6 +1873,7 @@ end:
   return code;
 }
 
+
 static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen) {
   int32_t        code = TSDB_CODE_SUCCESS;
   SHashObj*      pVgHash = NULL;
@@ -1882,7 +2086,7 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen)
         const SSchema* pColumn = &pTableMeta->schema[k];
         int32_t*       index = taosHashGet(schemaHash, pColumn->name, strlen(pColumn->name));
         if (!index) {
-          tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NULL, NULL, false, offset, k);
+          tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NONE, NULL, false, offset, k);
         } else {
           char* colData = rspObj.resInfo.row[*index];
           if (!colData) {
@@ -1963,7 +2167,7 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen)
   launchQueryImpl(pRequest, pQuery, true, NULL);
   code = pRequest->code;
 
-end:
+  end:
   tDeleteSTaosxRsp(&rspObj.rsp);
   rspObj.resInfo.pRspMsg = NULL;
   doFreeReqResultInfo(&rspObj.resInfo);

source/client/src/clientSml.c

@@ -163,6 +163,8 @@ typedef struct {
   SMLProtocolType protocol;
   int8_t          precision;
   bool            dataFormat;  // true means that the name and order of keys in each line are the same(only for influx protocol)
+  bool            isRawLine;
+  int32_t         ttl;
 
   SHashObj *childTables;
   SHashObj *superTables;
@@ -863,6 +865,7 @@ static int32_t smlParseTS(SSmlHandle *info, const char *data, int32_t len, SArra
   } else {
     ASSERT(0);
   }
+  uDebug("SML:0x%" PRIx64 " smlParseTS:%" PRId64, info->id, ts);
 
   if (ts == -1) return TSDB_CODE_INVALID_TIMESTAMP;
 
@@ -2063,7 +2066,7 @@ static int32_t smlParseJSONString(SSmlHandle *info, cJSON *root, SSmlTableInfo *
 
 static int32_t smlParseInfluxLine(SSmlHandle *info, const char *sql, const int len) {
   SSmlLineInfo elements = {0};
-  uDebug("SML:0x%" PRIx64 " smlParseInfluxLine sql", info->id);
+  uDebug("SML:0x%" PRIx64 " smlParseInfluxLine sql:%s", info->id, (info->isRawLine ? "rawdata" : sql));
 
   int ret = smlParseInfluxString(sql, sql + len, &elements, &info->msgBuf);
   if (ret != TSDB_CODE_SUCCESS) {
@@ -2323,7 +2326,7 @@ static int32_t smlInsertData(SSmlHandle *info) {
 
     code = smlBindData(info->exec, tableData->tags, (*pMeta)->cols, tableData->cols, info->dataFormat,
                        (*pMeta)->tableMeta, tableData->childTableName, tableData->sTableName, tableData->sTableNameLen,
-                       info->msgBuf.buf, info->msgBuf.len);
+                       info->ttl, info->msgBuf.buf, info->msgBuf.len);
     if (code != TSDB_CODE_SUCCESS) {
       uError("SML:0x%" PRIx64 " smlBindData failed", info->id);
       return code;
@@ -2514,7 +2517,7 @@ static void smlInsertCallback(void *param, void *res, int32_t code) {
 }
 
 TAOS_RES *taos_schemaless_insert_inner(SRequestObj *request, char *lines[], char *rawLine, char *rawLineEnd,
-                                       int numLines, int protocol, int precision) {
+                                       int numLines, int protocol, int precision, int32_t ttl) {
   int      batchs = 0;
   STscObj *pTscObj = request->pTscObj;
 
@@ -2575,6 +2578,9 @@ TAOS_RES *taos_schemaless_insert_inner(SRequestObj *request, char *lines[], char
       goto end;
     }
 
+    info->isRawLine = (rawLine == NULL);
+    info->ttl       = ttl;
+
     int32_t perBatch = LINE_BATCH;
 
     if (numLines > perBatch) {
@@ -2637,13 +2643,14 @@ end:
  * @return TAOS_RES
  */
 
-TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision) {
+TAOS_RES *taos_schemaless_insert_ttl_with_reqid(TAOS *taos, char *lines[], int numLines, int protocol, int precision,
+                                                int32_t ttl, int64_t reqid) {
   if (NULL == taos) {
     terrno = TSDB_CODE_TSC_DISCONNECTED;
     return NULL;
   }
 
-  SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT, 0);
+  SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT, reqid);
   if (!request) {
     uError("SML:taos_schemaless_insert error request is null");
     return NULL;
@@ -2656,40 +2663,29 @@ TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int pr
     return (TAOS_RES *)request;
   }
 
-  return taos_schemaless_insert_inner(request, lines, NULL, NULL, numLines, protocol, precision);
+  return taos_schemaless_insert_inner(request, lines, NULL, NULL, numLines, protocol, precision, ttl);
 }
 
-TAOS_RES *taos_schemaless_insert_with_reqid(TAOS *taos, char *lines[], int numLines, int protocol, int precision,
-                                            int64_t reqid) {
-  if (NULL == taos) {
-    terrno = TSDB_CODE_TSC_DISCONNECTED;
-    return NULL;
-  }
-
-  SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT, reqid);
-  if (!request) {
-    uError("SML:taos_schemaless_insert error request is null");
-    return NULL;
-  }
+TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision) {
+  return taos_schemaless_insert_ttl_with_reqid(taos, lines, numLines, protocol, precision, TSDB_DEFAULT_TABLE_TTL, 0);
+}
 
-  if (!lines) {
-    SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
-    request->code = TSDB_CODE_SML_INVALID_DATA;
-    smlBuildInvalidDataMsg(&msg, "lines is null", NULL);
-    return (TAOS_RES *)request;
-  }
+TAOS_RES *taos_schemaless_insert_ttl(TAOS *taos, char *lines[], int numLines, int protocol, int precision, int32_t ttl) {
+  return taos_schemaless_insert_ttl_with_reqid(taos, lines, numLines, protocol, precision, ttl, 0);
+}
 
-  return taos_schemaless_insert_inner(request, lines, NULL, NULL, numLines, protocol, precision);
+TAOS_RES *taos_schemaless_insert_with_reqid(TAOS *taos, char *lines[], int numLines, int protocol, int precision, int64_t reqid) {
+  return taos_schemaless_insert_ttl_with_reqid(taos, lines, numLines, protocol, precision, TSDB_DEFAULT_TABLE_TTL, reqid);
 }
 
-TAOS_RES *taos_schemaless_insert_raw(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol,
-                                     int precision) {
+TAOS_RES *taos_schemaless_insert_raw_ttl_with_reqid(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol,
+                                                int precision, int32_t ttl, int64_t reqid) {
   if (NULL == taos) {
     terrno = TSDB_CODE_TSC_DISCONNECTED;
     return NULL;
   }
 
-  SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT, 0);
+  SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT, reqid);
   if (!request) {
     uError("SML:taos_schemaless_insert error request is null");
     return NULL;
@@ -2714,40 +2710,16 @@ TAOS_RES *taos_schemaless_insert_raw(TAOS *taos, char *lines, int len, int32_t *
       tmp = lines + i + 1;
     }
   }
-  return taos_schemaless_insert_inner(request, NULL, lines, lines + len, numLines, protocol, precision);
+  return taos_schemaless_insert_inner(request, NULL, lines, lines + len, numLines, protocol, precision, ttl);
 }
 
-TAOS_RES *taos_schemaless_insert_raw_with_reqid(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol,
-                                                int precision, int64_t reqid) {
-  if (NULL == taos) {
-    terrno = TSDB_CODE_TSC_DISCONNECTED;
-    return NULL;
-  }
-
-  SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT, reqid);
-  if (!request) {
-    uError("SML:taos_schemaless_insert error request is null");
-    return NULL;
-  }
-
-  if (!lines || len <= 0) {
-    SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
-    request->code = TSDB_CODE_SML_INVALID_DATA;
-    smlBuildInvalidDataMsg(&msg, "lines is null", NULL);
-    return (TAOS_RES *)request;
-  }
+TAOS_RES *taos_schemaless_insert_raw_with_reqid(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol, int precision, int64_t reqid) {
+  return taos_schemaless_insert_raw_ttl_with_reqid(taos, lines, len, totalRows, protocol, precision, TSDB_DEFAULT_TABLE_TTL, reqid);
+}
+TAOS_RES *taos_schemaless_insert_raw_ttl(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol, int precision, int32_t ttl) {
+  return taos_schemaless_insert_raw_ttl_with_reqid(taos, lines, len, totalRows, protocol, precision, ttl, 0);
+}
 
-  int numLines = 0;
-  *totalRows = 0;
-  char *tmp = lines;
-  for (int i = 0; i < len; i++) {
-    if (lines[i] == '\n' || i == len - 1) {
-      numLines++;
-      if (tmp[0] != '#' || protocol != TSDB_SML_LINE_PROTOCOL) {  // ignore comment
-        (*totalRows)++;
-      }
-      tmp = lines + i + 1;
-    }
-  }
-  return taos_schemaless_insert_inner(request, NULL, lines, lines + len, numLines, protocol, precision);
+TAOS_RES *taos_schemaless_insert_raw(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol, int precision) {
+  return taos_schemaless_insert_raw_ttl_with_reqid(taos, lines, len, totalRows, protocol, precision, TSDB_DEFAULT_TABLE_TTL, 0);
 }

source/common/src/systable.c

@@ -206,16 +206,15 @@ static const SSysDbTableSchema vgroupsSchema[] = {
     {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
     {.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
     {.name = "tables", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
-    {.name = "v1_dnode", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
-    {.name = "v1_status", .bytes = 10 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
-    {.name = "v2_dnode", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
-    {.name = "v2_status", .bytes = 10 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
-    {.name = "v3_dnode", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
-    {.name = "v3_status", .bytes = 10 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
-    {.name = "status", .bytes = 12 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
+    {.name = "v1_dnode", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT, .sysInfo = true},
+    {.name = "v1_status", .bytes = 9 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
+    {.name = "v2_dnode", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT, .sysInfo = true},
+    {.name = "v2_status", .bytes = 9 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
+    {.name = "v3_dnode", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT, .sysInfo = true},
+    {.name = "v3_status", .bytes = 9 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
+    {.name = "v4_dnode", .bytes = 2, .type = TSDB_DATA_TYPE_SMALLINT, .sysInfo = true},
+    {.name = "v4_status", .bytes = 9 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
     {.name = "cacheload", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
-    {.name = "nfiles", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
-    {.name = "file_size", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
     {.name = "tsma", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT, .sysInfo = true},
 };
 

source/common/src/tglobal.c

@@ -16,9 +16,9 @@
 #define _DEFAULT_SOURCE
 #include "tglobal.h"
 #include "tconfig.h"
-#include "tmisce.h"
 #include "tgrant.h"
 #include "tlog.h"
+#include "tmisce.h"
 
 GRANT_CFG_DECLARE;
 
@@ -86,6 +86,10 @@ bool    tsQueryPlannerTrace = false;
 int32_t tsQueryNodeChunkSize = 32 * 1024;
 bool    tsQueryUseNodeAllocator = true;
 bool    tsKeepColumnName = false;
+int32_t tsRedirectPeriod = 10;
+int32_t tsRedirectFactor = 2;
+int32_t tsRedirectMaxPeriod = 1000;
+int32_t tsMaxRetryWaitTime = 10000;
 
 /*
  * denote if the server needs to compress response message at the application layer to client, including query rsp,
@@ -120,7 +124,7 @@ int32_t tsMinIntervalTime = 1;
 int32_t tsMaxMemUsedByInsert = 1024;
 
 float   tsSelectivityRatio = 1.0;
-int32_t tsTagFilterResCacheSize = 1024*10;
+int32_t tsTagFilterResCacheSize = 1024 * 10;
 
 // the maximum allowed query buffer size during query processing for each data node.
 // -1 no limit (default)
@@ -305,6 +309,7 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "maxMemUsedByInsert", tsMaxMemUsedByInsert, 1, INT32_MAX, true) != 0) return -1;
   if (cfgAddInt32(pCfg, "rpcRetryLimit", tsRpcRetryLimit, 1, 100000, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "rpcRetryInterval", tsRpcRetryInterval, 1, 100000, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "maxRetryWaitTime", tsMaxRetryWaitTime, 0, 86400000, 0) != 0) return -1;
 
   tsNumOfTaskQueueThreads = tsNumOfCores / 2;
   tsNumOfTaskQueueThreads = TMAX(tsNumOfTaskQueueThreads, 4);
@@ -659,6 +664,7 @@ static int32_t taosSetClientCfg(SConfig *pCfg) {
 
   tsRpcRetryLimit = cfgGetItem(pCfg, "rpcRetryLimit")->i32;
   tsRpcRetryInterval = cfgGetItem(pCfg, "rpcRetryInterval")->i32;
+  tsMaxRetryWaitTime = cfgGetItem(pCfg, "maxRetryWaitTime")->i32;
   return 0;
 }
 
@@ -874,6 +880,8 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) {
             tsMaxNumOfDistinctResults = cfgGetItem(pCfg, "maxNumOfDistinctRes")->i32;
           } else if (strcasecmp("maxMemUsedByInsert", name) == 0) {
             tsMaxMemUsedByInsert = cfgGetItem(pCfg, "maxMemUsedByInsert")->i32;
+          } else if (strcasecmp("maxRetryWaitTime", name) == 0) {
+            tsMaxRetryWaitTime = cfgGetItem(pCfg, "maxRetryWaitTime")->i32;
           }
           break;
         }

source/common/src/tmsg.c

@@ -4576,7 +4576,6 @@ int32_t tDeserializeSBatchRsp(void *buf, int32_t bufLen, SBatchRsp *pRsp) {
   return 0;
 }
 
-
 int32_t tSerializeSMqAskEpReq(void *buf, int32_t bufLen, SMqAskEpReq *pReq) {
   SEncoder encoder = {0};
   tEncoderInit(&encoder, buf, bufLen);
@@ -4664,7 +4663,7 @@ int32_t tSerializeSSubQueryMsg(void *buf, int32_t bufLen, SSubQueryMsg *pReq) {
   if (tEncodeU32(&encoder, pReq->sqlLen) < 0) return -1;
   if (tEncodeCStrWithLen(&encoder, pReq->sql, pReq->sqlLen) < 0) return -1;
   if (tEncodeU32(&encoder, pReq->msgLen) < 0) return -1;
-  if (tEncodeBinary(&encoder, (uint8_t*)pReq->msg, pReq->msgLen) < 0) return -1;
+  if (tEncodeBinary(&encoder, (uint8_t *)pReq->msg, pReq->msgLen) < 0) return -1;
 
   tEndEncode(&encoder);
 
@@ -4704,7 +4703,7 @@ int32_t tDeserializeSSubQueryMsg(void *buf, int32_t bufLen, SSubQueryMsg *pReq)
   if (tDecodeU32(&decoder, &pReq->sqlLen) < 0) return -1;
   if (tDecodeCStrAlloc(&decoder, &pReq->sql) < 0) return -1;
   if (tDecodeU32(&decoder, &pReq->msgLen) < 0) return -1;
-  if (tDecodeBinaryAlloc(&decoder, (void**)&pReq->msg, NULL) < 0) return -1;
+  if (tDecodeBinaryAlloc(&decoder, (void **)&pReq->msg, NULL) < 0) return -1;
 
   tEndDecode(&decoder);
 
@@ -4721,7 +4720,6 @@ void tFreeSSubQueryMsg(SSubQueryMsg *pReq) {
   taosMemoryFreeClear(pReq->msg);
 }
 
-
 int32_t tSerializeSResFetchReq(void *buf, int32_t bufLen, SResFetchReq *pReq) {
   int32_t headLen = sizeof(SMsgHead);
   if (buf != NULL) {
@@ -4775,7 +4773,6 @@ int32_t tDeserializeSResFetchReq(void *buf, int32_t bufLen, SResFetchReq *pReq)
   return 0;
 }
 
-
 int32_t tSerializeSTqOffsetVal(SEncoder *pEncoder, STqOffsetVal *pOffset) {
   if (tEncodeI8(pEncoder, pOffset->type) < 0) return -1;
   if (tEncodeI64(pEncoder, pOffset->uid) < 0) return -1;
@@ -4853,7 +4850,6 @@ int32_t tDeserializeSMqPollReq(void *buf, int32_t bufLen, SMqPollReq *pReq) {
   return 0;
 }
 
-
 int32_t tSerializeSTaskDropReq(void *buf, int32_t bufLen, STaskDropReq *pReq) {
   int32_t headLen = sizeof(SMsgHead);
   if (buf != NULL) {
@@ -4946,7 +4942,6 @@ int32_t tDeserializeSQueryTableRsp(void *buf, int32_t bufLen, SQueryTableRsp *pR
   return 0;
 }
 
-
 int32_t tSerializeSSchedulerHbReq(void *buf, int32_t bufLen, SSchedulerHbReq *pReq) {
   int32_t headLen = sizeof(SMsgHead);
   if (buf != NULL) {
@@ -6645,13 +6640,15 @@ void tDeleteSTaosxRsp(STaosxRsp *pRsp) {
 
 int32_t tEncodeSSingleDeleteReq(SEncoder *pEncoder, const SSingleDeleteReq *pReq) {
   if (tEncodeCStr(pEncoder, pReq->tbname) < 0) return -1;
-  if (tEncodeI64(pEncoder, pReq->ts) < 0) return -1;
+  if (tEncodeI64(pEncoder, pReq->startTs) < 0) return -1;
+  if (tEncodeI64(pEncoder, pReq->endTs) < 0) return -1;
   return 0;
 }
 
 int32_t tDecodeSSingleDeleteReq(SDecoder *pDecoder, SSingleDeleteReq *pReq) {
   if (tDecodeCStrTo(pDecoder, pReq->tbname) < 0) return -1;
-  if (tDecodeI64(pDecoder, &pReq->ts) < 0) return -1;
+  if (tDecodeI64(pDecoder, &pReq->startTs) < 0) return -1;
+  if (tDecodeI64(pDecoder, &pReq->endTs) < 0) return -1;
   return 0;
 }
 

source/dnode/mgmt/mgmt_dnode/src/dmHandle.c

@@ -150,7 +150,7 @@ static void dmGetServerRunStatus(SDnodeMgmt *pMgmt, SServerStatusRsp *pStatus) {
   SServerStatusRsp statusRsp = {0};
   SMonMloadInfo    minfo = {0};
   (*pMgmt->getMnodeLoadsFp)(&minfo);
-  if (minfo.isMnode && minfo.load.syncState == TAOS_SYNC_STATE_ERROR) {
+  if (minfo.isMnode && (minfo.load.syncState == TAOS_SYNC_STATE_ERROR || minfo.load.syncState == TAOS_SYNC_STATE_OFFLINE)) {
     pStatus->statusCode = TSDB_SRV_STATUS_SERVICE_DEGRADED;
     snprintf(pStatus->details, sizeof(pStatus->details), "mnode sync state is %s", syncStr(minfo.load.syncState));
     return;
@@ -160,7 +160,7 @@ static void dmGetServerRunStatus(SDnodeMgmt *pMgmt, SServerStatusRsp *pStatus) {
   (*pMgmt->getVnodeLoadsFp)(&vinfo);
   for (int32_t i = 0; i < taosArrayGetSize(vinfo.pVloads); ++i) {
     SVnodeLoad *pLoad = taosArrayGet(vinfo.pVloads, i);
-    if (pLoad->syncState == TAOS_SYNC_STATE_ERROR) {
+    if (pLoad->syncState == TAOS_SYNC_STATE_ERROR || pLoad->syncState == TAOS_SYNC_STATE_OFFLINE) {
       pStatus->statusCode = TSDB_SRV_STATUS_SERVICE_DEGRADED;
       snprintf(pStatus->details, sizeof(pStatus->details), "vnode:%d sync state is %s", pLoad->vgId,
                syncStr(pLoad->syncState));

source/dnode/mgmt/node_mgmt/src/dmTransport.c

@@ -258,8 +258,13 @@ int32_t dmInitClient(SDnode *pDnode) {
   rpcInit.parent = pDnode;
   rpcInit.rfp = rpcRfp;
   rpcInit.compressSize = tsCompressMsgSize;
+
   rpcInit.retryLimit = tsRpcRetryLimit;
   rpcInit.retryInterval = tsRpcRetryInterval;
+  rpcInit.retryMinInterval = tsRedirectPeriod;
+  rpcInit.retryStepFactor = tsRedirectFactor;
+  rpcInit.retryMaxInterval = tsRedirectMaxPeriod;
+  rpcInit.retryMaxTimouet = tsMaxRetryWaitTime;
 
   pTrans->clientRpc = rpcOpen(&rpcInit);
   if (pTrans->clientRpc == NULL) {

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

@@ -36,8 +36,6 @@ int64_t  mndGetVgroupMemory(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup);
 SArray *mndBuildDnodesArray(SMnode *, int32_t exceptDnodeId);
 int32_t mndAllocSmaVgroup(SMnode *, SDbObj *pDb, SVgObj *pVgroup);
 int32_t mndAllocVgroup(SMnode *, SDbObj *pDb, SVgObj **ppVgroups);
-int32_t mndAddVnodeToVgroup(SMnode *, SVgObj *pVgroup, SArray *pArray);
-int32_t mndRemoveVnodeFromVgroup(SMnode *, SVgObj *pVgroup, SArray *pArray, SVnodeGid *pDelVgid);
 int32_t mndAddCreateVnodeAction(SMnode *, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SVnodeGid *pVgid);
 int32_t mndAddAlterVnodeConfirmAction(SMnode *, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup);
 int32_t mndAddAlterVnodeAction(SMnode *, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, tmsg_t msgType);

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

@@ -538,7 +538,7 @@ static int32_t mndProcessSubscribeReq(SRpcMsg *pMsg) {
 
   int32_t code = -1;
   SArray *newSub = subscribe.topicNames;
-  taosArraySortString(newSub, taosArrayCompareString);
+  taosArraySort(newSub, taosArrayCompareString);
   taosArrayRemoveDuplicateP(newSub, taosArrayCompareString, taosMemoryFree);
 
   int32_t newTopicNum = taosArrayGetSize(newSub);
@@ -850,7 +850,8 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
 
     // add to current topic
     taosArrayPush(pOldConsumer->currentTopics, &addedTopic);
-    taosArraySortString(pOldConsumer->currentTopics, taosArrayCompareString);
+    taosArraySort(pOldConsumer->currentTopics, taosArrayCompareString);
+
     // set status
     if (taosArrayGetSize(pOldConsumer->rebNewTopics) == 0 && taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0) {
       if (pOldConsumer->status == MQ_CONSUMER_STATUS__MODIFY ||

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

@@ -151,10 +151,10 @@ static void mndSetVgroupOffline(SMnode *pMnode, int32_t dnodeId, int64_t curMs)
     bool roleChanged = false;
     for (int32_t vg = 0; vg < pVgroup->replica; ++vg) {
       if (pVgroup->vnodeGid[vg].dnodeId == dnodeId) {
-        if (pVgroup->vnodeGid[vg].syncState != TAOS_SYNC_STATE_ERROR) {
+        if (pVgroup->vnodeGid[vg].syncState != TAOS_SYNC_STATE_OFFLINE) {
           mInfo("vgId:%d, state changed by offline check, old state:%s restored:%d new state:error restored:0",
                 pVgroup->vgId, syncStr(pVgroup->vnodeGid[vg].syncState), pVgroup->vnodeGid[vg].syncRestore);
-          pVgroup->vnodeGid[vg].syncState = TAOS_SYNC_STATE_ERROR;
+          pVgroup->vnodeGid[vg].syncState = TAOS_SYNC_STATE_OFFLINE;
           pVgroup->vnodeGid[vg].syncRestore = 0;
           roleChanged = true;
         }
@@ -491,6 +491,15 @@ void mndPreClose(SMnode *pMnode) {
   if (pMnode != NULL) {
     syncLeaderTransfer(pMnode->syncMgmt.sync);
     syncPreStop(pMnode->syncMgmt.sync);
+
+    while (syncSnapshotRecving(pMnode->syncMgmt.sync)) {
+      mInfo("vgId:1, snapshot is recving");
+      taosMsleep(300);
+    }
+    while (syncSnapshotSending(pMnode->syncMgmt.sync)) {
+      mInfo("vgId:1, snapshot is sending");
+      taosMsleep(300);
+    }
   }
 }
 
@@ -747,7 +756,7 @@ int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgr
         tstrncpy(desc.status, "ready", sizeof(desc.status));
         pClusterInfo->vgroups_alive++;
       }
-      if (pVgid->syncState != TAOS_SYNC_STATE_ERROR) {
+      if (pVgid->syncState != TAOS_SYNC_STATE_ERROR && pVgid->syncState != TAOS_SYNC_STATE_OFFLINE) {
         pClusterInfo->vnodes_alive++;
       }
       pClusterInfo->vnodes_total++;

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

@@ -185,7 +185,7 @@ static int32_t mndMnodeActionInsert(SSdb *pSdb, SMnodeObj *pObj) {
     return -1;
   }
 
-  pObj->syncState = TAOS_SYNC_STATE_ERROR;
+  pObj->syncState = TAOS_SYNC_STATE_OFFLINE;
   mndReloadSyncConfig(pSdb->pMnode);
   return 0;
 }

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

@@ -179,6 +179,22 @@ static int32_t mndVgroupActionUpdate(SSdb *pSdb, SVgObj *pOld, SVgObj *pNew) {
   pOld->hashEnd = pNew->hashEnd;
   pOld->replica = pNew->replica;
   pOld->isTsma = pNew->isTsma;
+  for (int32_t i = 0; i < pNew->replica; ++i) {
+    SVnodeGid *pNewGid = &pNew->vnodeGid[i];
+    for (int32_t j = 0; j < pOld->replica; ++j) {
+      SVnodeGid *pOldGid = &pOld->vnodeGid[j];
+      if (pNewGid->dnodeId == pOldGid->dnodeId) {
+        pNewGid->syncState = pOldGid->syncState;
+        pNewGid->syncRestore = pOldGid->syncRestore;
+      }
+    }
+  }
+  pNew->numOfTables = pOld->numOfTables;
+  pNew->numOfTimeSeries = pOld->numOfTimeSeries;
+  pNew->totalStorage = pOld->totalStorage;
+  pNew->compStorage = pOld->compStorage;
+  pNew->pointsWritten = pOld->pointsWritten;
+  pNew->compact = pOld->compact;
   memcpy(pOld->vnodeGid, pNew->vnodeGid, TSDB_MAX_REPLICA * sizeof(SVnodeGid));
   return 0;
 }
@@ -659,11 +675,12 @@ static int32_t mndRetrieveVgroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *p
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->numOfTables, false);
 
-    // default 3 replica
-    for (int32_t i = 0; i < 3; ++i) {
+    // default 3 replica, add 1 replica if move vnode
+    for (int32_t i = 0; i < 4; ++i) {
       pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
       if (i < pVgroup->replica) {
-        colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->vnodeGid[i].dnodeId, false);
+        int16_t dnodeId = (int16_t)pVgroup->vnodeGid[i].dnodeId;
+        colDataAppend(pColInfo, numOfRows, (const char *)&dnodeId, false);
 
         bool       exist = false;
         bool       online = false;
@@ -695,16 +712,8 @@ static int32_t mndRetrieveVgroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *p
     }
 
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-    colDataAppendNULL(pColInfo, numOfRows);
-
-    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-    colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->cacheUsage, false);
-
-    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-    colDataAppendNULL(pColInfo, numOfRows);
-
-    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-    colDataAppendNULL(pColInfo, numOfRows);
+    int32_t cacheUsage = (int32_t)pVgroup->cacheUsage;
+    colDataAppend(pColInfo, numOfRows, (const char *)&cacheUsage, false);
 
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->isTsma, false);
@@ -851,7 +860,7 @@ static void mndCancelGetNextVnode(SMnode *pMnode, void *pIter) {
   sdbCancelFetch(pSdb, pIter);
 }
 
-int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
+static int32_t mndAddVnodeToVgroup(SMnode *pMnode, STrans *pTrans, SVgObj *pVgroup, SArray *pArray) {
   taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);
   for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
     SDnodeObj *pDnode = taosArrayGet(pArray, i);
@@ -887,12 +896,21 @@ int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
     }
 
     pVgid->dnodeId = pDnode->id;
-    pVgid->syncState = TAOS_SYNC_STATE_ERROR;
+    pVgid->syncState = TAOS_SYNC_STATE_OFFLINE;
     mInfo("db:%s, vgId:%d, vn:%d is added, memory:%" PRId64 ", dnode:%d avail:%" PRId64 " used:%" PRId64,
           pVgroup->dbName, pVgroup->vgId, pVgroup->replica, vgMem, pVgid->dnodeId, pDnode->memAvail, pDnode->memUsed);
 
     pVgroup->replica++;
     pDnode->numOfVnodes++;
+
+    SSdbRaw *pVgRaw = mndVgroupActionEncode(pVgroup);
+    if (pVgRaw == NULL) return -1;
+    if (mndTransAppendRedolog(pTrans, pVgRaw) != 0) {
+      sdbFreeRaw(pVgRaw);
+      return -1;
+    }
+    (void)sdbSetRawStatus(pVgRaw, SDB_STATUS_READY);
+
     return 0;
   }
 
@@ -901,7 +919,8 @@ int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
   return -1;
 }
 
-int32_t mndRemoveVnodeFromVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray, SVnodeGid *pDelVgid) {
+static int32_t mndRemoveVnodeFromVgroup(SMnode *pMnode, STrans *pTrans, SVgObj *pVgroup, SArray *pArray,
+                                        SVnodeGid *pDelVgid) {
   taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);
   for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
     SDnodeObj *pDnode = taosArrayGet(pArray, i);
@@ -941,6 +960,15 @@ _OVER:
     SVnodeGid *pVgid = &pVgroup->vnodeGid[vn];
     mInfo("db:%s, vgId:%d, vn:%d dnode:%d is reserved", pVgroup->dbName, pVgroup->vgId, vn, pVgid->dnodeId);
   }
+
+  SSdbRaw *pVgRaw = mndVgroupActionEncode(pVgroup);
+  if (pVgRaw == NULL) return -1;
+  if (mndTransAppendRedolog(pTrans, pVgRaw) != 0) {
+    sdbFreeRaw(pVgRaw);
+    return -1;
+  }
+  (void)sdbSetRawStatus(pVgRaw, SDB_STATUS_READY);
+
   return 0;
 }
 
@@ -1088,7 +1116,7 @@ int32_t mndSetMoveVgroupInfoToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb,
 
   if (!force) {
     mInfo("vgId:%d, will add 1 vnode", pVgroup->vgId);
-    if (mndAddVnodeToVgroup(pMnode, &newVg, pArray) != 0) return -1;
+    if (mndAddVnodeToVgroup(pMnode, pTrans, &newVg, pArray) != 0) return -1;
     for (int32_t i = 0; i < newVg.replica - 1; ++i) {
       if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, &newVg, newVg.vnodeGid[i].dnodeId) != 0) return -1;
     }
@@ -1100,6 +1128,16 @@ int32_t mndSetMoveVgroupInfoToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb,
     SVnodeGid del = newVg.vnodeGid[vnIndex];
     newVg.vnodeGid[vnIndex] = newVg.vnodeGid[newVg.replica];
     memset(&newVg.vnodeGid[newVg.replica], 0, sizeof(SVnodeGid));
+    {
+      SSdbRaw *pRaw = mndVgroupActionEncode(&newVg);
+      if (pRaw == NULL) return -1;
+      if (mndTransAppendRedolog(pTrans, pRaw) != 0) {
+        sdbFreeRaw(pRaw);
+        return -1;
+      }
+      (void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
+    }
+
     if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVg, &del, true) != 0) return -1;
     for (int32_t i = 0; i < newVg.replica; ++i) {
       if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, &newVg, newVg.vnodeGid[i].dnodeId) != 0) return -1;
@@ -1107,11 +1145,20 @@ int32_t mndSetMoveVgroupInfoToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb,
     if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVg) != 0) return -1;
   } else {
     mInfo("vgId:%d, will add 1 vnode and force remove 1 vnode", pVgroup->vgId);
-    if (mndAddVnodeToVgroup(pMnode, &newVg, pArray) != 0) return -1;
+    if (mndAddVnodeToVgroup(pMnode, pTrans, &newVg, pArray) != 0) return -1;
     newVg.replica--;
     SVnodeGid del = newVg.vnodeGid[vnIndex];
     newVg.vnodeGid[vnIndex] = newVg.vnodeGid[newVg.replica];
     memset(&newVg.vnodeGid[newVg.replica], 0, sizeof(SVnodeGid));
+    {
+      SSdbRaw *pRaw = mndVgroupActionEncode(&newVg);
+      if (pRaw == NULL) return -1;
+      if (mndTransAppendRedolog(pTrans, pRaw) != 0) {
+        sdbFreeRaw(pRaw);
+        return -1;
+      }
+      (void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
+    }
 
     for (int32_t i = 0; i < newVg.replica; ++i) {
       if (i != vnIndex) {
@@ -1128,16 +1175,12 @@ int32_t mndSetMoveVgroupInfoToTrans(SMnode *pMnode, STrans *pTrans, SDbObj *pDb,
 
   {
     SSdbRaw *pRaw = mndVgroupActionEncode(&newVg);
-    if (pRaw == NULL || mndTransAppendRedolog(pTrans, pRaw) != 0) return -1;
-    (void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
-    pRaw = NULL;
+    if (pRaw == NULL) return -1;
+    if (mndTransAppendCommitlog(pTrans, pRaw) != 0) {
+      sdbFreeRaw(pRaw);
+      return -1;
     }
-
-  {
-    SSdbRaw *pRaw = mndVgroupActionEncode(&newVg);
-    if (pRaw == NULL || mndTransAppendCommitlog(pTrans, pRaw) != 0) return -1;
     (void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
-    pRaw = NULL;
   }
 
   mInfo("vgId:%d, vgroup info after move, replica:%d", newVg.vgId, newVg.replica);
@@ -1193,7 +1236,15 @@ static int32_t mndAddIncVgroupReplicaToTrans(SMnode *pMnode, STrans *pTrans, SDb
   SVnodeGid *pGid = &pVgroup->vnodeGid[pVgroup->replica];
   pVgroup->replica++;
   pGid->dnodeId = newDnodeId;
-  pGid->syncState = TAOS_SYNC_STATE_ERROR;
+  pGid->syncState = TAOS_SYNC_STATE_OFFLINE;
+
+  SSdbRaw *pVgRaw = mndVgroupActionEncode(pVgroup);
+  if (pVgRaw == NULL) return -1;
+  if (mndTransAppendRedolog(pTrans, pVgRaw) != 0) {
+    sdbFreeRaw(pVgRaw);
+    return -1;
+  }
+  (void)sdbSetRawStatus(pVgRaw, SDB_STATUS_READY);
 
   for (int32_t i = 0; i < pVgroup->replica - 1; ++i) {
     if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, pVgroup, pVgroup->vnodeGid[i].dnodeId) != 0) return -1;
@@ -1224,6 +1275,14 @@ static int32_t mndAddDecVgroupReplicaFromTrans(SMnode *pMnode, STrans *pTrans, S
   memcpy(pGid, &pVgroup->vnodeGid[pVgroup->replica], sizeof(SVnodeGid));
   memset(&pVgroup->vnodeGid[pVgroup->replica], 0, sizeof(SVnodeGid));
 
+  SSdbRaw *pVgRaw = mndVgroupActionEncode(pVgroup);
+  if (pVgRaw == NULL) return -1;
+  if (mndTransAppendRedolog(pTrans, pVgRaw) != 0) {
+    sdbFreeRaw(pVgRaw);
+    return -1;
+  }
+  (void)sdbSetRawStatus(pVgRaw, SDB_STATUS_READY);
+
   if (mndAddDropVnodeAction(pMnode, pTrans, pDb, pVgroup, &delGid, true) != 0) return -1;
   for (int32_t i = 0; i < pVgroup->replica; ++i) {
     if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, pVgroup, pVgroup->vnodeGid[i].dnodeId) != 0) return -1;
@@ -1237,7 +1296,6 @@ static int32_t mndRedistributeVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb,
                                      SDnodeObj *pOld1, SDnodeObj *pNew2, SDnodeObj *pOld2, SDnodeObj *pNew3,
                                      SDnodeObj *pOld3) {
   int32_t code = -1;
-  SSdbRaw *pRaw = NULL;
   STrans *pTrans = NULL;
 
   pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, pReq, "red-vgroup");
@@ -1319,17 +1377,13 @@ static int32_t mndRedistributeVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb,
   }
 
   {
-    pRaw = mndVgroupActionEncode(&newVg);
-    if (pRaw == NULL || mndTransAppendRedolog(pTrans, pRaw) != 0) goto _OVER;
-    (void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
-    pRaw = NULL;
+    SSdbRaw *pRaw = mndVgroupActionEncode(&newVg);
+    if (pRaw == NULL) return -1;
+    if (mndTransAppendCommitlog(pTrans, pRaw) != 0) {
+      sdbFreeRaw(pRaw);
+      return -1;
     }
-
-  {
-    pRaw = mndVgroupActionEncode(&newVg);
-    if (pRaw == NULL || mndTransAppendCommitlog(pTrans, pRaw) != 0) goto _OVER;
     (void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
-    pRaw = NULL;
   }
 
   mInfo("vgId:%d, vgroup info after redistribute, replica:%d", newVg.vgId, newVg.replica);
@@ -1342,7 +1396,6 @@ static int32_t mndRedistributeVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb,
 
 _OVER:
   mndTransDrop(pTrans);
-  sdbFreeRaw(pRaw);
   mndReleaseDb(pMnode, pDb);
   return code;
 }
@@ -1593,13 +1646,13 @@ int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pOldDb
     mInfo("db:%s, vgId:%d, will add 2 vnodes, vn:0 dnode:%d", pVgroup->dbName, pVgroup->vgId,
           pVgroup->vnodeGid[0].dnodeId);
 
-    if (mndAddVnodeToVgroup(pMnode, &newVgroup, pArray) != 0) return -1;
+    if (mndAddVnodeToVgroup(pMnode, pTrans, &newVgroup, pArray) != 0) return -1;
     if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pNewDb, &newVgroup, newVgroup.vnodeGid[0].dnodeId) != 0)
       return -1;
     if (mndAddCreateVnodeAction(pMnode, pTrans, pNewDb, &newVgroup, &newVgroup.vnodeGid[1]) != 0) return -1;
     if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pNewDb, &newVgroup) != 0) return -1;
 
-    if (mndAddVnodeToVgroup(pMnode, &newVgroup, pArray) != 0) return -1;
+    if (mndAddVnodeToVgroup(pMnode, pTrans, &newVgroup, pArray) != 0) return -1;
     if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pNewDb, &newVgroup, newVgroup.vnodeGid[0].dnodeId) != 0)
       return -1;
     if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pNewDb, &newVgroup, newVgroup.vnodeGid[1].dnodeId) != 0)
@@ -1612,7 +1665,7 @@ int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pOldDb
 
     SVnodeGid del1 = {0};
     SVnodeGid del2 = {0};
-    if (mndRemoveVnodeFromVgroup(pMnode, &newVgroup, pArray, &del1) != 0) return -1;
+    if (mndRemoveVnodeFromVgroup(pMnode, pTrans, &newVgroup, pArray, &del1) != 0) return -1;
     if (mndAddDropVnodeAction(pMnode, pTrans, pNewDb, &newVgroup, &del1, true) != 0) return -1;
     if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pNewDb, &newVgroup, newVgroup.vnodeGid[0].dnodeId) != 0)
       return -1;
@@ -1620,7 +1673,7 @@ int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pOldDb
       return -1;
     if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pNewDb, &newVgroup) != 0) return -1;
 
-    if (mndRemoveVnodeFromVgroup(pMnode, &newVgroup, pArray, &del2) != 0) return -1;
+    if (mndRemoveVnodeFromVgroup(pMnode, pTrans, &newVgroup, pArray, &del2) != 0) return -1;
     if (mndAddDropVnodeAction(pMnode, pTrans, pNewDb, &newVgroup, &del2, true) != 0) return -1;
     if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pNewDb, &newVgroup, newVgroup.vnodeGid[0].dnodeId) != 0)
       return -1;
@@ -1629,16 +1682,6 @@ int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pOldDb
     return -1;
   }
 
-  {
-    SSdbRaw *pVgRaw = mndVgroupActionEncode(&newVgroup);
-    if (pVgRaw == NULL) return -1;
-    if (mndTransAppendRedolog(pTrans, pVgRaw) != 0) {
-      sdbFreeRaw(pVgRaw);
-      return -1;
-    }
-    (void)sdbSetRawStatus(pVgRaw, SDB_STATUS_READY);
-  }
-
   {
     SSdbRaw *pVgRaw = mndVgroupActionEncode(&newVgroup);
     if (pVgRaw == NULL) return -1;
@@ -1658,7 +1701,6 @@ static int32_t mndAddAdjustVnodeHashRangeAction(SMnode *pMnode, STrans *pTrans,
 
 static int32_t mndSplitVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SVgObj *pVgroup) {
   int32_t code = -1;
-  SSdbRaw *pRaw = NULL;
   STrans *pTrans = NULL;
   SArray *pArray = mndBuildDnodesArray(pMnode, 0);
 
@@ -1676,13 +1718,13 @@ static int32_t mndSplitVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SVgObj
   }
 
   if (newVg1.replica == 1) {
-    if (mndAddVnodeToVgroup(pMnode, &newVg1, pArray) != 0) goto _OVER;
+    if (mndAddVnodeToVgroup(pMnode, pTrans, &newVg1, pArray) != 0) goto _OVER;
     if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, &newVg1, newVg1.vnodeGid[0].dnodeId) != 0) goto _OVER;
     if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVg1, &newVg1.vnodeGid[1]) != 0) goto _OVER;
     if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVg1) != 0) goto _OVER;
   } else if (newVg1.replica == 3) {
     SVnodeGid del1 = {0};
-    if (mndRemoveVnodeFromVgroup(pMnode, &newVg1, pArray, &del1) != 0) goto _OVER;
+    if (mndRemoveVnodeFromVgroup(pMnode, pTrans, &newVg1, pArray, &del1) != 0) goto _OVER;
     if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVg1, &del1, true) != 0) goto _OVER;
     if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, &newVg1, newVg1.vnodeGid[0].dnodeId) != 0) goto _OVER;
     if (mndAddAlterVnodeReplicaAction(pMnode, pTrans, pDb, &newVg1, newVg1.vnodeGid[1].dnodeId) != 0) goto _OVER;
@@ -1727,17 +1769,23 @@ static int32_t mndSplitVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SVgObj
 #endif
 
   {
-    pRaw = mndVgroupActionEncode(&newVg1);
-    if (pRaw == NULL || mndTransAppendCommitlog(pTrans, pRaw) != 0) goto _OVER;
+    SSdbRaw *pRaw = mndVgroupActionEncode(&newVg1);
+    if (pRaw == NULL) return -1;
+    if (mndTransAppendCommitlog(pTrans, pRaw) != 0) {
+      sdbFreeRaw(pRaw);
+      return -1;
+    }
     (void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
-    pRaw = NULL;
   }
 
   {
-    pRaw = mndVgroupActionEncode(&newVg2);
-    if (pRaw == NULL || mndTransAppendCommitlog(pTrans, pRaw) != 0) goto _OVER;
+    SSdbRaw *pRaw = mndVgroupActionEncode(&newVg2);
+    if (pRaw == NULL) return -1;
+    if (mndTransAppendCommitlog(pTrans, pRaw) != 0) {
+      sdbFreeRaw(pRaw);
+      return -1;
+    }
     (void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
-    pRaw = NULL;
   }
 
   mInfo("vgId:%d, vgroup info after adjust hash, replica:%d hashBegin:%u hashEnd:%u vnode:0 dnode:%d", newVg1.vgId,
@@ -1757,7 +1805,6 @@ static int32_t mndSplitVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SVgObj
 _OVER:
   taosArrayDestroy(pArray);
   mndTransDrop(pTrans);
-  sdbFreeRaw(pRaw);
   return code;
 }
 
@@ -1802,16 +1849,8 @@ static int32_t mndSetBalanceVgroupInfoToTrans(SMnode *pMnode, STrans *pTrans, SD
 
   {
     SSdbRaw *pRaw = mndVgroupActionEncode(&newVg);
-    if (pRaw == NULL || mndTransAppendRedolog(pTrans, pRaw) != 0) {
-      sdbFreeRaw(pRaw);
-      return -1;
-    }
-    (void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
-  }
-
-  {
-    SSdbRaw *pRaw = mndVgroupActionEncode(&newVg);
-    if (pRaw == NULL || mndTransAppendCommitlog(pTrans, pRaw) != 0) {
+    if (pRaw == NULL) return -1;
+    if (mndTransAppendCommitlog(pTrans, pRaw) != 0) {
       sdbFreeRaw(pRaw);
       return -1;
     }

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

@@ -1127,7 +1127,7 @@ int32_t tqProcessTaskRecover1Req(STQ* pTq, SRpcMsg* pMsg) {
   SRpcMsg rpcMsg = {
       .code = 0,
       .contLen = len,
-      .msgType = TDMT_VND_STREAM_RECOVER_STEP2,
+      .msgType = TDMT_VND_STREAM_RECOVER_BLOCKING_STAGE,
       .pCont = serializedReq,
   };
 

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

@@ -21,14 +21,16 @@ int32_t tqBuildDeleteReq(SVnode* pVnode, const char* stbFullName, const SSDataBl
                          SBatchDeleteReq* deleteReq) {
   ASSERT(pDataBlock->info.type == STREAM_DELETE_RESULT);
   int32_t          totRow = pDataBlock->info.rows;
-  SColumnInfoData* pTsCol = taosArrayGet(pDataBlock->pDataBlock, START_TS_COLUMN_INDEX);
+  SColumnInfoData* pStartTsCol = taosArrayGet(pDataBlock->pDataBlock, START_TS_COLUMN_INDEX);
+  SColumnInfoData* pEndTsCol = taosArrayGet(pDataBlock->pDataBlock, END_TS_COLUMN_INDEX);
   SColumnInfoData* pGidCol = taosArrayGet(pDataBlock->pDataBlock, GROUPID_COLUMN_INDEX);
   SColumnInfoData* pTbNameCol = taosArrayGet(pDataBlock->pDataBlock, TABLE_NAME_COLUMN_INDEX);
 
   tqDebug("stream delete msg: row %d", totRow);
 
   for (int32_t row = 0; row < totRow; row++) {
-    int64_t ts = *(int64_t*)colDataGetData(pTsCol, row);
+    int64_t startTs = *(int64_t*)colDataGetData(pStartTsCol, row);
+    int64_t endTs = *(int64_t*)colDataGetData(pEndTsCol, row);
     int64_t groupId = *(int64_t*)colDataGetData(pGidCol, row);
     char*   name;
     void*   varTbName = NULL;
@@ -42,8 +44,8 @@ int32_t tqBuildDeleteReq(SVnode* pVnode, const char* stbFullName, const SSDataBl
     } else {
       name = buildCtbNameByGroupId(stbFullName, groupId);
     }
-    tqDebug("stream delete msg: vgId:%d, groupId :%" PRId64 ", name: %s, ts:%" PRId64, pVnode->config.vgId, groupId,
-            name, ts);
+    tqDebug("stream delete msg: vgId:%d, groupId :%" PRId64 ", name: %s, start ts:%" PRId64 "end ts:%" PRId64,
+            pVnode->config.vgId, groupId, name, startTs, endTs);
 #if 0
     SMetaReader mr = {0};
     metaReaderInit(&mr, pVnode->pMeta, 0);
@@ -59,7 +61,8 @@ int32_t tqBuildDeleteReq(SVnode* pVnode, const char* stbFullName, const SSDataBl
     taosMemoryFree(name);
 #endif
     SSingleDeleteReq req = {
-        .ts = ts,
+        .startTs = startTs,
+        .endTs = endTs,
     };
     strncpy(req.tbname, name, TSDB_TABLE_NAME_LEN);
     taosMemoryFree(name);

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

@@ -775,7 +775,10 @@ static void doCopyColVal(SColumnInfoData* pColInfoData, int32_t rowIndex, int32_
     } else {
       varDataSetLen(pSup->buildBuf[colIndex], pColVal->value.nData);
       ASSERT(pColVal->value.nData <= pColInfoData->info.bytes);
+      if (pColVal->value.nData > 0) {  // pData may be null, if nData is 0
         memcpy(varDataVal(pSup->buildBuf[colIndex]), pColVal->value.pData, pColVal->value.nData);
+      }
+
       colDataAppend(pColInfoData, rowIndex, pSup->buildBuf[colIndex], false);
     }
   } else {
@@ -2540,6 +2543,7 @@ int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader*
       goto _err;
     }
 
+    // TODO: opt the perf of read del index
     code = tsdbReadDelIdx(pDelFReader, aDelIdx);
     if (code != TSDB_CODE_SUCCESS) {
       taosArrayDestroy(aDelIdx);

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

@@ -271,7 +271,7 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp
         goto _err;
       }
     } break;
-    case TDMT_VND_STREAM_RECOVER_STEP2: {
+    case TDMT_VND_STREAM_RECOVER_BLOCKING_STAGE: {
       if (tqProcessTaskRecover2Req(pVnode->pTq, version, pMsg->pCont, pMsg->contLen) < 0) {
         goto _err;
       }
@@ -410,7 +410,7 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
       return tqProcessTaskRetrieveReq(pVnode->pTq, pMsg);
     case TDMT_STREAM_RETRIEVE_RSP:
       return tqProcessTaskRetrieveRsp(pVnode->pTq, pMsg);
-    case TDMT_VND_STREAM_RECOVER_STEP1:
+    case TDMT_VND_STREAM_RECOVER_NONBLOCKING_STAGE:
       return tqProcessTaskRecover1Req(pVnode->pTq, pMsg);
     case TDMT_STREAM_RECOVER_FINISH:
       return tqProcessTaskRecoverFinishReq(pVnode->pTq, pMsg);
@@ -1192,11 +1192,11 @@ static int32_t vnodeProcessBatchDeleteReq(SVnode *pVnode, int64_t version, void
 
     int64_t uid = mr.me.uid;
 
-    int32_t code = tsdbDeleteTableData(pVnode->pTsdb, version, deleteReq.suid, uid, pOneReq->ts, pOneReq->ts);
+    int32_t code = tsdbDeleteTableData(pVnode->pTsdb, version, deleteReq.suid, uid, pOneReq->startTs, pOneReq->endTs);
     if (code < 0) {
       terrno = code;
       vError("vgId:%d, delete error since %s, suid:%" PRId64 ", uid:%" PRId64 ", start ts:%" PRId64 ", end ts:%" PRId64,
-             TD_VID(pVnode), terrstr(), deleteReq.suid, uid, pOneReq->ts, pOneReq->ts);
+             TD_VID(pVnode), terrstr(), deleteReq.suid, uid, pOneReq->startTs, pOneReq->endTs);
     }
 
     tDecoderClear(&mr.coder);

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

@@ -66,7 +66,7 @@ void vnodeRedirectRpcMsg(SVnode *pVnode, SRpcMsg *pMsg) {
   }
   pMsg->info.hasEpSet = 1;
 
-  SRpcMsg rsp = {.code = TSDB_CODE_RPC_REDIRECT, .info = pMsg->info, .msgType = pMsg->msgType + 1};
+  SRpcMsg rsp = {.code = TSDB_CODE_SYN_NOT_LEADER, .info = pMsg->info, .msgType = pMsg->msgType + 1};
   tmsgSendRedirectRsp(&rsp, &newEpSet);
 }
 
@@ -514,6 +514,16 @@ void vnodeSyncPreClose(SVnode *pVnode) {
   vInfo("vgId:%d, pre close sync", pVnode->config.vgId);
   syncLeaderTransfer(pVnode->sync);
   syncPreStop(pVnode->sync);
+
+  while (syncSnapshotRecving(pVnode->sync)) {
+    vInfo("vgId:%d, snapshot is recving", pVnode->config.vgId);
+    taosMsleep(300);
+  }
+  while (syncSnapshotSending(pVnode->sync)) {
+    vInfo("vgId:%d, snapshot is sending", pVnode->config.vgId);
+    taosMsleep(300);
+  }
+
   taosThreadMutexLock(&pVnode->lock);
   if (pVnode->blocked) {
     vInfo("vgId:%d, post block after close sync", pVnode->config.vgId);

source/libs/executor/inc/executil.h

@@ -161,4 +161,6 @@ int32_t convertFillType(int32_t mode);
 int32_t resultrowComparAsc(const void* p1, const void* p2);
 int32_t isQualifiedTable(STableKeyInfo* info, SNode* pTagCond, void* metaHandle, bool* pQualified);
 
+void printDataBlock(SSDataBlock* pBlock, const char* flag);
+
 #endif  // TDENGINE_QUERYUTIL_H

source/libs/executor/inc/executorimpl.h

@@ -332,6 +332,7 @@ typedef struct STableScanInfo {
   int32_t                currentTable;
   int8_t                 scanMode;
   int8_t                 assignBlockUid;
+  bool                   hasGroupByTag;
 } STableScanInfo;
 
 typedef struct STableMergeScanInfo {
@@ -537,23 +538,6 @@ typedef struct SStreamIntervalOperatorInfo {
   SWinKey            delKey;
 } SStreamIntervalOperatorInfo;
 
-typedef struct SFillOperatorInfo {
-  struct SFillInfo* pFillInfo;
-  SSDataBlock*      pRes;
-  SSDataBlock*      pFinalRes;
-  int64_t           totalInputRows;
-  void**            p;
-  SSDataBlock*      existNewGroupBlock;
-  STimeWindow       win;
-  SColMatchInfo     matchInfo;
-  int32_t           primaryTsCol;
-  int32_t           primarySrcSlotId;
-  uint64_t          curGroupId;  // current handled group id
-  SExprInfo*        pExprInfo;
-  int32_t           numOfExpr;
-  SExprSupp         noFillExprSupp;
-} SFillOperatorInfo;
-
 typedef struct SDataGroupInfo {
   uint64_t groupId;
   int64_t  numOfRows;
@@ -805,8 +789,6 @@ int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SRead
 int32_t createDataSinkParam(SDataSinkNode* pNode, void** pParam, qTaskInfo_t* pTaskInfo, SReadHandle* readHandle);
 int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList);
 
-void    printTaskExecCostInLog(SExecTaskInfo* pTaskInfo);
-
 int32_t getMaximumIdleDurationSec();
 
 STimeWindow getActiveTimeWindow(SDiskbasedBuf* pBuf, SResultRowInfo* pResultRowInfo, int64_t ts, SInterval* pInterval,
@@ -824,7 +806,6 @@ bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup);
 bool isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SStreamState* pState, STimeWindowAggSupp* pTwSup);
 void appendOneRowToStreamSpecialBlock(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid,
                                       uint64_t* pGp, void* pTbName);
-void printDataBlock(SSDataBlock* pBlock, const char* flag);
 uint64_t    calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId);
 void        calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock);
 

source/libs/executor/inc/tfill.h

@@ -25,6 +25,8 @@ extern "C" {
 #include "tcommon.h"
 #include "tsimplehash.h"
 
+#define GET_DEST_SLOT_ID(_p) ((_p)->pExpr->base.resSchema.slotId)
+
 struct SSDataBlock;
 
 typedef struct SFillColInfo {
@@ -116,7 +118,7 @@ int64_t getNumOfResultsAfterFillGap(SFillInfo* pFillInfo, int64_t ekey, int32_t
 void          taosFillSetStartInfo(struct SFillInfo* pFillInfo, int32_t numOfRows, TSKEY endKey);
 void          taosResetFillInfo(struct SFillInfo* pFillInfo, TSKEY startTimestamp);
 void          taosFillSetInputDataBlock(struct SFillInfo* pFillInfo, const struct SSDataBlock* pInput);
-struct SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfFillExpr, SExprInfo* pNotFillExpr,
+SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfFillExpr, SExprInfo* pNotFillExpr,
                                 int32_t numOfNotFillCols, const struct SNodeListNode* val);
 bool          taosFillHasMoreResults(struct SFillInfo* pFillInfo);
 
@@ -128,6 +130,8 @@ void*   taosDestroyFillInfo(struct SFillInfo* pFillInfo);
 int64_t taosFillResultDataBlock(struct SFillInfo* pFillInfo, SSDataBlock* p, int32_t capacity);
 int64_t getFillInfoStart(struct SFillInfo* pFillInfo);
 
+bool fillIfWindowPseudoColumn(SFillInfo* pFillInfo, SFillColInfo* pCol, SColumnInfoData* pDstColInfoData,
+                                     int32_t rowIndex);
 #ifdef __cplusplus
 }
 #endif

source/libs/executor/src/exchangeoperator.c

@@ -15,26 +15,15 @@
 
 #include "filter.h"
 #include "function.h"
-#include "functionMgt.h"
 #include "os.h"
-#include "querynodes.h"
-#include "tfill.h"
 #include "tname.h"
 #include "tref.h"
-
 #include "tdatablock.h"
-#include "tglobal.h"
 #include "tmsg.h"
-#include "tsort.h"
-#include "ttime.h"
-
 #include "executorimpl.h"
 #include "index.h"
 #include "query.h"
-#include "tcompare.h"
 #include "thash.h"
-#include "ttypes.h"
-#include "vnode.h"
 
 typedef struct SFetchRspHandleWrapper {
   uint32_t exchangeId;

source/libs/executor/src/executil.c

@@ -2002,3 +2002,13 @@ int32_t createScanTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags
 
   return TSDB_CODE_SUCCESS;
 }
+
+void printDataBlock(SSDataBlock* pBlock, const char* flag) {
+  if (!pBlock || pBlock->info.rows == 0) {
+    qDebug("===stream===printDataBlock: Block is Null or Empty");
+    return;
+  }
+  char* pBuf = NULL;
+  qDebug("%s", dumpBlockData(pBlock, flag, &pBuf));
+  taosMemoryFree(pBuf);
+}
\ No newline at end of file

source/libs/executor/src/executor.c

@@ -704,6 +704,20 @@ int32_t qAsyncKillTask(qTaskInfo_t qinfo) {
   return TSDB_CODE_SUCCESS;
 }
 
+static void printTaskExecCostInLog(SExecTaskInfo* pTaskInfo) {
+  STaskCostInfo* pSummary = &pTaskInfo->cost;
+
+  SFileBlockLoadRecorder* pRecorder = pSummary->pRecoder;
+  if (pSummary->pRecoder != NULL) {
+    qDebug(
+        "%s :cost summary: elapsed time:%.2f ms, extract tableList:%.2f ms, createGroupIdMap:%.2f ms, total blocks:%d, "
+        "load block SMA:%d, load data block:%d, total rows:%" PRId64 ", check rows:%" PRId64,
+        GET_TASKID(pTaskInfo), pSummary->elapsedTime / 1000.0, pSummary->extractListTime, pSummary->groupIdMapTime,
+        pRecorder->totalBlocks, pRecorder->loadBlockStatis, pRecorder->loadBlocks, pRecorder->totalRows,
+        pRecorder->totalCheckedRows);
+  }
+}
+
 void qDestroyTask(qTaskInfo_t qTaskHandle) {
   SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)qTaskHandle;
   if (pTaskInfo == NULL) {

source/libs/executor/src/executorimpl.c

@@ -91,7 +91,6 @@ static void setBlockSMAInfo(SqlFunctionCtx* pCtx, SExprInfo* pExpr, SSDataBlock*
 
 static void releaseQueryBuf(size_t numOfTables);
 
-static void    destroyFillOperatorInfo(void* param);
 static void    destroyAggOperatorInfo(void* param);
 static void    initCtxOutputBuffer(SqlFunctionCtx* pCtx, int32_t size);
 static void    doSetTableGroupOutputBuf(SOperatorInfo* pOperator, int32_t numOfOutput, uint64_t groupId);
@@ -1157,20 +1156,6 @@ void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SG
   }
 }
 
-void printTaskExecCostInLog(SExecTaskInfo* pTaskInfo) {
-  STaskCostInfo* pSummary = &pTaskInfo->cost;
-
-  SFileBlockLoadRecorder* pRecorder = pSummary->pRecoder;
-  if (pSummary->pRecoder != NULL) {
-    qDebug(
-        "%s :cost summary: elapsed time:%.2f ms, extract tableList:%.2f ms, createGroupIdMap:%.2f ms, total blocks:%d, "
-        "load block SMA:%d, load data block:%d, total rows:%" PRId64 ", check rows:%" PRId64,
-        GET_TASKID(pTaskInfo), pSummary->elapsedTime / 1000.0, pSummary->extractListTime, pSummary->groupIdMapTime,
-        pRecorder->totalBlocks, pRecorder->loadBlockStatis, pRecorder->loadBlocks, pRecorder->totalRows,
-        pRecorder->totalCheckedRows);
-  }
-}
-
 // void skipBlocks(STaskRuntimeEnv *pRuntimeEnv) {
 //   STaskAttr *pQueryAttr = pRuntimeEnv->pQueryAttr;
 //
@@ -1419,6 +1404,78 @@ int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scan
   }
 }
 
+static int32_t createDataBlockForEmptyInput(SOperatorInfo* pOperator, SSDataBlock **ppBlock) {
+  if (!tsCountAlwaysReturnValue) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SOperatorInfo* downstream = pOperator->pDownstream[0];
+  if (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_PARTITION ||
+      (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN &&
+       ((STableScanInfo *)downstream->info)->hasGroupByTag == true)) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx;
+  bool hasCountFunc = false;
+  for (int32_t i = 0; i < pOperator->exprSupp.numOfExprs; ++i) {
+    if ((strcmp(pCtx[i].pExpr->pExpr->_function.functionName, "count") == 0) ||
+        (strcmp(pCtx[i].pExpr->pExpr->_function.functionName, "hyperloglog") == 0) ||
+        (strcmp(pCtx[i].pExpr->pExpr->_function.functionName, "_hyperloglog_partial") == 0) ||
+        (strcmp(pCtx[i].pExpr->pExpr->_function.functionName, "_hyperloglog_merge") == 0)) {
+      hasCountFunc = true;
+      break;
+    }
+  }
+
+  if (!hasCountFunc) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SSDataBlock* pBlock = createDataBlock();
+  pBlock->info.rows = 1;
+  pBlock->info.capacity = 0;
+
+  for (int32_t i = 0; i < pOperator->exprSupp.numOfExprs; ++i) {
+    SColumnInfoData colInfo = {0};
+    colInfo.hasNull = true;
+    colInfo.info.type = TSDB_DATA_TYPE_NULL;
+    colInfo.info.bytes = 1;
+
+    SExprInfo* pOneExpr = &pOperator->exprSupp.pExprInfo[i];
+    for (int32_t j = 0; j < pOneExpr->base.numOfParams; ++j) {
+      SFunctParam* pFuncParam = &pOneExpr->base.pParam[j];
+      if (pFuncParam->type == FUNC_PARAM_TYPE_COLUMN) {
+        int32_t slotId = pFuncParam->pCol->slotId;
+        int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
+        if (slotId >= numOfCols) {
+          taosArrayEnsureCap(pBlock->pDataBlock, slotId + 1);
+          for (int32_t k = numOfCols; k < slotId + 1; ++k) {
+            taosArrayPush(pBlock->pDataBlock, &colInfo);
+          }
+        }
+      } else if (pFuncParam->type == FUNC_PARAM_TYPE_VALUE) {
+        // do nothing
+      }
+    }
+  }
+
+  blockDataEnsureCapacity(pBlock, pBlock->info.rows);
+  *ppBlock = pBlock;
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static void destroyDataBlockForEmptyInput(bool blockAllocated, SSDataBlock **ppBlock) {
+  if (!blockAllocated) {
+    return;
+  }
+
+  blockDataDestroy(*ppBlock);
+  *ppBlock = NULL;
+}
+
+
 // this is a blocking operator
 static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
   if (OPTR_IS_OPENED(pOperator)) {
@@ -1436,22 +1493,36 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
   int32_t order = TSDB_ORDER_ASC;
   int32_t scanFlag = MAIN_SCAN;
 
+  bool    hasValidBlock = false;
+  bool    blockAllocated = false;
+
   while (1) {
     SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
     if (pBlock == NULL) {
+      if (!hasValidBlock) {
+        createDataBlockForEmptyInput(pOperator, &pBlock);
+        if (pBlock == NULL) {
+          break;
+        }
+        blockAllocated = true;
+      } else {
         break;
       }
+    }
+    hasValidBlock = true;
 
     int32_t code = getTableScanInfo(pOperator, &order, &scanFlag);
     if (code != TSDB_CODE_SUCCESS) {
+      destroyDataBlockForEmptyInput(blockAllocated, &pBlock);
       T_LONG_JMP(pTaskInfo->env, code);
     }
 
     // there is an scalar expression that needs to be calculated before apply the group aggregation.
-    if (pAggInfo->scalarExprSup.pExprInfo != NULL) {
+    if (pAggInfo->scalarExprSup.pExprInfo != NULL && !blockAllocated) {
       SExprSupp* pSup1 = &pAggInfo->scalarExprSup;
       code = projectApplyFunctions(pSup1->pExprInfo, pBlock, pBlock, pSup1->pCtx, pSup1->numOfExprs, NULL);
       if (code != TSDB_CODE_SUCCESS) {
+        destroyDataBlockForEmptyInput(blockAllocated, &pBlock);
         T_LONG_JMP(pTaskInfo->env, code);
       }
     }
@@ -1461,8 +1532,12 @@ static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
     setInputDataBlock(pSup, pBlock, order, scanFlag, true);
     code = doAggregateImpl(pOperator, pSup->pCtx);
     if (code != 0) {
+      destroyDataBlockForEmptyInput(blockAllocated, &pBlock);
       T_LONG_JMP(pTaskInfo->env, code);
     }
+
+    destroyDataBlockForEmptyInput(blockAllocated, &pBlock);
+
   }
 
   // the downstream operator may return with error code, so let's check the code before generating results.
@@ -1513,179 +1588,6 @@ static SSDataBlock* getAggregateResult(SOperatorInfo* pOperator) {
   return (rows == 0) ? NULL : pInfo->pRes;
 }
 
-static void doHandleRemainBlockForNewGroupImpl(SOperatorInfo* pOperator, SFillOperatorInfo* pInfo,
-                                               SResultInfo* pResultInfo, SExecTaskInfo* pTaskInfo) {
-  pInfo->totalInputRows = pInfo->existNewGroupBlock->info.rows;
-  SSDataBlock* pResBlock = pInfo->pFinalRes;
-
-  int32_t order = TSDB_ORDER_ASC;
-  int32_t scanFlag = MAIN_SCAN;
-  getTableScanInfo(pOperator, &order, &scanFlag);
-
-  int64_t ekey = pInfo->existNewGroupBlock->info.window.ekey;
-  taosResetFillInfo(pInfo->pFillInfo, getFillInfoStart(pInfo->pFillInfo));
-
-  blockDataCleanup(pInfo->pRes);
-  doApplyScalarCalculation(pOperator, pInfo->existNewGroupBlock, order, scanFlag);
-
-  taosFillSetStartInfo(pInfo->pFillInfo, pInfo->pRes->info.rows, ekey);
-  taosFillSetInputDataBlock(pInfo->pFillInfo, pInfo->pRes);
-
-  int32_t numOfResultRows = pResultInfo->capacity - pResBlock->info.rows;
-  taosFillResultDataBlock(pInfo->pFillInfo, pResBlock, numOfResultRows);
-
-  pInfo->curGroupId = pInfo->existNewGroupBlock->info.id.groupId;
-  pInfo->existNewGroupBlock = NULL;
-}
-
-static void doHandleRemainBlockFromNewGroup(SOperatorInfo* pOperator, SFillOperatorInfo* pInfo,
-                                            SResultInfo* pResultInfo, SExecTaskInfo* pTaskInfo) {
-  if (taosFillHasMoreResults(pInfo->pFillInfo)) {
-    int32_t numOfResultRows = pResultInfo->capacity - pInfo->pFinalRes->info.rows;
-    taosFillResultDataBlock(pInfo->pFillInfo, pInfo->pFinalRes, numOfResultRows);
-    pInfo->pRes->info.id.groupId = pInfo->curGroupId;
-    return;
-  }
-
-  // handle the cached new group data block
-  if (pInfo->existNewGroupBlock) {
-    doHandleRemainBlockForNewGroupImpl(pOperator, pInfo, pResultInfo, pTaskInfo);
-  }
-}
-
-static void doApplyScalarCalculation(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t order, int32_t scanFlag) {
-  SFillOperatorInfo* pInfo = pOperator->info;
-  SExprSupp*         pSup = &pOperator->exprSupp;
-  setInputDataBlock(pSup, pBlock, order, scanFlag, false);
-  projectApplyFunctions(pSup->pExprInfo, pInfo->pRes, pBlock, pSup->pCtx, pSup->numOfExprs, NULL);
-
-  // reset the row value before applying the no-fill functions to the input data block, which is "pBlock" in this case.
-  pInfo->pRes->info.rows = 0;
-  SExprSupp* pNoFillSupp = &pInfo->noFillExprSupp;
-  setInputDataBlock(pNoFillSupp, pBlock, order, scanFlag, false);
-
-  projectApplyFunctions(pNoFillSupp->pExprInfo, pInfo->pRes, pBlock, pNoFillSupp->pCtx, pNoFillSupp->numOfExprs, NULL);
-  pInfo->pRes->info.id.groupId = pBlock->info.id.groupId;
-}
-
-static SSDataBlock* doFillImpl(SOperatorInfo* pOperator) {
-  SFillOperatorInfo* pInfo = pOperator->info;
-  SExecTaskInfo*     pTaskInfo = pOperator->pTaskInfo;
-
-  SResultInfo* pResultInfo = &pOperator->resultInfo;
-  SSDataBlock* pResBlock = pInfo->pFinalRes;
-
-  blockDataCleanup(pResBlock);
-
-  int32_t order = TSDB_ORDER_ASC;
-  int32_t scanFlag = MAIN_SCAN;
-  getTableScanInfo(pOperator, &order, &scanFlag);
-
-  doHandleRemainBlockFromNewGroup(pOperator, pInfo, pResultInfo, pTaskInfo);
-  if (pResBlock->info.rows > 0) {
-    pResBlock->info.id.groupId = pInfo->curGroupId;
-    return pResBlock;
-  }
-
-  SOperatorInfo* pDownstream = pOperator->pDownstream[0];
-  while (1) {
-    SSDataBlock* pBlock = pDownstream->fpSet.getNextFn(pDownstream);
-    if (pBlock == NULL) {
-      if (pInfo->totalInputRows == 0) {
-        setOperatorCompleted(pOperator);
-        return NULL;
-      }
-
-      taosFillSetStartInfo(pInfo->pFillInfo, 0, pInfo->win.ekey);
-    } else {
-      blockDataUpdateTsWindow(pBlock, pInfo->primarySrcSlotId);
-
-      blockDataCleanup(pInfo->pRes);
-      blockDataEnsureCapacity(pInfo->pRes, pBlock->info.rows);
-      blockDataEnsureCapacity(pInfo->pFinalRes, pBlock->info.rows);
-      doApplyScalarCalculation(pOperator, pBlock, order, scanFlag);
-
-      if (pInfo->curGroupId == 0 || pInfo->curGroupId == pInfo->pRes->info.id.groupId) {
-        pInfo->curGroupId = pInfo->pRes->info.id.groupId;  // the first data block
-        pInfo->totalInputRows += pInfo->pRes->info.rows;
-
-        if (order == pInfo->pFillInfo->order) {
-          taosFillSetStartInfo(pInfo->pFillInfo, pInfo->pRes->info.rows, pBlock->info.window.ekey);
-        } else {
-          taosFillSetStartInfo(pInfo->pFillInfo, pInfo->pRes->info.rows, pBlock->info.window.skey);
-        }
-        taosFillSetInputDataBlock(pInfo->pFillInfo, pInfo->pRes);
-      } else if (pInfo->curGroupId != pBlock->info.id.groupId) {  // the new group data block
-        pInfo->existNewGroupBlock = pBlock;
-
-        // Fill the previous group data block, before handle the data block of new group.
-        // Close the fill operation for previous group data block
-        taosFillSetStartInfo(pInfo->pFillInfo, 0, pInfo->win.ekey);
-      }
-    }
-
-    int32_t numOfResultRows = pOperator->resultInfo.capacity - pResBlock->info.rows;
-    taosFillResultDataBlock(pInfo->pFillInfo, pResBlock, numOfResultRows);
-
-    // current group has no more result to return
-    if (pResBlock->info.rows > 0) {
-      // 1. The result in current group not reach the threshold of output result, continue
-      // 2. If multiple group results existing in one SSDataBlock is not allowed, return immediately
-      if (pResBlock->info.rows > pResultInfo->threshold || pBlock == NULL || pInfo->existNewGroupBlock != NULL) {
-        pResBlock->info.id.groupId = pInfo->curGroupId;
-        return pResBlock;
-      }
-
-      doHandleRemainBlockFromNewGroup(pOperator, pInfo, pResultInfo, pTaskInfo);
-      if (pResBlock->info.rows >= pOperator->resultInfo.threshold || pBlock == NULL) {
-        pResBlock->info.id.groupId = pInfo->curGroupId;
-        return pResBlock;
-      }
-    } else if (pInfo->existNewGroupBlock) {  // try next group
-      assert(pBlock != NULL);
-
-      blockDataCleanup(pResBlock);
-
-      doHandleRemainBlockForNewGroupImpl(pOperator, pInfo, pResultInfo, pTaskInfo);
-      if (pResBlock->info.rows > pResultInfo->threshold) {
-        pResBlock->info.id.groupId = pInfo->curGroupId;
-        return pResBlock;
-      }
-    } else {
-      return NULL;
-    }
-  }
-}
-
-static SSDataBlock* doFill(SOperatorInfo* pOperator) {
-  SFillOperatorInfo* pInfo = pOperator->info;
-  SExecTaskInfo*     pTaskInfo = pOperator->pTaskInfo;
-
-  if (pOperator->status == OP_EXEC_DONE) {
-    return NULL;
-  }
-
-  SSDataBlock* fillResult = NULL;
-  while (true) {
-    fillResult = doFillImpl(pOperator);
-    if (fillResult == NULL) {
-      setOperatorCompleted(pOperator);
-      break;
-    }
-
-    doFilter(fillResult, pOperator->exprSupp.pFilterInfo, &pInfo->matchInfo);
-    if (fillResult->info.rows > 0) {
-      break;
-    }
-  }
-
-  if (fillResult != NULL) {
-    pOperator->resultInfo.totalRows += fillResult->info.rows;
-  }
-
-  return fillResult;
-}
-
 void destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs) {
   for (int32_t i = 0; i < numOfExprs; ++i) {
     SExprInfo* pExprInfo = &pExpr[i];
@@ -1955,167 +1857,6 @@ void destroyAggOperatorInfo(void* param) {
   taosMemoryFreeClear(param);
 }
 
-void destroyFillOperatorInfo(void* param) {
-  SFillOperatorInfo* pInfo = (SFillOperatorInfo*)param;
-  pInfo->pFillInfo = taosDestroyFillInfo(pInfo->pFillInfo);
-  pInfo->pRes = blockDataDestroy(pInfo->pRes);
-  pInfo->pFinalRes = blockDataDestroy(pInfo->pFinalRes);
-
-  cleanupExprSupp(&pInfo->noFillExprSupp);
-
-  taosMemoryFreeClear(pInfo->p);
-  taosArrayDestroy(pInfo->matchInfo.pList);
-  taosMemoryFreeClear(param);
-}
-
-static int32_t initFillInfo(SFillOperatorInfo* pInfo, SExprInfo* pExpr, int32_t numOfCols, SExprInfo* pNotFillExpr,
-                            int32_t numOfNotFillCols, SNodeListNode* pValNode, STimeWindow win, int32_t capacity,
-                            const char* id, SInterval* pInterval, int32_t fillType, int32_t order) {
-  SFillColInfo* pColInfo = createFillColInfo(pExpr, numOfCols, pNotFillExpr, numOfNotFillCols, pValNode);
-
-  int64_t     startKey = (order == TSDB_ORDER_ASC) ? win.skey : win.ekey;
-  STimeWindow w = getAlignQueryTimeWindow(pInterval, pInterval->precision, startKey);
-  w = getFirstQualifiedTimeWindow(startKey, &w, pInterval, order);
-
-  pInfo->pFillInfo = taosCreateFillInfo(w.skey, numOfCols, numOfNotFillCols, capacity, pInterval, fillType, pColInfo,
-                                        pInfo->primaryTsCol, order, id);
-
-  if (order == TSDB_ORDER_ASC) {
-    pInfo->win.skey = win.skey;
-    pInfo->win.ekey = win.ekey;
-  } else {
-    pInfo->win.skey = win.ekey;
-    pInfo->win.ekey = win.skey;
-  }
-  pInfo->p = taosMemoryCalloc(numOfCols, POINTER_BYTES);
-
-  if (pInfo->pFillInfo == NULL || pInfo->p == NULL) {
-    taosMemoryFree(pInfo->pFillInfo);
-    taosMemoryFree(pInfo->p);
-    return TSDB_CODE_OUT_OF_MEMORY;
-  } else {
-    return TSDB_CODE_SUCCESS;
-  }
-}
-
-static bool isWstartColumnExist(SFillOperatorInfo* pInfo) {
-  if (pInfo->noFillExprSupp.numOfExprs == 0) {
-    return false;
-  }
-
-  for (int32_t i = 0; i < pInfo->noFillExprSupp.numOfExprs; ++i) {
-    SExprInfo* exprInfo = pInfo->noFillExprSupp.pExprInfo + i;
-    if (exprInfo->pExpr->nodeType == QUERY_NODE_COLUMN && exprInfo->base.numOfParams == 1 &&
-        exprInfo->base.pParam[0].pCol->colType == COLUMN_TYPE_WINDOW_START) {
-      return true;
-    }
-  }
-  return false;
-}
-
-static int32_t createPrimaryTsExprIfNeeded(SFillOperatorInfo* pInfo, SFillPhysiNode* pPhyFillNode, SExprSupp* pExprSupp,
-                                           const char* idStr) {
-  bool wstartExist = isWstartColumnExist(pInfo);
-
-  if (wstartExist == false) {
-    if (pPhyFillNode->pWStartTs->type != QUERY_NODE_TARGET) {
-      qError("pWStartTs of fill physical node is not a target node, %s", idStr);
-      return TSDB_CODE_QRY_SYS_ERROR;
-    }
-
-    SExprInfo* pExpr = taosMemoryRealloc(pExprSupp->pExprInfo, (pExprSupp->numOfExprs + 1) * sizeof(SExprInfo));
-    if (pExpr == NULL) {
-      return TSDB_CODE_OUT_OF_MEMORY;
-    }
-
-    createExprFromTargetNode(&pExpr[pExprSupp->numOfExprs], (STargetNode*)pPhyFillNode->pWStartTs);
-    pExprSupp->numOfExprs += 1;
-    pExprSupp->pExprInfo = pExpr;
-  }
-
-  return TSDB_CODE_SUCCESS;
-}
-
-SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pPhyFillNode,
-                                      SExecTaskInfo* pTaskInfo) {
-  SFillOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SFillOperatorInfo));
-  SOperatorInfo*     pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
-  if (pInfo == NULL || pOperator == NULL) {
-    goto _error;
-  }
-
-  pInfo->pRes = createDataBlockFromDescNode(pPhyFillNode->node.pOutputDataBlockDesc);
-  SExprInfo* pExprInfo = createExprInfo(pPhyFillNode->pFillExprs, NULL, &pInfo->numOfExpr);
-  pOperator->exprSupp.pExprInfo = pExprInfo;
-
-  SExprSupp* pNoFillSupp = &pInfo->noFillExprSupp;
-  pNoFillSupp->pExprInfo = createExprInfo(pPhyFillNode->pNotFillExprs, NULL, &pNoFillSupp->numOfExprs);
-  int32_t code = createPrimaryTsExprIfNeeded(pInfo, pPhyFillNode, pNoFillSupp, pTaskInfo->id.str);
-  if (code != TSDB_CODE_SUCCESS) {
-    goto _error;
-  }
-
-  code = initExprSupp(pNoFillSupp, pNoFillSupp->pExprInfo, pNoFillSupp->numOfExprs);
-  if (code != TSDB_CODE_SUCCESS) {
-    goto _error;
-  }
-
-  SInterval* pInterval =
-      QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL == downstream->operatorType
-          ? &((SMergeAlignedIntervalAggOperatorInfo*)downstream->info)->intervalAggOperatorInfo->interval
-          : &((SIntervalAggOperatorInfo*)downstream->info)->interval;
-
-  int32_t order = (pPhyFillNode->inputTsOrder == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
-  int32_t type = convertFillType(pPhyFillNode->mode);
-
-  SResultInfo* pResultInfo = &pOperator->resultInfo;
-
-  initResultSizeInfo(&pOperator->resultInfo, 4096);
-  blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
-  code = initExprSupp(&pOperator->exprSupp, pExprInfo, pInfo->numOfExpr);
-  if (code != TSDB_CODE_SUCCESS) {
-    goto _error;
-  }
-
-  pInfo->primaryTsCol = ((STargetNode*)pPhyFillNode->pWStartTs)->slotId;
-  pInfo->primarySrcSlotId = ((SColumnNode*)((STargetNode*)pPhyFillNode->pWStartTs)->pExpr)->slotId;
-
-  int32_t numOfOutputCols = 0;
-  code = extractColMatchInfo(pPhyFillNode->pFillExprs, pPhyFillNode->node.pOutputDataBlockDesc, &numOfOutputCols,
-                             COL_MATCH_FROM_SLOT_ID, &pInfo->matchInfo);
-
-  code = initFillInfo(pInfo, pExprInfo, pInfo->numOfExpr, pNoFillSupp->pExprInfo, pNoFillSupp->numOfExprs,
-                      (SNodeListNode*)pPhyFillNode->pValues, pPhyFillNode->timeRange, pResultInfo->capacity,
-                      pTaskInfo->id.str, pInterval, type, order);
-  if (code != TSDB_CODE_SUCCESS) {
-    goto _error;
-  }
-
-  pInfo->pFinalRes = createOneDataBlock(pInfo->pRes, false);
-  blockDataEnsureCapacity(pInfo->pFinalRes, pOperator->resultInfo.capacity);
-
-  code = filterInitFromNode((SNode*)pPhyFillNode->node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
-  if (code != TSDB_CODE_SUCCESS) {
-    goto _error;
-  }
-
-  setOperatorInfo(pOperator, "FillOperator", QUERY_NODE_PHYSICAL_PLAN_FILL, false, OP_NOT_OPENED, pInfo, pTaskInfo);
-  pOperator->exprSupp.numOfExprs = pInfo->numOfExpr;
-  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doFill, NULL, destroyFillOperatorInfo, NULL);
-
-  code = appendDownstream(pOperator, &downstream, 1);
-  return pOperator;
-
-_error:
-  if (pInfo != NULL) {
-    destroyFillOperatorInfo(pInfo);
-  }
-
-  pTaskInfo->code = code;
-  taosMemoryFreeClear(pOperator);
-  return NULL;
-}
-
 static SExecTaskInfo* createExecTaskInfo(uint64_t queryId, uint64_t taskId, EOPTR_EXEC_MODEL model, char* dbFName) {
   SExecTaskInfo* pTaskInfo = taosMemoryCalloc(1, sizeof(SExecTaskInfo));
   if (pTaskInfo == NULL) {

source/libs/executor/src/scanoperator.c

@@ -895,6 +895,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
 
   pInfo->currentGroupId = -1;
   pInfo->assignBlockUid = pTableScanNode->assignBlockUid;
+  pInfo->hasGroupByTag = pTableScanNode->pGroupTags ? true : false;
 
   setOperatorInfo(pOperator, "TableScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN, false, OP_NOT_OPENED, pInfo,
                   pTaskInfo);
@@ -1051,6 +1052,9 @@ static uint64_t getGroupIdByData(SStreamScanInfo* pInfo, uint64_t uid, TSKEY ts,
 }
 
 static bool prepareRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pBlock, int32_t* pRowIndex) {
+  if (pBlock->info.rows == 0) {
+    return false;
+  }
   if ((*pRowIndex) == pBlock->info.rows) {
     return false;
   }
@@ -1183,10 +1187,10 @@ static SSDataBlock* doRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32
 }
 
 static int32_t generateSessionScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock, SSDataBlock* pDestBlock) {
+  blockDataCleanup(pDestBlock);
   if (pSrcBlock->info.rows == 0) {
     return TSDB_CODE_SUCCESS;
   }
-  blockDataCleanup(pDestBlock);
   int32_t code = blockDataEnsureCapacity(pDestBlock, pSrcBlock->info.rows);
   if (code != TSDB_CODE_SUCCESS) {
     return code;
@@ -1335,30 +1339,6 @@ static int32_t generateScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSrcBlock,
   return code;
 }
 
-static void calBlockTag(SExprSupp* pTagCalSup, SSDataBlock* pBlock, SSDataBlock* pResBlock) {
-  if (pTagCalSup == NULL || pTagCalSup->numOfExprs == 0) return;
-  if (pBlock == NULL || pBlock->info.rows == 0) return;
-
-  SSDataBlock* pSrcBlock = blockCopyOneRow(pBlock, 0);
-  ASSERT(pSrcBlock->info.rows == 1);
-
-  blockDataEnsureCapacity(pResBlock, 1);
-
-  projectApplyFunctions(pTagCalSup->pExprInfo, pResBlock, pSrcBlock, pTagCalSup->pCtx, 1, NULL);
-  ASSERT(pResBlock->info.rows == 1);
-
-  // build tagArray
-  /*SArray* tagArray = taosArrayInit(0, sizeof(void*));*/
-  /*STagVal tagVal = {*/
-  /*.cid = 0,*/
-  /*.type = 0,*/
-  /*};*/
-  // build STag
-  // set STag
-
-  blockDataDestroy(pSrcBlock);
-}
-
 void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock) {
   SExprSupp*    pTbNameCalSup = &pInfo->tbnameCalSup;
   SStreamState* pState = pInfo->pStreamScanOp->pTaskInfo->streamInfo.pState;
@@ -1836,6 +1816,12 @@ FETCH_NEXT_BLOCK:
         }
         setBlockGroupIdByUid(pInfo, pDelBlock);
         printDataBlock(pDelBlock, "stream scan delete recv filtered");
+        if (pDelBlock->info.rows == 0) {
+          if (pInfo->tqReader) {
+            blockDataDestroy(pDelBlock);
+          }
+          goto FETCH_NEXT_BLOCK;
+        }
         if (!isIntervalWindow(pInfo) && !isSessionWindow(pInfo) && !isStateWindow(pInfo)) {
           generateDeleteResultBlock(pInfo, pDelBlock, pInfo->pDeleteDataRes);
           pInfo->pDeleteDataRes->info.type = STREAM_DELETE_RESULT;

source/libs/executor/src/timewindowoperator.c

@@ -677,16 +677,6 @@ static void doInterpUnclosedTimeWindow(SOperatorInfo* pOperatorInfo, int32_t num
   }
 }
 
-void printDataBlock(SSDataBlock* pBlock, const char* flag) {
-  if (!pBlock || pBlock->info.rows == 0) {
-    qDebug("===stream===printDataBlock: Block is Null or Empty");
-    return;
-  }
-  char* pBuf = NULL;
-  qDebug("%s", dumpBlockData(pBlock, flag, &pBuf));
-  taosMemoryFree(pBuf);
-}
-
 typedef int32_t (*__compare_fn_t)(void* pKey, void* data, int32_t index);
 
 int32_t binarySearchCom(void* keyList, int num, void* pKey, int order, __compare_fn_t comparefn) {
@@ -3242,8 +3232,8 @@ void doBuildDeleteDataBlock(SOperatorInfo* pOp, SSHashObj* pStDeleted, SSDataBlo
       char parTbName[VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN];
       STR_WITH_MAXSIZE_TO_VARSTR(parTbName, tbname, sizeof(parTbName));
       colDataAppend(pTableCol, pBlock->info.rows, (const char*)parTbName, false);
-    }
       tdbFree(tbname);
+    }
     pBlock->info.rows += 1;
   }
   if ((*Ite) == NULL) {
@@ -3854,7 +3844,7 @@ static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
   SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;
   SStreamStateAggOperatorInfo* pInfo = pOperator->info;
   int32_t                      numOfOutput = pOperator->exprSupp.numOfExprs;
-  int64_t                      groupId = pSDataBlock->info.id.groupId;
+  uint64_t                     groupId = pSDataBlock->info.id.groupId;
   int64_t                      code = TSDB_CODE_SUCCESS;
   TSKEY*                       tsCols = NULL;
   SResultRow*                  pResult = NULL;

source/libs/function/src/builtinsimpl.c

@@ -526,7 +526,7 @@ static int32_t getNumOfElems(SqlFunctionCtx* pCtx) {
  * count function does not use the pCtx->interResBuf to keep the intermediate buffer
  */
 int32_t countFunction(SqlFunctionCtx* pCtx) {
-  int32_t numOfElem = getNumOfElems(pCtx);
+  int32_t numOfElem = 0;
 
   SResultRowEntryInfo*  pResInfo = GET_RES_INFO(pCtx);
   SInputColumnInfoData* pInput = &pCtx->input;
@@ -539,6 +539,7 @@ int32_t countFunction(SqlFunctionCtx* pCtx) {
     numOfElem = 1;
     *((int64_t*)buf) = 0;
   } else {
+    numOfElem = getNumOfElems(pCtx);
     *((int64_t*)buf) += numOfElem;
   }
 
@@ -2043,12 +2044,16 @@ int32_t firstFunction(SqlFunctionCtx* pCtx) {
 
   pInfo->bytes = pInputCol->info.bytes;
 
+  if (IS_NULL_TYPE(pInputCol->info.type)) {
+    return TSDB_CODE_SUCCESS;
+  }
+
   // All null data column, return directly.
   if (pInput->colDataSMAIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows)) {
     ASSERT(pInputCol->hasNull == true);
     // save selectivity value for column consisted of all null values
     firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo);
-    return 0;
+    return TSDB_CODE_SUCCESS;
   }
 
   SColumnDataAgg* pColAgg = (pInput->colDataSMAIsSet) ? pInput->pColumnDataAgg[0] : NULL;
@@ -2147,12 +2152,16 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
   int32_t bytes = pInputCol->info.bytes;
   pInfo->bytes = bytes;
 
+  if (IS_NULL_TYPE(type)) {
+    return TSDB_CODE_SUCCESS;
+  }
+
   // All null data column, return directly.
   if (pInput->colDataSMAIsSet && (pInput->pColumnDataAgg[0]->numOfNull == pInput->totalRows)) {
     ASSERT(pInputCol->hasNull == true);
     // save selectivity value for column consisted of all null values
     firstlastSaveTupleData(pCtx->pSrcBlock, pInput->startRowIndex, pCtx, pInfo);
-    return 0;
+    return TSDB_CODE_SUCCESS;
   }
 
   SColumnDataAgg* pColAgg = (pInput->colDataSMAIsSet) ? pInput->pColumnDataAgg[0] : NULL;
@@ -2417,9 +2426,14 @@ int32_t lastRowFunction(SqlFunctionCtx* pCtx) {
   SInputColumnInfoData* pInput = &pCtx->input;
   SColumnInfoData*      pInputCol = pInput->pData[0];
 
+  int32_t type  = pInputCol->info.type;
   int32_t bytes = pInputCol->info.bytes;
   pInfo->bytes = bytes;
 
+  if (IS_NULL_TYPE(type)) {
+    return TSDB_CODE_SUCCESS;
+  }
+
   TSKEY startKey = getRowPTs(pInput->pPTS, 0);
   TSKEY endKey = getRowPTs(pInput->pPTS, pInput->totalRows - 1);
 
@@ -3869,6 +3883,10 @@ int32_t hllFunction(SqlFunctionCtx* pCtx) {
   int32_t numOfRows = pInput->numOfRows;
 
   int32_t numOfElems = 0;
+  if (IS_NULL_TYPE(type)) {
+    goto _hll_over;
+  }
+
   for (int32_t i = start; i < numOfRows + start; ++i) {
     if (pCol->hasNull && colDataIsNull_s(pCol, i)) {
       continue;
@@ -3890,6 +3908,7 @@ int32_t hllFunction(SqlFunctionCtx* pCtx) {
     }
   }
 
+_hll_over:
   pInfo->totalCount += numOfElems;
 
   if (pInfo->totalCount == 0 && !tsCountAlwaysReturnValue) {
@@ -3913,12 +3932,16 @@ static void hllTransferInfo(SHLLInfo* pInput, SHLLInfo* pOutput) {
 int32_t hllFunctionMerge(SqlFunctionCtx* pCtx) {
   SInputColumnInfoData* pInput = &pCtx->input;
   SColumnInfoData*      pCol = pInput->pData[0];
-  ASSERT(pCol->info.type == TSDB_DATA_TYPE_BINARY);
+
+  if (pCol->info.type != TSDB_DATA_TYPE_BINARY) {
+    return TSDB_CODE_SUCCESS;
+  }
 
   SHLLInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
 
   int32_t start = pInput->startRowIndex;
 
+
   for (int32_t i = start; i < start + pInput->numOfRows; ++i) {
     char*     data = colDataGetData(pCol, i);
     SHLLInfo* pInputInfo = (SHLLInfo*)varDataVal(data);

source/libs/nodes/src/nodesCodeFuncs.c

@@ -2952,6 +2952,7 @@ static const char* jkValueDuration = "Duration";
 static const char* jkValueTranslate = "Translate";
 static const char* jkValueNotReserved = "NotReserved";
 static const char* jkValueIsNull = "IsNull";
+static const char* jkValueUnit = "Unit";
 static const char* jkValueDatum = "Datum";
 
 static int32_t datumToJson(const void* pObj, SJson* pJson) {
@@ -3041,6 +3042,9 @@ static int32_t valueNodeToJson(const void* pObj, SJson* pJson) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonAddBoolToObject(pJson, jkValueIsNull, pNode->isNull);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonAddIntegerToObject(pJson, jkValueUnit, pNode->unit);
+  }
   if (TSDB_CODE_SUCCESS == code && pNode->translate && !pNode->isNull) {
     code = datumToJson(pNode, pJson);
   }
@@ -3191,6 +3195,9 @@ static int32_t jsonToValueNode(const SJson* pJson, void* pObj) {
   if (TSDB_CODE_SUCCESS == code) {
     code = tjsonGetBoolValue(pJson, jkValueIsNull, &pNode->isNull);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = tjsonGetTinyIntValue(pJson, jkValueUnit, &pNode->unit);
+  }
   if (TSDB_CODE_SUCCESS == code && pNode->translate && !pNode->isNull) {
     code = jsonToDatum(pJson, pNode);
   }

source/libs/parser/inc/parInsertUtil.h

@@ -135,7 +135,7 @@ int32_t insAllocateMemForSize(STableDataBlocks *pDataBlock, int32_t allSize);
 int32_t insCreateSName(SName *pName, struct SToken *pTableName, int32_t acctId, const char *dbName, SMsgBuf *pMsgBuf);
 int32_t insFindCol(struct SToken *pColname, int32_t start, int32_t end, SSchema *pSchema);
 void    insBuildCreateTbReq(SVCreateTbReq *pTbReq, const char *tname, STag *pTag, int64_t suid, const char *sname,
-                            SArray *tagName, uint8_t tagNum);
+                            SArray *tagName, uint8_t tagNum, int32_t ttl);
 int32_t insMemRowAppend(SMsgBuf *pMsgBuf, const void *value, int32_t len, void *param);
 int32_t insCheckTimestamp(STableDataBlocks *pDataBlocks, const char *start);
 int32_t insBuildOutput(SHashObj *pVgroupsHashObj, SArray *pVgDataBlocks, SArray **pDataBlocks);

source/libs/parser/src/parInsertSql.c

@@ -562,7 +562,7 @@ static int32_t parseTagValue(SInsertParseContext* pCxt, SVnodeModifOpStmt* pStmt
 
 static void buildCreateTbReq(SVnodeModifOpStmt* pStmt, STag* pTag, SArray* pTagName) {
   insBuildCreateTbReq(&pStmt->createTblReq, pStmt->targetTableName.tname, pTag, pStmt->pTableMeta->suid,
-                      pStmt->usingTableName.tname, pTagName, pStmt->pTableMeta->tableInfo.numOfTags);
+                      pStmt->usingTableName.tname, pTagName, pStmt->pTableMeta->tableInfo.numOfTags, TSDB_DEFAULT_TABLE_TTL);
 }
 
 static int32_t checkAndTrimValue(SToken* pToken, char* tmpTokenBuf, SMsgBuf* pMsgBuf) {

source/libs/parser/src/parInsertStmt.c

@@ -137,7 +137,7 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, const ch
   }
 
   SVCreateTbReq tbReq = {0};
-  insBuildCreateTbReq(&tbReq, tName, pTag, suid, sTableName, tagName, pDataBlock->pTableMeta->tableInfo.numOfTags);
+  insBuildCreateTbReq(&tbReq, tName, pTag, suid, sTableName, tagName, pDataBlock->pTableMeta->tableInfo.numOfTags, TSDB_DEFAULT_TABLE_TTL);
   code = insBuildCreateTbMsg(pDataBlock, &tbReq);
   tdDestroySVCreateTbReq(&tbReq);
 

source/libs/stream/src/streamRecover.c

@@ -36,7 +36,7 @@ int32_t streamTaskLaunchRecover(SStreamTask* pTask, int64_t version) {
     SRpcMsg rpcMsg = {
         .contLen = len,
         .pCont = serializedReq,
-        .msgType = TDMT_VND_STREAM_RECOVER_STEP1,
+        .msgType = TDMT_VND_STREAM_RECOVER_NONBLOCKING_STAGE,
     };
 
     if (tmsgPutToQueue(pTask->pMsgCb, STREAM_QUEUE, &rpcMsg) < 0) {