Merge branch '3.0' into enh/changeRpcLogLevel

include/common/tcommon.h

@@ -252,6 +252,7 @@ typedef struct SQueryTableDataCond {
   int32_t      order;    // desc|asc order to iterate the data block
   int32_t      numOfCols;
   SColumnInfo* colList;
+  int32_t*     pSlotList; // the column output destation slot, and it may be null
   int32_t      type;     // data block load type:
   STimeWindow  twindows;
   int64_t      startVersion;

include/common/tdatablock.h

@@ -41,9 +41,9 @@ typedef struct SBlockOrderInfo {
     BMCharPos(bm_, r_) |= (1u << (7u - BitPos(r_))); \
   } while (0)
 
-#define colDataSetNotNull_f(bm_, r_)                  \
+#define colDataClearNull_f(bm_, r_)                             \
   do {                                                          \
-    BMCharPos(bm_, r_) &= ~(1u << (7u - BitPos(r_))); \
+    BMCharPos(bm_, r_) &= ((char)(~(1u << (7u - BitPos(r_))))); \
   } while (0)
 
 #define colDataIsNull_var(pColumnInfoData, row)  (pColumnInfoData->varmeta.offset[row] == -1)
@@ -151,9 +151,6 @@ static FORCE_INLINE void colDataAppendNNULL(SColumnInfoData* pColumnInfoData, ui
     for (int32_t i = start; i < start + nRows; ++i) {
       colDataSetNull_f(pColumnInfoData->nullbitmap, i);
     }
-
-    int32_t bytes = pColumnInfoData->info.bytes;
-    memset(pColumnInfoData->pData + start * bytes, 0, nRows * bytes);
   }
 
   pColumnInfoData->hasNull = true;
@@ -234,9 +231,11 @@ int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullF
 
 int32_t colInfoDataEnsureCapacity(SColumnInfoData* pColumn, uint32_t numOfRows, bool clearPayload);
 int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows);
+int32_t blockDataEnsureCapacityNoClear(SSDataBlock* pDataBlock, uint32_t numOfRows);
 
 void colInfoDataCleanup(SColumnInfoData* pColumn, uint32_t numOfRows);
 void blockDataCleanup(SSDataBlock* pDataBlock);
+void blockDataEmpty(SSDataBlock* pDataBlock);
 
 size_t blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize);
 

include/common/tmsgcb.h

@@ -43,7 +43,6 @@ typedef int32_t (*PutToQueueFp)(void* pMgmt, EQueueType qtype, SRpcMsg* pMsg);
 typedef int32_t (*GetQueueSizeFp)(void* pMgmt, int32_t vgId, EQueueType qtype);
 typedef int32_t (*SendReqFp)(const SEpSet* pEpSet, SRpcMsg* pMsg);
 typedef void (*SendRspFp)(SRpcMsg* pMsg);
-typedef void (*SendRedirectRspFp)(SRpcMsg* pMsg, const SEpSet* pNewEpSet);
 typedef void (*RegisterBrokenLinkArgFp)(SRpcMsg* pMsg);
 typedef void (*ReleaseHandleFp)(SRpcHandleInfo* pHandle, int8_t type);
 typedef void (*ReportStartup)(const char* name, const char* desc);
@@ -55,7 +54,6 @@ typedef struct {
   GetQueueSizeFp          qsizeFp;
   SendReqFp               sendReqFp;
   SendRspFp               sendRspFp;
-  SendRedirectRspFp       sendRedirectRspFp;
   RegisterBrokenLinkArgFp registerBrokenLinkArgFp;
   ReleaseHandleFp         releaseHandleFp;
   ReportStartup           reportStartupFp;
@@ -66,7 +64,6 @@ int32_t tmsgPutToQueue(const SMsgCb* msgcb, EQueueType qtype, SRpcMsg* pMsg);
 int32_t tmsgGetQueueSize(const SMsgCb* msgcb, int32_t vgId, EQueueType qtype);
 int32_t tmsgSendReq(const SEpSet* epSet, SRpcMsg* pMsg);
 void    tmsgSendRsp(SRpcMsg* pMsg);
-void    tmsgSendRedirectRsp(SRpcMsg* pMsg, const SEpSet* pNewEpSet);
 void    tmsgRegisterBrokenLinkArg(SRpcMsg* pMsg);
 void    tmsgReleaseHandle(SRpcHandleInfo* pHandle, int8_t type);
 void    tmsgReportStartup(const char* name, const char* desc);

include/common/ttypes.h

@@ -278,11 +278,9 @@ typedef struct {
 #define IS_VALID_TINYINT(_t)   ((_t) >= INT8_MIN && (_t) <= INT8_MAX)
 #define IS_VALID_SMALLINT(_t)  ((_t) >= INT16_MIN && (_t) <= INT16_MAX)
 #define IS_VALID_INT(_t)       ((_t) >= INT32_MIN && (_t) <= INT32_MAX)
-#define IS_VALID_BIGINT(_t)    ((_t) >= INT64_MIN && (_t) <= INT64_MAX)
 #define IS_VALID_UTINYINT(_t)  ((_t) >= 0 && (_t) <= UINT8_MAX)
 #define IS_VALID_USMALLINT(_t) ((_t) >= 0 && (_t) <= UINT16_MAX)
 #define IS_VALID_UINT(_t)      ((_t) >= 0 && (_t) <= UINT32_MAX)
-#define IS_VALID_UBIGINT(_t)   ((_t) >= 0 && (_t) <= UINT64_MAX)
 #define IS_VALID_FLOAT(_t)     ((_t) >= -FLT_MAX && (_t) <= FLT_MAX)
 #define IS_VALID_DOUBLE(_t)    ((_t) >= -DBL_MAX && (_t) <= DBL_MAX)
 

include/libs/function/function.h

@@ -137,10 +137,12 @@ typedef struct SqlFunctionCtx {
   int16_t              functionId;  // function id
   char                *pOutput;     // final result output buffer, point to sdata->data
   int32_t              numOfParams;
-  SFunctParam     *param;  // input parameter, e.g., top(k, 20), the number of results for top query is kept in param
-  SColumnInfoData *pTsOutput;  // corresponding output buffer for timestamp of each result, e.g., top/bottom*/
+  // input parameter, e.g., top(k, 20), the number of results of top query is kept in param
+  SFunctParam         *param;
+  // corresponding output buffer for timestamp of each result, e.g., diff/csum
+  SColumnInfoData     *pTsOutput;
   int32_t              offset;
-  struct SResultRowEntryInfo *resultInfo;
+  SResultRowEntryInfo *resultInfo;
   SSubsidiaryResInfo   subsidiaries;
   SPoint1              start;
   SPoint1              end;
@@ -150,8 +152,6 @@ typedef struct SqlFunctionCtx {
   struct SSDataBlock  *pSrcBlock;
   struct SSDataBlock  *pDstBlock;  // used by indefinite rows function to set selectivity
   SSerializeDataHandle saveHandle;
-  bool                        isStream;
-
   char                 udfName[TSDB_FUNC_NAME_LEN];
 } SqlFunctionCtx;
 
@@ -183,7 +183,6 @@ struct SScalarParam {
 };
 
 void    cleanupResultRowEntry(struct SResultRowEntryInfo *pCell);
-//int32_t getNumOfResult(SqlFunctionCtx *pCtx, int32_t num, SSDataBlock *pResBlock);
 bool    isRowEntryCompleted(struct SResultRowEntryInfo *pEntry);
 bool    isRowEntryInitialized(struct SResultRowEntryInfo *pEntry);
 
@@ -195,32 +194,6 @@ typedef struct SPoint {
 int32_t taosGetLinearInterpolationVal(SPoint *point, int32_t outputType, SPoint *point1, SPoint *point2,
                                       int32_t inputType);
 
-///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
-// udf api
-/**
- * create udfd proxy, called once in process that call doSetupUdf/callUdfxxx/doTeardownUdf
- * @return error code
- */
-int32_t udfcOpen();
-
-/**
- * destroy udfd proxy
- * @return error code
- */
-int32_t udfcClose();
-
-/**
- * start udfd that serves udf function invocation under dnode startDnodeId
- * @param startDnodeId
- * @return
- */
-int32_t udfStartUdfd(int32_t startDnodeId);
-/**
- * stop udfd
- * @return
- */
-int32_t udfStopUdfd();
-
 #ifdef __cplusplus
 }
 #endif

include/libs/function/tudf.h

@@ -85,6 +85,32 @@ int32_t callUdfScalarFunc(char *udfName, SScalarParam *input, int32_t numOfCols,
 
 int32_t cleanUpUdfs();
 
+///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+// udf api
+/**
+ * create udfd proxy, called once in process that call doSetupUdf/callUdfxxx/doTeardownUdf
+ * @return error code
+ */
+int32_t udfcOpen();
+
+/**
+ * destroy udfd proxy
+ * @return error code
+ */
+int32_t udfcClose();
+
+/**
+ * start udfd that serves udf function invocation under dnode startDnodeId
+ * @param startDnodeId
+ * @return
+ */
+int32_t udfStartUdfd(int32_t startDnodeId);
+/**
+ * stop udfd
+ * @return
+ */
+int32_t udfStopUdfd();
+
 #ifdef __cplusplus
 }
 #endif

include/libs/qcom/query.h

@@ -261,14 +261,13 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
    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 || (_code) == TSDB_CODE_VND_STOPPED)
-#define SYNC_SELF_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_RESTORING || (_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_RPC_BROKEN_LINK || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL ||  \
    (_code) == TSDB_CODE_MNODE_NOT_FOUND || SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || \
    SYNC_SELF_LEADER_REDIRECT_ERROR(_code) || SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) || \
-   (_code) == TSDB_CODE_SYN_RESTORING || (_code) == TSDB_CODE_RPC_BROKEN_LINK ||        \
    (_code) == TSDB_CODE_APP_IS_STARTING || (_code) == TSDB_CODE_APP_IS_STOPPING)
 
 #define NEED_CLIENT_RM_TBLMETA_REQ(_type)                                                                  \
@@ -276,9 +275,8 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
    (_type) == TDMT_MND_DROP_STB)
 
 #define NEED_SCHEDULER_REDIRECT_ERROR(_code)                                            \
-  ((_code) == TSDB_CODE_RPC_REDIRECT || (_code) == TSDB_CODE_MNODE_NOT_FOUND ||           \
-   SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || SYNC_SELF_LEADER_REDIRECT_ERROR(_code) || \
-   SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) || (_code) == TSDB_CODE_SYN_RESTORING ||       \
+  ((_code) == TSDB_CODE_MNODE_NOT_FOUND || SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || \
+   SYNC_SELF_LEADER_REDIRECT_ERROR(_code) || SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) || \
    (_code) == TSDB_CODE_APP_IS_STARTING || (_code) == TSDB_CODE_APP_IS_STOPPING)
 
 #define REQUEST_TOTAL_EXEC_TIMES 2

include/util/taoserror.h

@@ -40,61 +40,61 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_FAILED                    -1   // unknown or needn't tell detail error
 
 // rpc
-// #define TSDB_CODE_RPC_ACTION_IN_PROGRESS     TAOS_DEF_ERROR_CODE(0, 0x0001) //2.x
-// #define TSDB_CODE_RPC_AUTH_REQUIRED          TAOS_DEF_ERROR_CODE(0, 0x0002) //2.x
-#define TSDB_CODE_RPC_AUTH_FAILURE              TAOS_DEF_ERROR_CODE(0, 0x0003)
-#define TSDB_CODE_RPC_REDIRECT                  TAOS_DEF_ERROR_CODE(0, 0x0004)
-// #define TSDB_CODE_RPC_NOT_READY              TAOS_DEF_ERROR_CODE(0, 0x0005) //2.x
-// #define TSDB_CODE_RPC_ALREADY_PROCESSED      TAOS_DEF_ERROR_CODE(0, 0x0006) //2.x
-// #define TSDB_CODE_RPC_LAST_SESSION_NOT_FINI. TAOS_DEF_ERROR_CODE(0, 0x0007) //2.x
-// #define TSDB_CODE_RPC_MISMATCHED_LINK_ID     TAOS_DEF_ERROR_CODE(0, 0x0008) //2.x
-// #define TSDB_CODE_RPC_TOO_SLOW               TAOS_DEF_ERROR_CODE(0, 0x0009) //2.x
-// #define TSDB_CODE_RPC_MAX_SESSIONS           TAOS_DEF_ERROR_CODE(0, 0x000A) //2.x
+// #define TSDB_CODE_RPC_ACTION_IN_PROGRESS     TAOS_DEF_ERROR_CODE(0, 0x0001) // 2.x
+// #define TSDB_CODE_RPC_AUTH_REQUIRED          TAOS_DEF_ERROR_CODE(0, 0x0002) // 2.x
+// #define TSDB_CODE_RPC_AUTH_FAILURE           TAOS_DEF_ERROR_CODE(0, 0x0003) // 2.x
+// #define TSDB_CODE_RPC_REDIRECT               TAOS_DEF_ERROR_CODE(0, 0x0004) // 2.x
+// #define TSDB_CODE_RPC_NOT_READY              TAOS_DEF_ERROR_CODE(0, 0x0005) // 2.x
+// #define TSDB_CODE_RPC_ALREADY_PROCESSED      TAOS_DEF_ERROR_CODE(0, 0x0006) // 2.x
+// #define TSDB_CODE_RPC_LAST_SESSION_NOT_FINI. TAOS_DEF_ERROR_CODE(0, 0x0007) // 2.x
+// #define TSDB_CODE_RPC_MISMATCHED_LINK_ID     TAOS_DEF_ERROR_CODE(0, 0x0008) // 2.x
+// #define TSDB_CODE_RPC_TOO_SLOW               TAOS_DEF_ERROR_CODE(0, 0x0009) // 2.x
+// #define TSDB_CODE_RPC_MAX_SESSIONS           TAOS_DEF_ERROR_CODE(0, 0x000A) // 2.x
 #define TSDB_CODE_RPC_NETWORK_UNAVAIL           TAOS_DEF_ERROR_CODE(0, 0x000B)
-// #define TSDB_CODE_RPC_APP_ERROR              TAOS_DEF_ERROR_CODE(0, 0x000C) //2.x
-// #define TSDB_CODE_RPC_UNEXPECTED_RESPONSE    TAOS_DEF_ERROR_CODE(0, 0x000D) //2.x
-// #define TSDB_CODE_RPC_INVALID_VALUE          TAOS_DEF_ERROR_CODE(0, 0x000E) //2.x
-// #define TSDB_CODE_RPC_INVALID_TRAN_ID        TAOS_DEF_ERROR_CODE(0, 0x000F) //2.x
-// #define TSDB_CODE_RPC_INVALID_SESSION_ID     TAOS_DEF_ERROR_CODE(0, 0x0010) //2.x
-// #define TSDB_CODE_RPC_INVALID_MSG_TYPE       TAOS_DEF_ERROR_CODE(0, 0x0011) //2.x
-// #define TSDB_CODE_RPC_INVALID_RESPONSE_TYPE  TAOS_DEF_ERROR_CODE(0, 0x0012) //2.x
-#define TSDB_CODE_TIME_UNSYNCED                 TAOS_DEF_ERROR_CODE(0, 0x0013)
-// #define TSDB_CODE_APP_NOT_READY              TAOS_DEF_ERROR_CODE(0, 0x0014) //2.x
+// #define TSDB_CODE_RPC_APP_ERROR              TAOS_DEF_ERROR_CODE(0, 0x000C) // 2.x
+// #define TSDB_CODE_RPC_UNEXPECTED_RESPONSE    TAOS_DEF_ERROR_CODE(0, 0x000D) // 2.x
+// #define TSDB_CODE_RPC_INVALID_VALUE          TAOS_DEF_ERROR_CODE(0, 0x000E) // 2.x
+// #define TSDB_CODE_RPC_INVALID_TRAN_ID        TAOS_DEF_ERROR_CODE(0, 0x000F) // 2.x
+// #define TSDB_CODE_RPC_INVALID_SESSION_ID     TAOS_DEF_ERROR_CODE(0, 0x0010) // 2.x
+// #define TSDB_CODE_RPC_INVALID_MSG_TYPE       TAOS_DEF_ERROR_CODE(0, 0x0011) // 2.x
+// #define TSDB_CODE_RPC_INVALID_RESPONSE_TYPE  TAOS_DEF_ERROR_CODE(0, 0x0012) // 2.x
+#define TSDB_CODE_TIME_UNSYNCED                 TAOS_DEF_ERROR_CODE(0, 0x0013) //
+// #define TSDB_CODE_APP_NOT_READY              TAOS_DEF_ERROR_CODE(0, 0x0014) // 2.x
 #define TSDB_CODE_RPC_FQDN_ERROR                TAOS_DEF_ERROR_CODE(0, 0x0015)
-// #define TSDB_CODE_RPC_INVALID_VERSION        TAOS_DEF_ERROR_CODE(0, 0x0016) //2.x
-#define TSDB_CODE_RPC_PORT_EADDRINUSE           TAOS_DEF_ERROR_CODE(0, 0x0017)
-#define TSDB_CODE_RPC_BROKEN_LINK               TAOS_DEF_ERROR_CODE(0, 0x0018)
-#define TSDB_CODE_RPC_TIMEOUT                   TAOS_DEF_ERROR_CODE(0, 0x0019)
+// #define TSDB_CODE_RPC_INVALID_VERSION        TAOS_DEF_ERROR_CODE(0, 0x0016) // 2.x
+#define TSDB_CODE_RPC_PORT_EADDRINUSE           TAOS_DEF_ERROR_CODE(0, 0x0017) //
+#define TSDB_CODE_RPC_BROKEN_LINK               TAOS_DEF_ERROR_CODE(0, 0x0018) //
+#define TSDB_CODE_RPC_TIMEOUT                   TAOS_DEF_ERROR_CODE(0, 0x0019) //
 
 //common & util
-#define TSDB_CODE_OPS_NOT_SUPPORT               TAOS_DEF_ERROR_CODE(0, 0x0100)
-#define TSDB_CODE_MEMORY_CORRUPTED              TAOS_DEF_ERROR_CODE(0, 0x0101)
+#define TSDB_CODE_OPS_NOT_SUPPORT               TAOS_DEF_ERROR_CODE(0, 0x0100) //
+// #define TSDB_CODE_MEMORY_CORRUPTED           TAOS_DEF_ERROR_CODE(0, 0x0101) // 2.x
 #define TSDB_CODE_OUT_OF_MEMORY                 TAOS_DEF_ERROR_CODE(0, 0x0102)
 // #define TSDB_CODE_COM_INVALID_CFG_MSG        TAOS_DEF_ERROR_CODE(0, 0x0103) // 2.x
-#define TSDB_CODE_FILE_CORRUPTED                TAOS_DEF_ERROR_CODE(0, 0x0104)
-#define TSDB_CODE_REF_NO_MEMORY                 TAOS_DEF_ERROR_CODE(0, 0x0105)
-#define TSDB_CODE_REF_FULL                      TAOS_DEF_ERROR_CODE(0, 0x0106)
-#define TSDB_CODE_REF_ID_REMOVED                TAOS_DEF_ERROR_CODE(0, 0x0107)
-#define TSDB_CODE_REF_INVALID_ID                TAOS_DEF_ERROR_CODE(0, 0x0108)
-#define TSDB_CODE_REF_ALREADY_EXIST             TAOS_DEF_ERROR_CODE(0, 0x0109)
-#define TSDB_CODE_REF_NOT_EXIST                 TAOS_DEF_ERROR_CODE(0, 0x010A)
-
-#define TSDB_CODE_APP_ERROR                     TAOS_DEF_ERROR_CODE(0, 0x0110)
-#define TSDB_CODE_ACTION_IN_PROGRESS            TAOS_DEF_ERROR_CODE(0, 0x0111)
-#define TSDB_CODE_OUT_OF_RANGE                  TAOS_DEF_ERROR_CODE(0, 0x0112)
-#define TSDB_CODE_OUT_OF_SHM_MEM                TAOS_DEF_ERROR_CODE(0, 0x0113)
-#define TSDB_CODE_INVALID_SHM_ID                TAOS_DEF_ERROR_CODE(0, 0x0114)
-#define TSDB_CODE_INVALID_MSG                   TAOS_DEF_ERROR_CODE(0, 0x0115)
+#define TSDB_CODE_FILE_CORRUPTED                TAOS_DEF_ERROR_CODE(0, 0x0104) //
+#define TSDB_CODE_REF_NO_MEMORY                 TAOS_DEF_ERROR_CODE(0, 0x0105) // internal
+#define TSDB_CODE_REF_FULL                      TAOS_DEF_ERROR_CODE(0, 0x0106) // internal
+#define TSDB_CODE_REF_ID_REMOVED                TAOS_DEF_ERROR_CODE(0, 0x0107) // internal
+#define TSDB_CODE_REF_INVALID_ID                TAOS_DEF_ERROR_CODE(0, 0x0108) // internal
+#define TSDB_CODE_REF_ALREADY_EXIST             TAOS_DEF_ERROR_CODE(0, 0x0109) // internal
+#define TSDB_CODE_REF_NOT_EXIST                 TAOS_DEF_ERROR_CODE(0, 0x010A) // internal
+
+#define TSDB_CODE_APP_ERROR                     TAOS_DEF_ERROR_CODE(0, 0x0110) //
+#define TSDB_CODE_ACTION_IN_PROGRESS            TAOS_DEF_ERROR_CODE(0, 0x0111) // internal
+#define TSDB_CODE_OUT_OF_RANGE                  TAOS_DEF_ERROR_CODE(0, 0x0112) //
+// #define TSDB_CODE_OUT_OF_SHM_MEM             TAOS_DEF_ERROR_CODE(0, 0x0113)
+// #define TSDB_CODE_INVALID_SHM_ID             TAOS_DEF_ERROR_CODE(0, 0x0114)
+#define TSDB_CODE_INVALID_MSG                   TAOS_DEF_ERROR_CODE(0, 0x0115) //
 #define TSDB_CODE_INVALID_MSG_LEN               TAOS_DEF_ERROR_CODE(0, 0x0116) //
-#define TSDB_CODE_INVALID_PTR                   TAOS_DEF_ERROR_CODE(0, 0x0117)
-#define TSDB_CODE_INVALID_PARA                  TAOS_DEF_ERROR_CODE(0, 0x0118)
-#define TSDB_CODE_INVALID_CFG                   TAOS_DEF_ERROR_CODE(0, 0x0119)
-#define TSDB_CODE_INVALID_OPTION                TAOS_DEF_ERROR_CODE(0, 0x011A)
-#define TSDB_CODE_INVALID_JSON_FORMAT           TAOS_DEF_ERROR_CODE(0, 0x011B)
-#define TSDB_CODE_INVALID_VERSION_NUMBER        TAOS_DEF_ERROR_CODE(0, 0x011C)
-#define TSDB_CODE_INVALID_VERSION_STRING        TAOS_DEF_ERROR_CODE(0, 0x011D)
-#define TSDB_CODE_VERSION_NOT_COMPATIBLE        TAOS_DEF_ERROR_CODE(0, 0x011E)
-#define TSDB_CODE_CHECKSUM_ERROR                TAOS_DEF_ERROR_CODE(0, 0x011F)
+#define TSDB_CODE_INVALID_PTR                   TAOS_DEF_ERROR_CODE(0, 0x0117) // internal
+#define TSDB_CODE_INVALID_PARA                  TAOS_DEF_ERROR_CODE(0, 0x0118) //
+#define TSDB_CODE_INVALID_CFG                   TAOS_DEF_ERROR_CODE(0, 0x0119) // internal
+#define TSDB_CODE_INVALID_OPTION                TAOS_DEF_ERROR_CODE(0, 0x011A) // internal
+#define TSDB_CODE_INVALID_JSON_FORMAT           TAOS_DEF_ERROR_CODE(0, 0x011B) // internal
+#define TSDB_CODE_INVALID_VERSION_NUMBER        TAOS_DEF_ERROR_CODE(0, 0x011C) // internal
+#define TSDB_CODE_INVALID_VERSION_STRING        TAOS_DEF_ERROR_CODE(0, 0x011D) // internal
+#define TSDB_CODE_VERSION_NOT_COMPATIBLE        TAOS_DEF_ERROR_CODE(0, 0x011E) // internal
+#define TSDB_CODE_CHECKSUM_ERROR                TAOS_DEF_ERROR_CODE(0, 0x011F) // internal
 
 #define TSDB_CODE_COMPRESS_ERROR                TAOS_DEF_ERROR_CODE(0, 0x0120)
 #define TSDB_CODE_MSG_NOT_PROCESSED             TAOS_DEF_ERROR_CODE(0, 0x0121) //

source/client/src/clientMain.c

@@ -146,7 +146,6 @@ void taos_close(TAOS *taos) {
 
 int taos_errno(TAOS_RES *res) {
   if (res == NULL || TD_RES_TMQ_META(res)) {
-    if (terrno == TSDB_CODE_RPC_REDIRECT) terrno = TSDB_CODE_QRY_NOT_READY;
     return terrno;
   }
 
@@ -154,12 +153,11 @@ int taos_errno(TAOS_RES *res) {
     return 0;
   }
 
-  return ((SRequestObj *)res)->code == TSDB_CODE_RPC_REDIRECT ? TSDB_CODE_QRY_NOT_READY : ((SRequestObj *)res)->code;
+  return ((SRequestObj *)res)->code;
 }
 
 const char *taos_errstr(TAOS_RES *res) {
   if (res == NULL || TD_RES_TMQ_META(res)) {
-    if (terrno == TSDB_CODE_RPC_REDIRECT) terrno = TSDB_CODE_QRY_NOT_READY;
     return (const char *)tstrerror(terrno);
   }
 
@@ -171,8 +169,7 @@ const char *taos_errstr(TAOS_RES *res) {
   if (NULL != pRequest->msgBuf && (strlen(pRequest->msgBuf) > 0 || pRequest->code == TSDB_CODE_RPC_FQDN_ERROR)) {
     return pRequest->msgBuf;
   } else {
-    return pRequest->code == TSDB_CODE_RPC_REDIRECT ? (const char *)tstrerror(TSDB_CODE_QRY_NOT_READY)
-                                                    : (const char *)tstrerror(pRequest->code);
+    return (const char *)tstrerror(pRequest->code);
   }
 }
 

source/common/src/tdatablock.c

@@ -1137,14 +1137,15 @@ int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullF
 }
 
 void blockDataCleanup(SSDataBlock* pDataBlock) {
+  blockDataEmpty(pDataBlock);
   SDataBlockInfo* pInfo = &pDataBlock->info;
-
-  pInfo->rows = 0;
   pInfo->id.uid = 0;
   pInfo->id.groupId = 0;
-  pInfo->window.ekey = 0;
-  pInfo->window.skey = 0;
+}
 
+void blockDataEmpty(SSDataBlock* pDataBlock) {
+  SDataBlockInfo* pInfo = &pDataBlock->info;
+  ASSERT(pInfo->rows <= pDataBlock->info.capacity);
   if (pInfo->capacity == 0) {
     return;
   }
@@ -1154,6 +1155,10 @@ void blockDataCleanup(SSDataBlock* pDataBlock) {
     SColumnInfoData* p = taosArrayGet(pDataBlock->pDataBlock, i);
     colInfoDataCleanup(p, pInfo->capacity);
   }
+
+  pInfo->rows = 0;
+  pInfo->window.ekey = 0;
+  pInfo->window.skey = 0;
 }
 
 // todo temporarily disable it
@@ -1249,6 +1254,25 @@ int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows) {
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t blockDataEnsureCapacityNoClear(SSDataBlock* pDataBlock, uint32_t numOfRows) {
+  int32_t code = 0;
+  if (numOfRows == 0 || numOfRows <= pDataBlock->info.capacity) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  size_t numOfCols = taosArrayGetSize(pDataBlock->pDataBlock);
+  for (int32_t i = 0; i < numOfCols; ++i) {
+    SColumnInfoData* p = taosArrayGet(pDataBlock->pDataBlock, i);
+    code = doEnsureCapacity(p, &pDataBlock->info, numOfRows, false);
+    if (code) {
+      return code;
+    }
+  }
+
+  pDataBlock->info.capacity = numOfRows;
+  return TSDB_CODE_SUCCESS;
+}
+
 void blockDataFreeRes(SSDataBlock* pBlock) {
   int32_t numOfOutput = taosArrayGetSize(pBlock->pDataBlock);
   for (int32_t i = 0; i < numOfOutput; ++i) {
@@ -1621,6 +1645,8 @@ static int32_t colDataMoveVarData(SColumnInfoData* pColInfoData, size_t start, s
 static void colDataTrimFirstNRows(SColumnInfoData* pColInfoData, size_t n, size_t total) {
   if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
     pColInfoData->varmeta.length = colDataMoveVarData(pColInfoData, n, total);
+
+    // clear the offset value of the unused entries.
     memset(&pColInfoData->varmeta.offset[total - n], 0, n);
   } else {
     int32_t bytes = pColInfoData->info.bytes;
@@ -1635,7 +1661,7 @@ int32_t blockDataTrimFirstNRows(SSDataBlock* pBlock, size_t n) {
   }
 
   if (pBlock->info.rows <= n) {
-    blockDataCleanup(pBlock);
+    blockDataEmpty(pBlock);
   } else {
     size_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
     for (int32_t i = 0; i < numOfCols; ++i) {
@@ -1652,12 +1678,22 @@ static void colDataKeepFirstNRows(SColumnInfoData* pColInfoData, size_t n, size_
   if (IS_VAR_DATA_TYPE(pColInfoData->info.type)) {
     pColInfoData->varmeta.length = colDataMoveVarData(pColInfoData, 0, n);
     memset(&pColInfoData->varmeta.offset[n], 0, total - n);
+  } else {  // reset the bitmap value
+    /*int32_t stopIndex = BitmapLen(n) * 8;
+    for(int32_t i = n; i < stopIndex; ++i) {
+      colDataClearNull_f(pColInfoData->nullbitmap, i);
+    }
+
+    int32_t remain = BitmapLen(total) - BitmapLen(n);
+    if (remain > 0) {
+      memset(pColInfoData->nullbitmap+BitmapLen(n), 0, remain);
+    }*/
   }
 }
 
 int32_t blockDataKeepFirstNRows(SSDataBlock* pBlock, size_t n) {
   if (n == 0) {
-    blockDataCleanup(pBlock);
+    blockDataEmpty(pBlock);
     return TSDB_CODE_SUCCESS;
   }
 

source/dnode/mgmt/mgmt_mnode/src/mmWorker.c

@@ -61,7 +61,7 @@ static void mmProcessRpcMsg(SQueueInfo *pInfo, SRpcMsg *pMsg) {
     pMsg->info.rsp = NULL;
   }
 
-  if (code == TSDB_CODE_RPC_REDIRECT) {
+  if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING) {
     mndPostProcessQueryMsg(pMsg);
   }
 

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

@@ -33,23 +33,6 @@ static inline void dmBuildMnodeRedirectRsp(SDnode *pDnode, SRpcMsg *pMsg) {
   }
 }
 
-static inline void dmSendRedirectRsp(SRpcMsg *pMsg, const SEpSet *pNewEpSet) {
-  pMsg->info.hasEpSet = 1;
-  SRpcMsg rsp = {.code = TSDB_CODE_RPC_REDIRECT, .info = pMsg->info, .msgType = pMsg->msgType};
-  int32_t contLen = tSerializeSEpSet(NULL, 0, pNewEpSet);
-
-  rsp.pCont = rpcMallocCont(contLen);
-  if (rsp.pCont == NULL) {
-    pMsg->code = TSDB_CODE_OUT_OF_MEMORY;
-  } else {
-    tSerializeSEpSet(rsp.pCont, contLen, pNewEpSet);
-    rsp.contLen = contLen;
-  }
-  dmSendRsp(&rsp);
-  rpcFreeCont(pMsg->pCont);
-  pMsg->pCont = NULL;
-}
-
 int32_t dmProcessNodeMsg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg) {
   const STraceId *trace = &pMsg->info.traceId;
 
@@ -243,9 +226,9 @@ static inline void dmRegisterBrokenLinkArg(SRpcMsg *pMsg) { rpcRegisterBrokenLin
 static inline void dmReleaseHandle(SRpcHandleInfo *pHandle, int8_t type) { rpcReleaseHandle(pHandle, type); }
 
 static bool rpcRfp(int32_t code, tmsg_t msgType) {
-  if (code == TSDB_CODE_RPC_REDIRECT || code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_MNODE_NOT_FOUND ||
-      code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_RPC_BROKEN_LINK ||
-      code == TSDB_CODE_VND_STOPPED || code == TSDB_CODE_APP_IS_STARTING || code == TSDB_CODE_APP_IS_STOPPING) {
+  if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_RPC_BROKEN_LINK || code == TSDB_CODE_MNODE_NOT_FOUND ||
+      code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_VND_STOPPED ||
+      code == TSDB_CODE_APP_IS_STARTING || code == TSDB_CODE_APP_IS_STOPPING) {
     if (msgType == TDMT_SCH_QUERY || msgType == TDMT_SCH_MERGE_QUERY || msgType == TDMT_SCH_FETCH ||
         msgType == TDMT_SCH_MERGE_FETCH) {
       return false;
@@ -334,7 +317,6 @@ SMsgCb dmGetMsgcb(SDnode *pDnode) {
       .clientRpc = pDnode->trans.clientRpc,
       .sendReqFp = dmSendReq,
       .sendRspFp = dmSendRsp,
-      .sendRedirectRspFp = dmSendRedirectRsp,
       .registerBrokenLinkArgFp = dmRegisterBrokenLinkArg,
       .releaseHandleFp = dmReleaseHandle,
       .reportStartupFp = dmReportStartup,

source/dnode/mgmt/node_util/inc/dmUtil.h

@@ -39,7 +39,7 @@
 #include "sync.h"
 #include "wal.h"
 
-#include "libs/function/function.h"
+#include "libs/function/tudf.h"
 #ifdef __cplusplus
 extern "C" {
 #endif

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

@@ -220,7 +220,7 @@ static int32_t mndProcessCreateAcctReq(SRpcMsg *pReq) {
     return -1;
   }
 
-  terrno = TSDB_CODE_MSG_NOT_PROCESSED;
+  terrno = TSDB_CODE_OPS_NOT_SUPPORT;
   mError("failed to process create acct request since %s", terrstr());
   return -1;
 }
@@ -230,7 +230,7 @@ static int32_t mndProcessAlterAcctReq(SRpcMsg *pReq) {
     return -1;
   }
 
-  terrno = TSDB_CODE_MSG_NOT_PROCESSED;
+  terrno = TSDB_CODE_OPS_NOT_SUPPORT;
   mError("failed to process create acct request since %s", terrstr());
   return -1;
 }
@@ -240,7 +240,7 @@ static int32_t mndProcessDropAcctReq(SRpcMsg *pReq) {
     return -1;
   }
 
-  terrno = TSDB_CODE_MSG_NOT_PROCESSED;
+  terrno = TSDB_CODE_OPS_NOT_SUPPORT;
   mError("failed to process create acct request since %s", terrstr());
   return -1;
 }
\ No newline at end of file

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

@@ -945,7 +945,7 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
         code = TSDB_CODE_MND_TRANS_NETWORK_UNAVAILL;
       }
       if (i != 0 && code == 0) {
-        code = TSDB_CODE_RPC_REDIRECT;
+        code = TSDB_CODE_MNODE_NOT_FOUND;
       }
       mInfo("trans:%d, client:%d send rsp, code:0x%x stage:%s app:%p", pTrans->id, i, code, mndTransStr(pTrans->stage),
             pInfo->ahandle);
@@ -1042,8 +1042,8 @@ static void mndTransResetAction(SMnode *pMnode, STrans *pTrans, STransAction *pA
   pAction->rawWritten = 0;
   pAction->msgSent = 0;
   pAction->msgReceived = 0;
-  if (pAction->errCode == TSDB_CODE_RPC_REDIRECT || pAction->errCode == TSDB_CODE_SYN_NEW_CONFIG_ERROR ||
-      pAction->errCode == TSDB_CODE_SYN_INTERNAL_ERROR || pAction->errCode == TSDB_CODE_SYN_NOT_LEADER) {
+  if (pAction->errCode == TSDB_CODE_SYN_NEW_CONFIG_ERROR || pAction->errCode == TSDB_CODE_SYN_INTERNAL_ERROR ||
+      pAction->errCode == TSDB_CODE_SYN_NOT_LEADER) {
     pAction->epSet.inUse = (pAction->epSet.inUse + 1) % pAction->epSet.numOfEps;
     mInfo("trans:%d, %s:%d execute status is reset and set epset inuse:%d", pTrans->id, mndTransStr(pAction->stage),
           pAction->id, pAction->epSet.inUse);

source/dnode/mnode/impl/test/acct/acct.cpp

@@ -32,7 +32,7 @@ TEST_F(MndTestAcct, 01_Create_Acct) {
 
   SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_ACCT, pReq, contLen);
   ASSERT_NE(pRsp, nullptr);
-  ASSERT_EQ(pRsp->code, TSDB_CODE_MSG_NOT_PROCESSED);
+  ASSERT_EQ(pRsp->code, TSDB_CODE_OPS_NOT_SUPPORT);
 }
 
 TEST_F(MndTestAcct, 02_Alter_Acct) {
@@ -42,7 +42,7 @@ TEST_F(MndTestAcct, 02_Alter_Acct) {
 
   SRpcMsg* pRsp = test.SendReq(TDMT_MND_ALTER_ACCT, pReq, contLen);
   ASSERT_NE(pRsp, nullptr);
-  ASSERT_EQ(pRsp->code, TSDB_CODE_MSG_NOT_PROCESSED);
+  ASSERT_EQ(pRsp->code, TSDB_CODE_OPS_NOT_SUPPORT);
 }
 
 TEST_F(MndTestAcct, 03_Drop_Acct) {
@@ -52,5 +52,5 @@ TEST_F(MndTestAcct, 03_Drop_Acct) {
 
   SRpcMsg* pRsp = test.SendReq(TDMT_MND_DROP_ACCT, pReq, contLen);
   ASSERT_NE(pRsp, nullptr);
-  ASSERT_EQ(pRsp->code, TSDB_CODE_MSG_NOT_PROCESSED);
+  ASSERT_EQ(pRsp->code, TSDB_CODE_OPS_NOT_SUPPORT);
 }

source/dnode/vnode/inc/vnode.h

@@ -169,14 +169,13 @@ typedef struct STsdbReader STsdbReader;
 
 int32_t tsdbSetTableList(STsdbReader *pReader, const void *pTableList, int32_t num);
 int32_t tsdbReaderOpen(SVnode *pVnode, SQueryTableDataCond *pCond, void *pTableList, int32_t numOfTables,
-                       STsdbReader **ppReader, const char *idstr);
+                       SSDataBlock *pResBlock, STsdbReader **ppReader, const char *idstr);
 
 void         tsdbReaderClose(STsdbReader *pReader);
 bool         tsdbNextDataBlock(STsdbReader *pReader);
-bool     tsdbTableNextDataBlock(STsdbReader *pReader, uint64_t uid);
 void         tsdbRetrieveDataBlockInfo(const STsdbReader *pReader, int32_t *rows, uint64_t *uid, STimeWindow *pWindow);
-int32_t  tsdbRetrieveDatablockSMA(STsdbReader *pReader, SColumnDataAgg ***pBlockStatis, bool *allHave);
-SArray  *tsdbRetrieveDataBlock(STsdbReader *pTsdbReadHandle, SArray *pColumnIdList);
+int32_t      tsdbRetrieveDatablockSMA(STsdbReader *pReader, SColumnDataAgg ***pBlockSMA, bool *allHave);
+SSDataBlock *tsdbRetrieveDataBlock(STsdbReader *pTsdbReadHandle, SArray *pColumnIdList);
 int32_t      tsdbReaderReset(STsdbReader *pReader, SQueryTableDataCond *pCond);
 int32_t      tsdbGetFileBlocksDistInfo(STsdbReader *pReader, STableBlockDistInfo *pTableBlockInfo);
 int64_t      tsdbGetNumOfRowsInMemTable(STsdbReader *pHandle);

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

@@ -100,7 +100,7 @@ int32_t vnodeSyncOpen(SVnode* pVnode, char* path);
 int32_t vnodeSyncStart(SVnode* pVnode);
 void    vnodeSyncPreClose(SVnode* pVnode);
 void    vnodeSyncClose(SVnode* pVnode);
-void    vnodeRedirectRpcMsg(SVnode* pVnode, SRpcMsg* pMsg);
+void    vnodeRedirectRpcMsg(SVnode* pVnode, SRpcMsg* pMsg, int32_t code);
 bool    vnodeIsLeader(SVnode* pVnode);
 bool    vnodeIsRoleLeader(SVnode* pVnode);
 

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

@@ -228,8 +228,7 @@ int32_t tsdbCacheInsertLastrow(SLRUCache *pCache, STsdb *pTsdb, tb_uid_t uid, ST
             invalidate = true;
 
             break;
-          }
-        } else {
+          } else {  // new inserting key is greater than cached, update cached entry
             SLastCol lastCol = {.ts = keyTs, .colVal = colVal};
             if (IS_VAR_DATA_TYPE(colVal.type) && colVal.value.nData > 0) {
               SLastCol *pLastCol = (SLastCol *)taosArrayGet(pLast, iCol);
@@ -248,70 +247,16 @@ int32_t tsdbCacheInsertLastrow(SLRUCache *pCache, STsdb *pTsdb, tb_uid_t uid, ST
           }
         }
       }
+    }
 
   _invalidate:
     taosMemoryFreeClear(pTSchema);
 
     taosLRUCacheRelease(pCache, h, invalidate);
-    /*
-    cacheRow = (STSRow *)taosLRUCacheValue(pCache, h);
-    if (row->ts >= cacheRow->ts) {
-      if (row->ts == cacheRow->ts) {
-        STSRow    *mergedRow = NULL;
-        SRowMerger merger = {0};
-        STSchema  *pTSchema = metaGetTbTSchema(pTsdb->pVnode->pMeta, uid, -1, 1);
-
-        tRowMergerInit(&merger, &tsdbRowFromTSRow(0, cacheRow), pTSchema);
-
-        tRowMerge(&merger, &tsdbRowFromTSRow(1, row));
-
-        tRowMergerGetRow(&merger, &mergedRow);
-        tRowMergerClear(&merger);
-
-        taosMemoryFreeClear(pTSchema);
-
-        row = mergedRow;
-        dup = false;
-      }
-
-      if (TD_ROW_LEN(row) <= TD_ROW_LEN(cacheRow)) {
-        tdRowCpy(cacheRow, row);
-        if (!dup) {
-          taosMemoryFree(row);
-        }
-
-        taosLRUCacheRelease(pCache, h, false);
-      } else {
-        taosLRUCacheRelease(pCache, h, true);
-        // tsdbCacheDeleteLastrow(pCache, uid, TSKEY_MAX);
-        if (dup) {
-          cacheRow = tdRowDup(row);
-        } else {
-          cacheRow = row;
-        }
-        _taos_lru_deleter_t deleter = deleteTableCacheLastrow;
-        LRUStatus status = taosLRUCacheInsert(pCache, key, keyLen, cacheRow, TD_ROW_LEN(cacheRow), deleter, NULL,
-                                              TAOS_LRU_PRIORITY_LOW);
-        if (status != TAOS_LRU_STATUS_OK) {
-          code = -1;
-        }
-        // tsdbCacheInsertLastrow(pCache, uid, row, dup);
-      }
-    }*/
-  } /*else {
-    if (dup) {
-      cacheRow = tdRowDup(row);
-    } else {
-      cacheRow = row;
+    if (invalidate) {
+      taosLRUCacheErase(pCache, key, keyLen);
     }
-
-    _taos_lru_deleter_t deleter = deleteTableCacheLastrow;
-    LRUStatus           status =
-        taosLRUCacheInsert(pCache, key, keyLen, cacheRow, TD_ROW_LEN(cacheRow), deleter, NULL, TAOS_LRU_PRIORITY_LOW);
-    if (status != TAOS_LRU_STATUS_OK) {
-      code = -1;
   }
-    }*/
 
   return code;
 }
@@ -349,12 +294,11 @@ int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, STSRow *row, STsdb
 
         SColVal colVal = {0};
         tTSRowGetVal(row, pTSchema, iCol, &colVal);
-        if (!COL_VAL_IS_VALUE(&colVal)) {
+        if (COL_VAL_IS_VALUE(&colVal)) {
           if (keyTs == tTsVal1->ts && COL_VAL_IS_VALUE(tColVal)) {
             invalidate = true;
 
             break;
-          }
           } else {
             SLastCol lastCol = {.ts = keyTs, .colVal = colVal};
             if (IS_VAR_DATA_TYPE(colVal.type) && colVal.value.nData > 0) {
@@ -374,14 +318,15 @@ int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, STSRow *row, STsdb
           }
         }
       }
+    }
 
   _invalidate:
     taosMemoryFreeClear(pTSchema);
 
     taosLRUCacheRelease(pCache, h, invalidate);
-
-    // clear last cache anyway, lazy load when get last lookup
-    // taosLRUCacheRelease(pCache, h, true);
+    if (invalidate) {
+      taosLRUCacheErase(pCache, key, keyLen);
+    }
   }
 
   return code;

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

@@ -344,7 +344,7 @@ int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg) {
   vTrace("message in vnode query queue is processing");
   // if ((pMsg->msgType == TDMT_SCH_QUERY) && !vnodeIsLeader(pVnode)) {
   if ((pMsg->msgType == TDMT_SCH_QUERY) && !syncIsReadyForRead(pVnode->sync)) {
-    vnodeRedirectRpcMsg(pVnode, pMsg);
+    vnodeRedirectRpcMsg(pVnode, pMsg, terrno);
     return 0;
   }
 
@@ -367,12 +367,12 @@ int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo) {
        pMsg->msgType == TDMT_VND_BATCH_META) &&
       !syncIsReadyForRead(pVnode->sync)) {
     //      !vnodeIsLeader(pVnode)) {
-    vnodeRedirectRpcMsg(pVnode, pMsg);
+    vnodeRedirectRpcMsg(pVnode, pMsg, terrno);
     return 0;
   }
 
   if (pMsg->msgType == TDMT_VND_TMQ_CONSUME && !pVnode->restored) {
-    vnodeRedirectRpcMsg(pVnode, pMsg);
+    vnodeRedirectRpcMsg(pVnode, pMsg, TSDB_CODE_SYN_RESTORING);
     return 0;
   }
 

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

@@ -16,7 +16,7 @@
 #define _DEFAULT_SOURCE
 #include "vnd.h"
 
-#define BATCH_DISABLE 1
+#define BATCH_ENABLE 0
 
 static inline bool vnodeIsMsgBlock(tmsg_t type) {
   return (type == TDMT_VND_CREATE_TABLE) || (type == TDMT_VND_ALTER_TABLE) || (type == TDMT_VND_DROP_TABLE) ||
@@ -53,7 +53,7 @@ static inline void vnodePostBlockMsg(SVnode *pVnode, const SRpcMsg *pMsg) {
   }
 }
 
-void vnodeRedirectRpcMsg(SVnode *pVnode, SRpcMsg *pMsg) {
+void vnodeRedirectRpcMsg(SVnode *pVnode, SRpcMsg *pMsg, int32_t code) {
   SEpSet newEpSet = {0};
   syncGetRetryEpSet(pVnode->sync, &newEpSet);
 
@@ -66,8 +66,20 @@ void vnodeRedirectRpcMsg(SVnode *pVnode, SRpcMsg *pMsg) {
   }
   pMsg->info.hasEpSet = 1;
 
-  SRpcMsg rsp = {.code = TSDB_CODE_SYN_NOT_LEADER, .info = pMsg->info, .msgType = pMsg->msgType + 1};
-  tmsgSendRedirectRsp(&rsp, &newEpSet);
+  if (code == 0) code = TSDB_CODE_SYN_NOT_LEADER;
+
+  SRpcMsg rsp = {.code = code, .info = pMsg->info, .msgType = pMsg->msgType + 1};
+  int32_t contLen = tSerializeSEpSet(NULL, 0, &newEpSet);
+
+  rsp.pCont = rpcMallocCont(contLen);
+  if (rsp.pCont == NULL) {
+    pMsg->code = TSDB_CODE_OUT_OF_MEMORY;
+  } else {
+    tSerializeSEpSet(rsp.pCont, contLen, &newEpSet);
+    rsp.contLen = contLen;
+  }
+
+  tmsgSendRsp(&rsp);
 }
 
 static void inline vnodeHandleWriteMsg(SVnode *pVnode, SRpcMsg *pMsg) {
@@ -87,8 +99,8 @@ static void inline vnodeHandleWriteMsg(SVnode *pVnode, SRpcMsg *pMsg) {
 }
 
 static void vnodeHandleProposeError(SVnode *pVnode, SRpcMsg *pMsg, int32_t code) {
-  if (code == TSDB_CODE_SYN_NOT_LEADER) {
-    vnodeRedirectRpcMsg(pVnode, pMsg);
+  if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING) {
+    vnodeRedirectRpcMsg(pVnode, pMsg, code);
   } else {
     const STraceId *trace = &pMsg->info.traceId;
     vGError("vgId:%d, msg:%p failed to propose since %s, code:0x%x", pVnode->config.vgId, pMsg, tstrerror(code), code);
@@ -99,15 +111,12 @@ static void vnodeHandleProposeError(SVnode *pVnode, SRpcMsg *pMsg, int32_t code)
   }
 }
 
+#if BATCH_ENABLE
+
 static void inline vnodeProposeBatchMsg(SVnode *pVnode, SRpcMsg **pMsgArr, bool *pIsWeakArr, int32_t *arrSize) {
   if (*arrSize <= 0) return;
 
-#if BATCH_DISABLE
-  int32_t code = syncPropose(pVnode->sync, pMsgArr[0], pIsWeakArr[0]);
-#else
   int32_t code = syncProposeBatch(pVnode->sync, pMsgArr, pIsWeakArr, *arrSize);
-#endif
-
   if (code > 0) {
     for (int32_t i = 0; i < *arrSize; ++i) {
       vnodeHandleWriteMsg(pVnode, pMsgArr[i]);
@@ -177,7 +186,7 @@ void vnodeProposeWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs)
       continue;
     }
 
-    if (isBlock || BATCH_DISABLE) {
+    if (isBlock) {
       vnodeProposeBatchMsg(pVnode, pMsgArr, pIsWeakArr, &arrayPos);
     }
 
@@ -185,7 +194,7 @@ void vnodeProposeWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs)
     pIsWeakArr[arrayPos] = isWeak;
     arrayPos++;
 
-    if (isBlock || msg == numOfMsgs - 1 || BATCH_DISABLE) {
+    if (isBlock || msg == numOfMsgs - 1) {
       vnodeProposeBatchMsg(pVnode, pMsgArr, pIsWeakArr, &arrayPos);
     }
   }
@@ -194,6 +203,65 @@ void vnodeProposeWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs)
   taosMemoryFree(pIsWeakArr);
 }
 
+#else
+
+static int32_t inline vnodeProposeMsg(SVnode *pVnode, SRpcMsg *pMsg, bool isWeak) {
+  int32_t code = syncPropose(pVnode->sync, pMsg, isWeak);
+  if (code > 0) {
+    vnodeHandleWriteMsg(pVnode, pMsg);
+  } else if (code == 0) {
+    vnodeWaitBlockMsg(pVnode, pMsg);
+  } else {
+    if (terrno != 0) code = terrno;
+    vnodeHandleProposeError(pVnode, pMsg, code);
+  }
+
+  return code;
+}
+
+void vnodeProposeWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
+  SVnode  *pVnode = pInfo->ahandle;
+  int32_t  vgId = pVnode->config.vgId;
+  int32_t  code = 0;
+  SRpcMsg *pMsg = NULL;
+  vTrace("vgId:%d, get %d msgs from vnode-write queue", vgId, numOfMsgs);
+
+  for (int32_t msg = 0; msg < numOfMsgs; msg++) {
+    if (taosGetQitem(qall, (void **)&pMsg) == 0) continue;
+    bool isWeak = vnodeIsMsgWeak(pMsg->msgType);
+
+    const STraceId *trace = &pMsg->info.traceId;
+    vGTrace("vgId:%d, msg:%p get from vnode-write queue, weak:%d block:%d msg:%d:%d, handle:%p", vgId, pMsg, isWeak,
+            vnodeIsMsgBlock(pMsg->msgType), msg, numOfMsgs, pMsg->info.handle);
+
+    if (!pVnode->restored) {
+      vGError("vgId:%d, msg:%p failed to process since restore not finished", vgId, pMsg);
+      vnodeHandleProposeError(pVnode, pMsg, TSDB_CODE_SYN_RESTORING);
+      rpcFreeCont(pMsg->pCont);
+      taosFreeQitem(pMsg);
+      continue;
+    }
+
+    code = vnodePreProcessWriteMsg(pVnode, pMsg);
+    if (code != 0) {
+      vGError("vgId:%d, msg:%p failed to pre-process since %s", vgId, pMsg, terrstr());
+      if (terrno != 0) code = terrno;
+      vnodeHandleProposeError(pVnode, pMsg, code);
+      rpcFreeCont(pMsg->pCont);
+      taosFreeQitem(pMsg);
+      continue;
+    }
+
+    code = vnodeProposeMsg(pVnode, pMsg, isWeak);
+
+    vGTrace("vgId:%d, msg:%p is freed, code:0x%x", vgId, pMsg, code);
+    rpcFreeCont(pMsg->pCont);
+    taosFreeQitem(pMsg);
+  }
+}
+
+#endif
+
 void vnodeApplyWriteMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnode  *pVnode = pInfo->ahandle;
   int32_t  vgId = pVnode->config.vgId;

source/libs/executor/inc/executorimpl.h

@@ -489,8 +489,8 @@ typedef struct STableCountScanSupp {
 
   bool groupByDbName;
   bool groupByStbName;
-  char    dbName[TSDB_DB_NAME_LEN];
-  char    stbName[TSDB_TABLE_NAME_LEN];
+  char    dbNameFilter[TSDB_DB_NAME_LEN];
+  char    stbNameFilter[TSDB_TABLE_NAME_LEN];
 
 } STableCountScanSupp;
 
@@ -498,16 +498,10 @@ typedef struct STableCountScanOperatorInfo {
   SReadHandle  readHandle;
   SSDataBlock* pRes;
 
-  SName tableName;
-
-  SNodeList* groupTags;
-  SNodeList* scanCols;
-  SNodeList* pseudoCols;
-
   STableCountScanSupp supp;
 
   int32_t   currGrpIdx;
-  SArray*   stbUidList; // when group by db_name and stable_name
+  SArray*   stbUidList; // when group by db_name and/or stable_name
 } STableCountScanOperatorInfo;
 
 typedef struct SOptrBasicInfo {

source/libs/executor/src/exchangeoperator.c

@@ -198,7 +198,7 @@ static SSDataBlock* doLoadRemoteDataImpl(SOperatorInfo* pOperator) {
   }
 }
 
-static SSDataBlock* doLoadRemoteData(SOperatorInfo* pOperator) {
+static SSDataBlock* loadRemoteData(SOperatorInfo* pOperator) {
   SExchangeInfo* pExchangeInfo = pOperator->info;
   SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
 
@@ -307,7 +307,7 @@ SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode
   pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pDummyBlock->pDataBlock);
 
   pOperator->fpSet =
-      createOperatorFpSet(prepareLoadRemoteData, doLoadRemoteData, NULL, destroyExchangeOperatorInfo, NULL);
+      createOperatorFpSet(prepareLoadRemoteData, loadRemoteData, NULL, destroyExchangeOperatorInfo, NULL);
   return pOperator;
 
 _error:
@@ -570,13 +570,10 @@ int32_t prepareConcurrentlyLoad(SOperatorInfo* pOperator) {
 
   pOperator->status = OP_RES_TO_RETURN;
   pOperator->cost.openCost = taosGetTimestampUs() - startTs;
-
-  tsem_wait(&pExchangeInfo->ready);
   if (isTaskKilled(pTaskInfo)) {
     longjmp(pTaskInfo->env, pTaskInfo->code);
   }
 
-  tsem_post(&pExchangeInfo->ready);
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/executor/src/executil.c

@@ -438,13 +438,12 @@ static SColumnInfoData* getColInfoResult(void* metaHandle, int64_t suid, SArray*
       goto end;
     }
   }
+  removeInvalidTable(uidList, tags);
 
   int32_t rows = taosArrayGetSize(uidList);
   if (rows == 0) {
     goto end;
   }
-  //  int64_t stt1 = taosGetTimestampUs();
-  //  qDebug("generate tag meta rows:%d, cost:%ld us", rows, stt1-stt);
 
   code = blockDataEnsureCapacity(pResBlock, rows);
   if (code != TSDB_CODE_SUCCESS) {
@@ -452,7 +451,6 @@ static SColumnInfoData* getColInfoResult(void* metaHandle, int64_t suid, SArray*
     goto end;
   }
 
-  //  int64_t st = taosGetTimestampUs();
   for (int32_t i = 0; i < rows; i++) {
     int64_t* uid = taosArrayGet(uidList, i);
     for (int32_t j = 0; j < taosArrayGetSize(pResBlock->pDataBlock); j++) {
@@ -467,7 +465,9 @@ static SColumnInfoData* getColInfoResult(void* metaHandle, int64_t suid, SArray*
 #endif
       } else {
         void* tag = taosHashGet(tags, uid, sizeof(int64_t));
-        ASSERT(tag);
+        if (tag == NULL) {
+          continue;
+        }
         STagVal tagVal = {0};
         tagVal.cid = pColInfo->info.colId;
         const char* p = metaGetTableTagVal(tag, pColInfo->info.type, &tagVal);
@@ -925,12 +925,12 @@ static int32_t optimizeTbnameInCondImpl(void* metaHandle, int64_t suid, SArray*
 
     SArray*   pTbList = getTableNameList(pList);
     int32_t   numOfTables = taosArrayGetSize(pTbList);
-    SHashObj *uHash = NULL;
+    SHashObj* uHash = NULL;
     size_t    listlen = taosArrayGetSize(list);  // len > 0 means there already have uids
     if (listlen > 0) {
       uHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
       for (int i = 0; i < listlen; i++) {
-        int64_t *uid = taosArrayGet(list, i);
+        int64_t* uid = taosArrayGet(list, i);
         taosHashPut(uHash, uid, sizeof(int64_t), &i, sizeof(i));
       }
     }
@@ -1241,6 +1241,7 @@ int32_t extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNod
     }
   }
 
+  // set the output flag for each column in SColMatchInfo, according to the
   *numOfOutputCols = 0;
   int32_t num = LIST_LENGTH(pOutputNodeList->pSlots);
   for (int32_t i = 0; i < num; ++i) {
@@ -1537,8 +1538,6 @@ SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput,
     pCtx->start.key = INT64_MIN;
     pCtx->end.key = INT64_MIN;
     pCtx->numOfParams = pExpr->base.numOfParams;
-    pCtx->isStream = false;
-
     pCtx->param = pFunct->pParam;
     pCtx->saveHandle.currentPage = -1;
   }
@@ -1602,20 +1601,22 @@ SColumn extractColumnFromColumnNode(SColumnNode* pColNode) {
 int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysiNode* pTableScanNode) {
   pCond->order = pTableScanNode->scanSeq[0] > 0 ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
   pCond->numOfCols = LIST_LENGTH(pTableScanNode->scan.pScanCols);
+
   pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));
-  if (pCond->colList == NULL) {
+  pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t)*pCond->numOfCols);
+  if (pCond->colList == NULL || pCond->pSlotList == NULL) {
     terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
+    taosMemoryFreeClear(pCond->colList);
+    taosMemoryFreeClear(pCond->pSlotList);
     return terrno;
   }
 
-  // pCond->twindow = pTableScanNode->scanRange;
   // TODO: get it from stable scan node
   pCond->twindows = pTableScanNode->scanRange;
   pCond->suid = pTableScanNode->scan.suid;
   pCond->type = TIMEWINDOW_RANGE_CONTAINED;
   pCond->startVersion = -1;
   pCond->endVersion = -1;
-  //  pCond->type = pTableScanNode->scanFlag;
 
   int32_t j = 0;
   for (int32_t i = 0; i < pCond->numOfCols; ++i) {
@@ -1628,6 +1629,8 @@ int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysi
     pCond->colList[j].type = pColNode->node.resType.type;
     pCond->colList[j].bytes = pColNode->node.resType.bytes;
     pCond->colList[j].colId = pColNode->colId;
+
+    pCond->pSlotList[j] = pNode->slotId;
     j += 1;
   }
 
@@ -1635,7 +1638,10 @@ int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysi
   return TSDB_CODE_SUCCESS;
 }
 
-void cleanupQueryTableDataCond(SQueryTableDataCond* pCond) { taosMemoryFreeClear(pCond->colList); }
+void cleanupQueryTableDataCond(SQueryTableDataCond* pCond) {
+  taosMemoryFreeClear(pCond->colList);
+  taosMemoryFreeClear(pCond->pSlotList);
+}
 
 int32_t convertFillType(int32_t mode) {
   int32_t type = TSDB_FILL_NONE;
@@ -1965,7 +1971,7 @@ int32_t buildGroupIdMapForAllTables(STableListInfo* pTableListInfo, SReadHandle*
 
 int32_t createScanTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags, bool groupSort, SReadHandle* pHandle,
                                 STableListInfo* pTableListInfo, SNode* pTagCond, SNode* pTagIndexCond,
-                                struct SExecTaskInfo* pTaskInfo) {
+                                SExecTaskInfo* pTaskInfo) {
   int64_t     st = taosGetTimestampUs();
   const char* idStr = GET_TASKID(pTaskInfo);
 

source/libs/executor/src/executor.c

@@ -971,21 +971,27 @@ int32_t initQueryTableDataCondForTmq(SQueryTableDataCond* pCond, SSnapContext* s
   pCond->order = TSDB_ORDER_ASC;
   pCond->numOfCols = pMtInfo->schema->nCols;
   pCond->colList = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnInfo));
-  if (pCond->colList == NULL) {
+  pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t) * pCond->numOfCols);
+  if (pCond->colList == NULL || pCond->pSlotList == NULL) {
+    taosMemoryFreeClear(pCond->colList);
+    taosMemoryFreeClear(pCond->pSlotList);
     terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
     return terrno;
   }
 
-  pCond->twindows = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
+  pCond->twindows = TSWINDOW_INITIALIZER;
   pCond->suid = pMtInfo->suid;
   pCond->type = TIMEWINDOW_RANGE_CONTAINED;
   pCond->startVersion = -1;
   pCond->endVersion = sContext->snapVersion;
 
   for (int32_t i = 0; i < pCond->numOfCols; ++i) {
-    pCond->colList[i].type = pMtInfo->schema->pSchema[i].type;
-    pCond->colList[i].bytes = pMtInfo->schema->pSchema[i].bytes;
-    pCond->colList[i].colId = pMtInfo->schema->pSchema[i].colId;
+    SColumnInfo* pColInfo = &pCond->colList[i];
+    pColInfo->type = pMtInfo->schema->pSchema[i].type;
+    pColInfo->bytes = pMtInfo->schema->pSchema[i].bytes;
+    pColInfo->colId = pMtInfo->schema->pSchema[i].colId;
+
+    pCond->pSlotList[i] = i;
   }
 
   return TSDB_CODE_SUCCESS;
@@ -1078,7 +1084,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
         int32_t        num = tableListGetSize(pTaskInfo->pTableInfoList);
 
         if (tsdbReaderOpen(pTableScanInfo->base.readHandle.vnode, &pTableScanInfo->base.cond, pList, num,
-                           &pTableScanInfo->base.dataReader, NULL) < 0 ||
+                           pTableScanInfo->pResBlock, &pTableScanInfo->base.dataReader, NULL) < 0 ||
             pTableScanInfo->base.dataReader == NULL) {
           ASSERT(0);
         }
@@ -1130,7 +1136,7 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT
     int32_t        size = tableListGetSize(pTaskInfo->pTableInfoList);
     ASSERT(size == 1);
 
-    tsdbReaderOpen(pInfo->vnode, &pTaskInfo->streamInfo.tableCond, pList, size, &pInfo->dataReader, NULL);
+    tsdbReaderOpen(pInfo->vnode, &pTaskInfo->streamInfo.tableCond, pList, size, NULL, &pInfo->dataReader, NULL);
 
     cleanupQueryTableDataCond(&pTaskInfo->streamInfo.tableCond);
     strcpy(pTaskInfo->streamInfo.tbName, mtInfo.tbName);

source/libs/executor/src/executorimpl.c

@@ -1164,8 +1164,6 @@ void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SG
 //       T_LONG_JMP(pRuntimeEnv->env, TSDB_CODE_TSC_QUERY_CANCELLED);
 //     }
 //
-//     tsdbRetrieveDataBlockInfo(pTsdbReadHandle, &blockInfo);
-//
 //     if (pQueryAttr->limit.offset > blockInfo.rows) {
 //       pQueryAttr->limit.offset -= blockInfo.rows;
 //       pTableQueryInfo->lastKey = (QUERY_IS_ASC_QUERY(pQueryAttr)) ? blockInfo.window.ekey : blockInfo.window.skey;
@@ -1641,7 +1639,7 @@ int32_t doInitAggInfoSup(SAggSupporter* pAggSup, SqlFunctionCtx* pCtx, int32_t n
   pAggSup->currentPageId = -1;
   pAggSup->resultRowSize = getResultRowSize(pCtx, numOfOutput);
   pAggSup->keyBuf = taosMemoryCalloc(1, keyBufSize + POINTER_BYTES + sizeof(int64_t));
-  pAggSup->pResultRowHashTable = tSimpleHashInit(10, hashFn);
+  pAggSup->pResultRowHashTable = tSimpleHashInit(100, hashFn);
 
   if (pAggSup->keyBuf == NULL || pAggSup->pResultRowHashTable == NULL) {
     return TSDB_CODE_OUT_OF_MEMORY;

source/libs/executor/src/sortoperator.c

@@ -559,7 +559,7 @@ typedef struct SMultiwayMergeOperatorInfo {
   STupleHandle*  prefetchedTuple;
 } SMultiwayMergeOperatorInfo;
 
-int32_t doOpenMultiwayMergeOperator(SOperatorInfo* pOperator) {
+int32_t openMultiwayMergeOperator(SOperatorInfo* pOperator) {
   SMultiwayMergeOperatorInfo* pInfo = pOperator->info;
   SExecTaskInfo*              pTaskInfo = pOperator->pTaskInfo;
 
@@ -577,9 +577,15 @@ int32_t doOpenMultiwayMergeOperator(SOperatorInfo* pOperator) {
   tsortSetCompareGroupId(pInfo->pSortHandle, pInfo->groupSort);
 
   for (int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
+    SOperatorInfo* pDownstream = pOperator->pDownstream[i];
+    if (pDownstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_EXCHANGE) {
+      pDownstream->fpSet._openFn(pDownstream);
+    }
+
     SSortSource* ps = taosMemoryCalloc(1, sizeof(SSortSource));
-    ps->param = pOperator->pDownstream[i];
+    ps->param = pDownstream;
     ps->onlyRef = true;
+
     tsortAddSource(pInfo->pSortHandle, ps);
   }
 
@@ -714,7 +720,6 @@ SSDataBlock* doMultiwayMerge(SOperatorInfo* pOperator) {
   }
 
   qDebug("start to merge final sorted rows, %s", GET_TASKID(pTaskInfo));
-
   SSDataBlock* pBlock = getMultiwaySortedBlockData(pInfo->pSortHandle, pInfo->binfo.pRes, pInfo->matchInfo.pList, pOperator);
   if (pBlock != NULL) {
     pOperator->resultInfo.totalRows += pBlock->info.rows;
@@ -781,7 +786,7 @@ SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** downStreams, size
   SPhysiNode*  pChildNode = (SPhysiNode*)nodesListGetNode(pPhyNode->pChildren, 0);
   SSDataBlock* pInputBlock = createDataBlockFromDescNode(pChildNode->pOutputDataBlockDesc);
 
-  initResultSizeInfo(&pOperator->resultInfo, 4096);
+  initResultSizeInfo(&pOperator->resultInfo, 1024);
   blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
 
   pInfo->groupSort = pMergePhyNode->groupSort;
@@ -792,7 +797,7 @@ SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** downStreams, size
   pInfo->sortBufSize = pInfo->bufPageSize * (numStreams + 1);  // one additional is reserved for merged result.
 
   setOperatorInfo(pOperator, "MultiwayMergeOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE, false, OP_NOT_OPENED, pInfo, pTaskInfo);
-  pOperator->fpSet = createOperatorFpSet(doOpenMultiwayMergeOperator, doMultiwayMerge, NULL,
+  pOperator->fpSet = createOperatorFpSet(openMultiwayMergeOperator, doMultiwayMerge, NULL,
                                          destroyMultiwayMergeOperatorInfo, getMultiwayMergeExplainExecInfo);
 
   code = appendDownstream(pOperator, downStreams, numStreams);

source/libs/executor/src/sysscanoperator.c

@@ -1879,7 +1879,8 @@ static int32_t initTableblockDistQueryCond(uint64_t uid, SQueryTableDataCond* pC
   pCond->order = TSDB_ORDER_ASC;
   pCond->numOfCols = 1;
   pCond->colList = taosMemoryCalloc(1, sizeof(SColumnInfo));
-    if (pCond->colList == NULL) {
+  pCond->pSlotList = taosMemoryMalloc(sizeof(int32_t));
+  if (pCond->colList == NULL || pCond->pSlotList == NULL) {
     terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
     return terrno;
   }
@@ -1888,6 +1889,8 @@ static int32_t initTableblockDistQueryCond(uint64_t uid, SQueryTableDataCond* pC
   pCond->colList->type = TSDB_DATA_TYPE_TIMESTAMP;
   pCond->colList->bytes = sizeof(TSKEY);
 
+  pCond->pSlotList[0] = 0;
+
   pCond->twindows = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
   pCond->suid = uid;
   pCond->type = TIMEWINDOW_RANGE_CONTAINED;
@@ -1906,9 +1909,11 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi
     goto _error;
   }
 
+  pInfo->pResBlock = createDataBlockFromDescNode(pBlockScanNode->node.pOutputDataBlockDesc);
+  blockDataEnsureCapacity(pInfo->pResBlock, 1);
+
   {
     SQueryTableDataCond cond = {0};
-
     int32_t             code = initTableblockDistQueryCond(pBlockScanNode->suid, &cond);
     if (code != TSDB_CODE_SUCCESS) {
       goto _error;
@@ -1918,7 +1923,7 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi
     size_t          num = tableListGetSize(pTableListInfo);
     void*           pList = tableListGetInfo(pTableListInfo, 0);
 
-        code = tsdbReaderOpen(readHandle->vnode, &cond, pList, num, &pInfo->pHandle, pTaskInfo->id.str);
+    code = tsdbReaderOpen(readHandle->vnode, &cond, pList, num, pInfo->pResBlock, &pInfo->pHandle, pTaskInfo->id.str);
     cleanupQueryTableDataCond(&cond);
     if (code != 0) {
       goto _error;
@@ -1928,9 +1933,6 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi
   pInfo->readHandle = *readHandle;
   pInfo->uid = pBlockScanNode->suid;
 
-    pInfo->pResBlock = createDataBlockFromDescNode(pBlockScanNode->node.pOutputDataBlockDesc);
-    blockDataEnsureCapacity(pInfo->pResBlock, 1);
-
   int32_t    numOfCols = 0;
   SExprInfo* pExprInfo = createExprInfo(pBlockScanNode->pScanPseudoCols, NULL, &numOfCols);
   int32_t    code = initExprSupp(&pOperator->exprSupp, pExprInfo, numOfCols);
@@ -1944,7 +1946,7 @@ SOperatorInfo* createDataBlockInfoScanOperator(SReadHandle* readHandle, SBlockDi
       createOperatorFpSet(operatorDummyOpenFn, doBlockInfoScan, NULL, destroyBlockDistScanOperatorInfo, NULL);
   return pOperator;
 
-    _error:
+_error:
   taosMemoryFreeClear(pInfo);
   taosMemoryFreeClear(pOperator);
   return NULL;

source/libs/executor/src/tfill.c

@@ -45,8 +45,18 @@ static void setNullRow(SSDataBlock* pBlock, SFillInfo* pFillInfo, int32_t rowInd
     if (pCol->notFillCol) {
       bool filled = fillIfWindowPseudoColumn(pFillInfo, pCol, pDstColInfo, rowIndex);
       if (!filled) {
-        SArray*     p = FILL_IS_ASC_FILL(pFillInfo) ? pFillInfo->prev.pRowVal : pFillInfo->next.pRowVal;
-        SGroupKeys* pKey = taosArrayGet(p, i);
+        SRowVal* p = NULL;
+        if (FILL_IS_ASC_FILL(pFillInfo)) {
+          if (pFillInfo->prev.key != 0) {
+            p = &pFillInfo->prev;  // prev has been set value
+          } else { // otherwise, use the value in the next row
+            p = &pFillInfo->next;
+          }
+        } else {
+          p = &pFillInfo->next;
+        }
+
+        SGroupKeys* pKey = taosArrayGet(p->pRowVal, i);
         doSetVal(pDstColInfo, rowIndex, pKey);
       }
     } else {
@@ -246,7 +256,10 @@ static void initBeforeAfterDataBuf(SFillInfo* pFillInfo) {
 
 static void saveColData(SArray* rowBuf, int32_t columnIndex, const char* src, bool isNull);
 
-static void copyCurrentRowIntoBuf(SFillInfo* pFillInfo, int32_t rowIndex, SArray* pRow) {
+static void copyCurrentRowIntoBuf(SFillInfo* pFillInfo, int32_t rowIndex, SRowVal* pRowVal) {
+  SColumnInfoData* pTsCol = taosArrayGet(pFillInfo->pSrcBlock->pDataBlock, pFillInfo->srcTsSlotId);
+  pRowVal->key = ((int64_t*)pTsCol->pData)[rowIndex];
+
   for (int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
     int32_t type = pFillInfo->pFillCol[i].pExpr->pExpr->nodeType;
     if (type == QUERY_NODE_COLUMN || type == QUERY_NODE_OPERATOR || type == QUERY_NODE_FUNCTION) {
@@ -257,7 +270,7 @@ static void copyCurrentRowIntoBuf(SFillInfo* pFillInfo, int32_t rowIndex, SArray
       bool  isNull = colDataIsNull_s(pSrcCol, rowIndex);
       char* p = colDataGetData(pSrcCol, rowIndex);
 
-      saveColData(pRow, i, p, isNull);
+      saveColData(pRowVal->pRowVal, i, p, isNull);
     } else {
       ASSERT(0);
     }
@@ -281,7 +294,7 @@ static int32_t fillResultImpl(SFillInfo* pFillInfo, SSDataBlock* pBlock, int32_t
 
     // set the next value for interpolation
     if ((pFillInfo->currentKey < ts && ascFill) || (pFillInfo->currentKey > ts && !ascFill)) {
-      copyCurrentRowIntoBuf(pFillInfo, pFillInfo->index, pFillInfo->next.pRowVal);
+      copyCurrentRowIntoBuf(pFillInfo, pFillInfo->index, &pFillInfo->next);
     }
 
     if (((pFillInfo->currentKey < ts && ascFill) || (pFillInfo->currentKey > ts && !ascFill)) &&
@@ -303,7 +316,7 @@ static int32_t fillResultImpl(SFillInfo* pFillInfo, SSDataBlock* pBlock, int32_t
 
       if (pFillInfo->type == TSDB_FILL_NEXT && (pFillInfo->index + 1) < pFillInfo->numOfRows) {
         int32_t nextRowIndex = pFillInfo->index + 1;
-        copyCurrentRowIntoBuf(pFillInfo, nextRowIndex, pFillInfo->next.pRowVal);
+        copyCurrentRowIntoBuf(pFillInfo, nextRowIndex, &pFillInfo->next);
       }
 
       // copy rows to dst buffer
@@ -319,6 +332,9 @@ static int32_t fillResultImpl(SFillInfo* pFillInfo, SSDataBlock* pBlock, int32_t
         if (!colDataIsNull_s(pSrc, pFillInfo->index)) {
           colDataAppend(pDst, index, src, false);
           saveColData(pFillInfo->prev.pRowVal, i, src, false);
+          if (pFillInfo->srcTsSlotId == dstSlotId) {
+            pFillInfo->prev.key = *(int64_t*)src;
+          }
         } else {  // the value is null
           if (pDst->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
             colDataAppend(pDst, index, (const char*)&pFillInfo->currentKey, false);

source/libs/executor/src/timewindowoperator.c

@@ -1276,7 +1276,6 @@ static SSDataBlock* doBuildIntervalResult(SOperatorInfo* pOperator) {
     return NULL;
   }
 
-  blockDataEnsureCapacity(pBlock, pOperator->resultInfo.capacity);
   while (1) {
     doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
     doFilter(pBlock, pOperator->exprSupp.pFilterInfo, NULL);
@@ -1649,6 +1648,16 @@ static bool allInvertible(SqlFunctionCtx* pFCtx, int32_t numOfCols) {
 static bool timeWindowinterpNeeded(SqlFunctionCtx* pCtx, int32_t numOfCols, SIntervalAggOperatorInfo* pInfo) {
   // the primary timestamp column
   bool needed = false;
+
+  for(int32_t i = 0; i < numOfCols; ++i) {
+    SExprInfo* pExpr = pCtx[i].pExpr;
+    if (fmIsIntervalInterpoFunc(pCtx[i].functionId)) {
+      needed = true;
+      break;
+    }
+  }
+
+  if (needed) {
     pInfo->pInterpCols = taosArrayInit(4, sizeof(SColumn));
     pInfo->pPrevValues = taosArrayInit(4, sizeof(SGroupKeys));
 
@@ -1660,13 +1669,14 @@ static bool timeWindowinterpNeeded(SqlFunctionCtx* pCtx, int32_t numOfCols, SInt
       c.bytes = sizeof(int64_t);
       taosArrayPush(pInfo->pInterpCols, &c);
 
-    SGroupKeys key = {0};
+      SGroupKeys key;
       key.bytes = c.bytes;
       key.type = c.type;
       key.isNull = true;  // to denote no value is assigned yet
       key.pData = taosMemoryCalloc(1, c.bytes);
       taosArrayPush(pInfo->pPrevValues, &key);
     }
+  }
 
   for (int32_t i = 0; i < numOfCols; ++i) {
     SExprInfo* pExpr = pCtx[i].pExpr;
@@ -1676,7 +1686,6 @@ static bool timeWindowinterpNeeded(SqlFunctionCtx* pCtx, int32_t numOfCols, SInt
 
       SColumn c = *pParam->pCol;
       taosArrayPush(pInfo->pInterpCols, &c);
-      needed = true;
 
       SGroupKeys key = {0};
       key.bytes = c.bytes;
@@ -1708,7 +1717,7 @@ void initIntervalDownStream(SOperatorInfo* downstream, uint16_t type, SAggSuppor
 
 void initStreamFunciton(SqlFunctionCtx* pCtx, int32_t numOfExpr) {
   for (int32_t i = 0; i < numOfExpr; i++) {
-    pCtx[i].isStream = true;
+//    pCtx[i].isStream = true;
   }
 }
 
@@ -1727,7 +1736,8 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SIntervalPh
   pInfo->primaryTsIndex = ((SColumnNode*)pPhyNode->window.pTspk)->slotId;
 
   size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
-  initResultSizeInfo(&pOperator->resultInfo, 4096);
+  initResultSizeInfo(&pOperator->resultInfo, 512);
+  blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
 
   int32_t    num = 0;
   SExprInfo* pExprInfo = createExprInfo(pPhyNode->window.pFuncs, NULL, &num);
@@ -1773,11 +1783,6 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SIntervalPh
     goto _error;
   }
 
-  if (isStream) {
-    ASSERT(num > 0);
-    initStreamFunciton(pSup->pCtx, pSup->numOfExprs);
-  }
-
   initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->win);
   pInfo->timeWindowInterpo = timeWindowinterpNeeded(pSup->pCtx, num, pInfo);
   if (pInfo->timeWindowInterpo) {
@@ -4314,7 +4319,7 @@ SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream,
   iaInfo->binfo.mergeResultBlock = pNode->window.mergeDataBlock;
 
   size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
-  initResultSizeInfo(&pOperator->resultInfo, 4096);
+  initResultSizeInfo(&pOperator->resultInfo, 512);
 
   int32_t    num = 0;
   SExprInfo* pExprInfo = createExprInfo(pNode->window.pFuncs, NULL, &num);

source/libs/executor/src/tsort.c

@@ -34,10 +34,8 @@ struct SSortHandle {
   int32_t        pageSize;
   int32_t        numOfPages;
   SDiskbasedBuf* pBuf;
-
   SArray*        pSortInfo;
   SArray*        pOrderedSource;
-
   int32_t        loops;
   uint64_t       sortElapsed;
   int64_t        startTs;
@@ -99,9 +97,9 @@ SSortHandle* tsortCreateSortHandle(SArray* pSortInfo, int32_t type, int32_t page
 }
 
 static int32_t sortComparCleanup(SMsortComparParam* cmpParam) {
+  // NOTICE: pSource may be, if it is SORT_MULTISOURCE_MERGE
   for (int32_t i = 0; i < cmpParam->numOfSources; ++i) {
-    SSortSource* pSource =
-        cmpParam->pSources[i];  // NOTICE: pSource may be SGenericSource *, if it is SORT_MULTISOURCE_MERGE
+    SSortSource* pSource = cmpParam->pSources[i];
     blockDataDestroy(pSource->src.pBlock);
     taosMemoryFreeClear(pSource);
   }
@@ -231,15 +229,15 @@ static int32_t doAddToBuf(SSDataBlock* pDataBlock, SSortHandle* pHandle) {
   return doAddNewExternalMemSource(pHandle->pBuf, pHandle->pOrderedSource, pBlock, &pHandle->sourceId, pPageIdList);
 }
 
-static void setCurrentSourceIsDone(SSortSource* pSource, SSortHandle* pHandle) {
+static void setCurrentSourceDone(SSortSource* pSource, SSortHandle* pHandle) {
   pSource->src.rowIndex = -1;
   ++pHandle->numOfCompletedSources;
 }
 
-static int32_t sortComparInit(SMsortComparParam* cmpParam, SArray* pSources, int32_t startIndex, int32_t endIndex,
+static int32_t sortComparInit(SMsortComparParam* pParam, SArray* pSources, int32_t startIndex, int32_t endIndex,
                               SSortHandle* pHandle) {
-  cmpParam->pSources = taosArrayGet(pSources, startIndex);
-  cmpParam->numOfSources = (endIndex - startIndex + 1);
+  pParam->pSources = taosArrayGet(pSources, startIndex);
+  pParam->numOfSources = (endIndex - startIndex + 1);
 
   int32_t code = 0;
 
@@ -247,7 +245,7 @@ static int32_t sortComparInit(SMsortComparParam* cmpParam, SArray* pSources, int
   if (pHandle->pBuf == NULL) {
     if (!osTempSpaceAvailable()) {
       code = TSDB_CODE_NO_AVAIL_DISK;
-      qError("Sort compare init failed since %s", terrstr(code));
+      qError("Sort compare init failed since %s, %s", terrstr(code), pHandle->idStr);
       return code;
     }
 
@@ -260,12 +258,12 @@ static int32_t sortComparInit(SMsortComparParam* cmpParam, SArray* pSources, int
   }
 
   if (pHandle->type == SORT_SINGLESOURCE_SORT) {
-    for (int32_t i = 0; i < cmpParam->numOfSources; ++i) {
-      SSortSource* pSource = cmpParam->pSources[i];
+    for (int32_t i = 0; i < pParam->numOfSources; ++i) {
+      SSortSource* pSource = pParam->pSources[i];
 
       // set current source is done
       if (taosArrayGetSize(pSource->pageIdList) == 0) {
-        setCurrentSourceIsDone(pSource, pHandle);
+        setCurrentSourceDone(pSource, pHandle);
         continue;
       }
 
@@ -280,15 +278,21 @@ static int32_t sortComparInit(SMsortComparParam* cmpParam, SArray* pSources, int
       releaseBufPage(pHandle->pBuf, pPage);
     }
   } else {
-    for (int32_t i = 0; i < cmpParam->numOfSources; ++i) {
-      SSortSource* pSource = cmpParam->pSources[i];
+    qDebug("start init for the multiway merge sort, %s", pHandle->idStr);
+    int64_t st = taosGetTimestampUs();
+
+    for (int32_t i = 0; i < pParam->numOfSources; ++i) {
+      SSortSource* pSource = pParam->pSources[i];
       pSource->src.pBlock = pHandle->fetchfp(pSource->param);
 
       // set current source is done
       if (pSource->src.pBlock == NULL) {
-        setCurrentSourceIsDone(pSource, pHandle);
+        setCurrentSourceDone(pSource, pHandle);
       }
     }
+
+    int64_t et = taosGetTimestampUs();
+    qDebug("init for merge sort completed, elapsed time:%.2f ms, %s", (et - st) / 1000.0, pHandle->idStr);
   }
 
   return code;

source/libs/function/src/udfd.c

@@ -598,8 +598,9 @@ int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
   return 0;
 }
 static bool udfdRpcRfp(int32_t code, tmsg_t msgType) {
-  if (code == TSDB_CODE_RPC_REDIRECT || code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_RPC_BROKEN_LINK ||
-      code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_MNODE_NOT_FOUND|| code == TSDB_CODE_APP_IS_STARTING || code == TSDB_CODE_APP_IS_STOPPING) {
+  if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_RPC_BROKEN_LINK || code == TSDB_CODE_SYN_NOT_LEADER ||
+      code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_MNODE_NOT_FOUND || code == TSDB_CODE_APP_IS_STARTING ||
+      code == TSDB_CODE_APP_IS_STOPPING) {
       if (msgType == TDMT_SCH_QUERY || msgType == TDMT_SCH_MERGE_QUERY || msgType == TDMT_SCH_FETCH || msgType == TDMT_SCH_MERGE_FETCH) {
       return false;
     } 

source/libs/qworker/src/qwDbg.c

@@ -214,20 +214,20 @@ void qwDbgSimulateRedirect(SQWMsg *qwMsg, SQWTaskCtx *ctx, bool *rsped) {
       epSet.eps[2].port = 7300;
 
       ctx->phase = QW_PHASE_POST_QUERY;
-      qwDbgBuildAndSendRedirectRsp(qwMsg->msgType + 1, &qwMsg->connInfo, TSDB_CODE_RPC_REDIRECT, &epSet);
+      qwDbgBuildAndSendRedirectRsp(qwMsg->msgType + 1, &qwMsg->connInfo, TSDB_CODE_SYN_NOT_LEADER, &epSet);
       *rsped = true;
       return;
     }
 
     if (TDMT_SCH_MERGE_QUERY == qwMsg->msgType && (0 == taosRand() % 3)) {
       QW_SET_PHASE(ctx, QW_PHASE_POST_QUERY);
-      qwDbgBuildAndSendRedirectRsp(qwMsg->msgType + 1, &qwMsg->connInfo, TSDB_CODE_RPC_REDIRECT, NULL);
+      qwDbgBuildAndSendRedirectRsp(qwMsg->msgType + 1, &qwMsg->connInfo, TSDB_CODE_SYN_NOT_LEADER, NULL);
       *rsped = true;
       return;
     }
 
     if ((TDMT_SCH_FETCH == qwMsg->msgType) && (0 == taosRand() % 9)) {
-      qwDbgBuildAndSendRedirectRsp(qwMsg->msgType + 1, &qwMsg->connInfo, TSDB_CODE_RPC_REDIRECT, NULL);
+      qwDbgBuildAndSendRedirectRsp(qwMsg->msgType + 1, &qwMsg->connInfo, TSDB_CODE_SYN_NOT_LEADER, NULL);
       *rsped = true;
       return;
     }

source/libs/scalar/src/sclvector.c

@@ -1543,9 +1543,44 @@ void vectorBitOr(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam *pOut,
 int32_t doVectorCompareImpl(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam *pOut, int32_t startIndex,
                             int32_t numOfRows, int32_t step, __compar_fn_t fp, int32_t optr) {
   int32_t num = 0;
-  bool *  pRes = (bool *)pOut->columnData->pData;
+  bool   *pRes = (bool *)pOut->columnData->pData;
 
-  if (GET_PARAM_TYPE(pLeft) == TSDB_DATA_TYPE_JSON || GET_PARAM_TYPE(pRight) == TSDB_DATA_TYPE_JSON) {
+  if (IS_MATHABLE_TYPE(GET_PARAM_TYPE(pLeft)) && IS_MATHABLE_TYPE(GET_PARAM_TYPE(pRight))) {
+    if (!(pLeft->columnData->hasNull || pRight->columnData->hasNull)) {
+      for (int32_t i = startIndex; i < numOfRows && i >= 0; i += step) {
+        int32_t leftIndex = (i >= pLeft->numOfRows) ? 0 : i;
+        int32_t rightIndex = (i >= pRight->numOfRows) ? 0 : i;
+
+        char *pLeftData = colDataGetData(pLeft->columnData, leftIndex);
+        char *pRightData = colDataGetData(pRight->columnData, rightIndex);
+
+        pRes[i] = filterDoCompare(fp, optr, pLeftData, pRightData);
+        if (pRes[i]) {
+          ++num;
+        }
+      }
+    } else {
+      for (int32_t i = startIndex; i < numOfRows && i >= 0; i += step) {
+        int32_t leftIndex = (i >= pLeft->numOfRows) ? 0 : i;
+        int32_t rightIndex = (i >= pRight->numOfRows) ? 0 : i;
+
+        if (colDataIsNull_f(pLeft->columnData->nullbitmap, leftIndex) ||
+            colDataIsNull_f(pRight->columnData->nullbitmap, rightIndex)) {
+          pRes[i] = false;
+          continue;
+        }
+
+        char *pLeftData = colDataGetData(pLeft->columnData, leftIndex);
+        char *pRightData = colDataGetData(pRight->columnData, rightIndex);
+
+        pRes[i] = filterDoCompare(fp, optr, pLeftData, pRightData);
+        if (pRes[i]) {
+          ++num;
+        }
+      }
+    }
+  } else {
+    //  if (GET_PARAM_TYPE(pLeft) == TSDB_DATA_TYPE_JSON || GET_PARAM_TYPE(pRight) == TSDB_DATA_TYPE_JSON) {
     for (int32_t i = startIndex; i < numOfRows && i >= startIndex; i += step) {
       int32_t leftIndex = (i >= pLeft->numOfRows) ? 0 : i;
       int32_t rightIndex = (i >= pRight->numOfRows) ? 0 : i;
@@ -1556,8 +1591,8 @@ int32_t doVectorCompareImpl(SScalarParam *pLeft, SScalarParam *pRight, SScalarPa
         continue;
       }
 
-      char *  pLeftData = colDataGetData(pLeft->columnData, leftIndex);
-      char *  pRightData = colDataGetData(pRight->columnData, rightIndex);
+      char   *pLeftData = colDataGetData(pLeft->columnData, leftIndex);
+      char   *pRightData = colDataGetData(pRight->columnData, rightIndex);
       int64_t leftOut = 0;
       int64_t rightOut = 0;
       bool    freeLeft = false;
@@ -1592,25 +1627,6 @@ int32_t doVectorCompareImpl(SScalarParam *pLeft, SScalarParam *pRight, SScalarPa
         taosMemoryFreeClear(pRightData);
       }
     }
-  } else {
-    for (int32_t i = startIndex; i < numOfRows && i >= 0; i += step) {
-      int32_t leftIndex = (i >= pLeft->numOfRows) ? 0 : i;
-      int32_t rightIndex = (i >= pRight->numOfRows) ? 0 : i;
-
-      if (colDataIsNull_s(pLeft->columnData, leftIndex) ||
-          colDataIsNull_s(pRight->columnData, rightIndex)) {
-        pRes[i] = false;
-        continue;
-      }
-
-      char *pLeftData = colDataGetData(pLeft->columnData, leftIndex);
-      char *pRightData = colDataGetData(pRight->columnData, rightIndex);
-
-      pRes[i] = filterDoCompare(fp, optr, pLeftData, pRightData);
-      if (pRes[i]) {
-        ++num;
-      }
-    }
   }
 
   return num;
@@ -1766,7 +1782,7 @@ void vectorIsTrue(SScalarParam *pLeft, SScalarParam *pRight, SScalarParam *pOut,
     if (colDataIsNull_s(pOut->columnData, i)) {
       int8_t v = 0;
       colDataAppendInt8(pOut->columnData, i, &v);
-      colDataSetNotNull_f(pOut->columnData->nullbitmap, i);
+      colDataClearNull_f(pOut->columnData->nullbitmap, i);
     }
   }
   pOut->columnData->hasNull = false;

source/libs/transport/src/tmsgcb.c

@@ -54,8 +54,6 @@ void tmsgSendRsp(SRpcMsg* pMsg) {
 #endif
 }
 
-void tmsgSendRedirectRsp(SRpcMsg* pMsg, const SEpSet* pNewEpSet) { (*defaultMsgCb.sendRedirectRspFp)(pMsg, pNewEpSet); }
-
 void tmsgRegisterBrokenLinkArg(SRpcMsg* pMsg) { (*defaultMsgCb.registerBrokenLinkArgFp)(pMsg); }
 
 void tmsgReleaseHandle(SRpcHandleInfo* pHandle, int8_t type) { (*defaultMsgCb.releaseHandleFp)(pHandle, type); }

source/libs/transport/src/transCli.c

@@ -1519,8 +1519,9 @@ bool cliGenRetryRule(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
     transFreeMsg(pResp->pCont);
     transUnrefCliHandle(pConn);
   } else if (code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_INTERNAL_ERROR ||
-             code == TSDB_CODE_SYN_PROPOSE_NOT_READY || code == TSDB_CODE_RPC_REDIRECT ||
-             code == TSDB_CODE_VND_STOPPED) {
+             code == TSDB_CODE_SYN_PROPOSE_NOT_READY || code == TSDB_CODE_VND_STOPPED ||
+             code == TSDB_CODE_MNODE_NOT_FOUND || code == TSDB_CODE_APP_IS_STARTING ||
+             code == TSDB_CODE_APP_IS_STOPPING) {
     tTrace("code str %s, contlen:%d 1", tstrerror(code), pResp->contLen);
     noDelay = cliResetEpset(pCtx, pResp, true);
     transFreeMsg(pResp->pCont);

source/util/src/tarray.c

@@ -319,9 +319,9 @@ SArray* taosArrayDup(const SArray* pSrc, __array_item_dup_fn_t fn) {
   } else {
     ASSERT(pSrc->elemSize == sizeof(void*));
 
-    for(int32_t i = 0; i < pSrc->size; ++i) {
+    for (int32_t i = 0; i < pSrc->size; ++i) {
       void* p = fn(taosArrayGetP(pSrc, i));
-      memcpy(((char*)dst->pData )+ i * dst->elemSize, &p, dst->elemSize);
+      memcpy(((char*)dst->pData) + i * dst->elemSize, &p, dst->elemSize);
     }
   }
 

source/util/src/terror.c

@@ -46,8 +46,6 @@ STaosError errors[] = {
 #endif
 
 // rpc
-TAOS_DEFINE_ERROR(TSDB_CODE_RPC_AUTH_FAILURE,             "Authentication failure")
-TAOS_DEFINE_ERROR(TSDB_CODE_RPC_REDIRECT,                 "Database not ready, need retry")
 TAOS_DEFINE_ERROR(TSDB_CODE_RPC_NETWORK_UNAVAIL,          "Unable to establish connection")
 TAOS_DEFINE_ERROR(TSDB_CODE_RPC_FQDN_ERROR,               "Unable to resolve FQDN")
 TAOS_DEFINE_ERROR(TSDB_CODE_RPC_PORT_EADDRINUSE,          "Port already in use")
@@ -57,7 +55,6 @@ TAOS_DEFINE_ERROR(TSDB_CODE_RPC_TIMEOUT,                  "Conn read timeout")
 //common & util
 TAOS_DEFINE_ERROR(TSDB_CODE_TIME_UNSYNCED,                "Client and server's time is not synchronized")
 TAOS_DEFINE_ERROR(TSDB_CODE_OPS_NOT_SUPPORT,              "Operation not supported")
-TAOS_DEFINE_ERROR(TSDB_CODE_MEMORY_CORRUPTED,             "Memory corrupted")
 TAOS_DEFINE_ERROR(TSDB_CODE_OUT_OF_MEMORY,                "Out of Memory")
 TAOS_DEFINE_ERROR(TSDB_CODE_FILE_CORRUPTED,               "Data file corrupted")
 TAOS_DEFINE_ERROR(TSDB_CODE_REF_NO_MEMORY,                "Ref out of memory")
@@ -70,8 +67,6 @@ TAOS_DEFINE_ERROR(TSDB_CODE_REF_NOT_EXIST,                "Ref is not there")
 TAOS_DEFINE_ERROR(TSDB_CODE_APP_ERROR,                    "Unexpected generic error")
 TAOS_DEFINE_ERROR(TSDB_CODE_ACTION_IN_PROGRESS,           "Action in progress")
 TAOS_DEFINE_ERROR(TSDB_CODE_OUT_OF_RANGE,                 "Out of range")
-TAOS_DEFINE_ERROR(TSDB_CODE_OUT_OF_SHM_MEM,               "Out of Shared memory")
-TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_SHM_ID,               "Invalid SHM ID")
 TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_MSG,                  "Invalid message")
 TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_MSG_LEN,              "Invalid message len")
 TAOS_DEFINE_ERROR(TSDB_CODE_INVALID_PTR,                  "Invalid pointer")
@@ -320,7 +315,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_SNODE_ALREADY_DEPLOYED,       "Snode already deploye
 TAOS_DEFINE_ERROR(TSDB_CODE_SNODE_NOT_DEPLOYED,           "Snode not deployed")
 
 // vnode
-TAOS_DEFINE_ERROR(TSDB_CODE_VND_INVALID_VGROUP_ID,        "Invalid Vgroup ID")
+TAOS_DEFINE_ERROR(TSDB_CODE_VND_INVALID_VGROUP_ID,        "Vnode moved to another dnode or was deleted")
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_NO_WRITE_AUTH,            "Database write operation denied")
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_NOT_EXIST,                "Vnode not exist")
 TAOS_DEFINE_ERROR(TSDB_CODE_VND_ALREADY_EXIST,            "Vnode already exist")

source/util/src/thash.c

@@ -244,7 +244,7 @@ SHashObj *taosHashInit(size_t capacity, _hash_fn_t fn, bool update, SHashLockTyp
     capacity = 4;
   }
 
-  SHashObj *pHashObj = (SHashObj *)taosMemoryCalloc(1, sizeof(SHashObj));
+  SHashObj *pHashObj = (SHashObj *)taosMemoryMalloc(sizeof(SHashObj));
   if (pHashObj == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return NULL;
@@ -264,7 +264,7 @@ SHashObj *taosHashInit(size_t capacity, _hash_fn_t fn, bool update, SHashLockTyp
 
   ASSERT((pHashObj->capacity & (pHashObj->capacity - 1)) == 0);
 
-  pHashObj->hashList = (SHashEntry **)taosMemoryCalloc(pHashObj->capacity, sizeof(void *));
+  pHashObj->hashList = (SHashEntry **)taosMemoryMalloc(pHashObj->capacity * sizeof(void *));
   if (pHashObj->hashList == NULL) {
     taosMemoryFree(pHashObj);
     terrno = TSDB_CODE_OUT_OF_MEMORY;
@@ -279,7 +279,7 @@ SHashObj *taosHashInit(size_t capacity, _hash_fn_t fn, bool update, SHashLockTyp
     return NULL;
   }
 
-  void *p = taosMemoryCalloc(pHashObj->capacity, sizeof(SHashEntry));
+  void *p = taosMemoryMalloc(pHashObj->capacity * sizeof(SHashEntry));
   if (p == NULL) {
     taosArrayDestroy(pHashObj->pMemBlock);
     taosMemoryFree(pHashObj->hashList);
@@ -290,6 +290,9 @@ SHashObj *taosHashInit(size_t capacity, _hash_fn_t fn, bool update, SHashLockTyp
 
   for (int32_t i = 0; i < pHashObj->capacity; ++i) {
     pHashObj->hashList[i] = (void *)((char *)p + i * sizeof(SHashEntry));
+    pHashObj->hashList[i]->num = 0;
+    pHashObj->hashList[i]->latch = 0;
+    pHashObj->hashList[i]->next = NULL;
   }
 
   taosArrayPush(pHashObj->pMemBlock, &p);

source/util/src/tpagedbuf.c

@@ -507,7 +507,9 @@ void destroyDiskbasedBuf(SDiskbasedBuf* pBuf) {
 
   dBufPrintStatis(pBuf);
 
+  bool needRemoveFile = false;
   if (pBuf->pFile != NULL) {
+    needRemoveFile = true;
     uDebug(
         "Paged buffer closed, total:%.2f Kb (%d Pages), inmem size:%.2f Kb (%d Pages), file size:%.2f Kb, page "
         "size:%.2f Kb, %s\n",
@@ -534,9 +536,13 @@ void destroyDiskbasedBuf(SDiskbasedBuf* pBuf) {
     }
   }
 
-  if (taosRemoveFile(pBuf->path) < 0) {
-    uDebug("WARNING tPage remove file failed. path=%s", pBuf->path);
+  if (needRemoveFile) {
+    int32_t ret = taosRemoveFile(pBuf->path);
+    if (ret != 0) {  // print the error and discard this error info
+      uDebug("WARNING tPage remove file failed. path=%s, code:%s", pBuf->path, strerror(errno));
     }
+  }
+
   taosMemoryFreeClear(pBuf->path);
 
   size_t n = taosArrayGetSize(pBuf->pIdList);

tests/parallel_test/cases.task

@@ -624,12 +624,12 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/union1.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/concat2.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/json_tag.py
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery.py
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery_str.py
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery_math.py
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery_time.py
-,,,system-test,python3 ./test.py -f 2-query/stablity.py
-,,,system-test,python3 ./test.py -f 2-query/stablity_1.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_str.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_math.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_time.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stablity.py
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stablity_1.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/elapsed.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/csum.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/function_diff.py
@@ -789,12 +789,12 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/arctan.py -Q 2
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/query_cols_tags_and_or.py -Q 2
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/interp.py -Q 2
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery.py -Q 2
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery_str.py -Q 2
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery_math.py -Q 2
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery_time.py -Q 2
-,,,system-test,python3 ./test.py -f 2-query/stablity.py -Q 2
-,,,system-test,python3 ./test.py -f 2-query/stablity_1.py -Q 2
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery.py -Q 2
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_str.py -Q 2
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_math.py -Q 2
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_time.py -Q 2
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stablity.py -Q 2
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stablity_1.py -Q 2
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/avg.py -Q 2
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/elapsed.py -Q 2
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/csum.py -Q 2
@@ -882,12 +882,12 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/arccos.py -Q 3
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/arctan.py -Q 3
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/query_cols_tags_and_or.py -Q 3
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery.py -Q 3
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery_str.py -Q 3
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery_math.py -Q 3
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery_time.py -Q 3
-,,,system-test,python3 ./test.py -f 2-query/stablity.py -Q 3
-,,,system-test,python3 ./test.py -f 2-query/stablity_1.py -Q 3
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery.py -Q 3
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_str.py -Q 3
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_math.py -Q 3
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_time.py -Q 3
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stablity.py -Q 3
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stablity_1.py -Q 3
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/avg.py -Q 3
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/elapsed.py -Q 3
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/csum.py -Q 3
@@ -975,10 +975,10 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/arccos.py -Q 4
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/arctan.py -Q 4
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/query_cols_tags_and_or.py -Q 4
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery.py -Q 4
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery_str.py -Q 4
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery_math.py -Q 4
-,,,system-test,python3 ./test.py -f 2-query/nestedQuery_time.py -Q 4
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery.py -Q 4
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_str.py -Q 4
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_math.py -Q 4
+,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/nestedQuery_time.py -Q 4
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stablity.py -Q 4
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/stablity_1.py -Q 4
 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/avg.py -Q 4

tests/script/tsim/parser/groupby-basic.sim

@@ -78,8 +78,8 @@ $ts2 = $tb2 . .ts
 
 print ===============================groupby_operation
 print
-print ==== select count(*), c1 from group_tb0 group by c1
-sql select count(*), c1 from group_tb0 group by c1
+print ==== select count(*), c1 from group_tb0 group by c1 order by c1
+sql select count(*), c1 from group_tb0 group by c1 order by c1
 print rows: $rows
 print $data00 $data01 $data02 $data03
 print $data10 $data11 $data12 $data13
@@ -98,18 +98,18 @@ endi
 if $data90 != 10 then
   return -1
 endi
-if $data01 != 7 then
+if $data01 != 0 then
   return -1
 endi
-if $data11 != 6 then
+if $data11 != 1 then
   return -1
 endi
-if $data91 != 3 then
+if $data91 != 9 then
   return -1
 endi
 
-print ==== select first(ts),c1 from group_tb0 group by c1;
-sql select first(ts),c1 from group_tb0 group by c1;
+print ==== select first(ts),c1 from group_tb0 group by c1 order by c1;
+sql select first(ts),c1 from group_tb0 group by c1 order by c1;
 print rows: $rows
 print $data00 $data01 $data02 $data03
 print $data10 $data11 $data12 $data13
@@ -120,16 +120,16 @@ if $row != 10 then
   return -1
 endi
 
-if $data00 != @22-01-01 00:00:00.007@ then
+if $data00 != @22-01-01 00:00:00.000@ then
   return -1
 endi
-if $data01 != 7 then
+if $data01 != 0 then
   return -1
 endi
-if $data90 != @22-01-01 00:00:00.003@ then
+if $data90 != @22-01-01 00:00:00.009@ then
   return -1
 endi
-if $data91 != 3 then
+if $data91 != 9 then
   return -1
 endi
 

tests/script/tsim/parser/lastrow_query.sim

@@ -70,10 +70,10 @@ sql select _wstart, t1,t1,count(*),t1,t1 from lr_stb0 where ts>'2018-09-24 00:00
 if $row != 2 then
   return -1
 endi
-if $data01 != NULL then
+if $data01 != 8 then
   return -1
 endi
-if $data02 != NULL then
+if $data02 != 8 then
   return -1
 endi
 if  $data03 != NULL then

tests/script/tsim/valgrind/checkUdf.sim

@@ -121,12 +121,13 @@ if $data01 != 152.420471066 then
   return -1
 endi
 
-sql select udf2(f2) from udf.t2 group by 1-udf1(f1);
+sql select udf2(f2) from udf.t2 group by 1-udf1(f1) order by 1-udf1(f1)
 print $rows , $data00 , $data10
 if $rows != 2 then
   return -1
 endi
 if $data00 != 2.000000000 then
+  print expect 2.000000000 , actual: $data00
   return -1
 endi
 if $data10 != 12.083045974 then

tests/system-test/2-query/unique.py

@@ -429,10 +429,10 @@ class TDTestCase:
         tdSql.checkRows(2)
 
         # nest query
-        tdSql.query(f"select  unique(c1) from (select _rowts , t1 ,c1 , tbname from  {dbname}.stb1 ) ")
+        tdSql.query(f"select  unique(c1) v from (select _rowts , t1 ,c1 , tbname from  {dbname}.stb1 ) order by v")
         tdSql.checkRows(11)
-        tdSql.checkData(0,0,6)
-        tdSql.checkData(10,0,3)
+        tdSql.checkData(1,0,0)
+        tdSql.checkData(10,0,9)
         tdSql.query(f"select  unique(t1) from (select _rowts , t1 , tbname from  {dbname}.stb1 )")
         tdSql.checkRows(2)
         tdSql.checkData(0,0,4)