Merge branch '3.0' into feature/tq

Jenkinsfile2

@@ -121,7 +121,7 @@ def pre_test_win(){
         set
         date /t
         time /t
-        rd /s /Q C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\debug
+        rd /s /Q C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\debug || exit 0
     '''
     bat '''
         cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal

include/libs/nodes/querynodes.h

@@ -236,6 +236,7 @@ typedef struct SSelectStmt {
   bool        isTimeOrderQuery;
   bool        hasAggFuncs;
   bool        hasRepeatScanFuncs;
+  bool        hasNonstdSQLFunc;
 } SSelectStmt;
 
 typedef enum ESetOperatorType { SET_OP_TYPE_UNION_ALL = 1, SET_OP_TYPE_UNION } ESetOperatorType;

include/libs/sync/sync.h

@@ -20,14 +20,11 @@
 extern "C" {
 #endif
 
-#include <stdbool.h>
-#include <stdint.h>
-//#include <tdatablock.h>
+#include "os.h"
+
 #include "cJSON.h"
 #include "tdef.h"
-//#include "taosdef.h"
-//#include "trpc.h"
-//#include "wal.h"
+#include "tmsgcb.h"
 
 typedef uint64_t SyncNodeId;
 typedef int32_t  SyncGroupId;
@@ -132,11 +129,10 @@ typedef struct SSyncInfo {
   char        path[TSDB_FILENAME_LEN];
   SWal*       pWal;
   SSyncFSM*   pFsm;
+  SMsgCb*     msgcb;
 
-  void* rpcClient;
-  int32_t (*FpSendMsg)(void* rpcClient, const SEpSet* pEpSet, SRpcMsg* pMsg);
-  void* queue;
-  int32_t (*FpEqMsg)(void* queue, SRpcMsg* pMsg);
+  int32_t (*FpSendMsg)(const SEpSet* pEpSet, SRpcMsg* pMsg);
+  int32_t (*FpEqMsg)(const SMsgCb* msgcb, SRpcMsg* pMsg);
 
 } SSyncInfo;
 

include/libs/sync/syncTools.h

@@ -20,13 +20,10 @@
 extern "C" {
 #endif
 
-#include <stdbool.h>
-#include <stdint.h>
-//#include <tdatablock.h>
+#include "os.h"
+
 #include "cJSON.h"
-//#include "taosdef.h"
 #include "trpc.h"
-//#include "wal.h"
 
 // ------------------ ds -------------------
 typedef struct SRaftId {
@@ -43,8 +40,7 @@ void       syncNodeRelease(SSyncNode* pNode);
 
 int32_t syncGetRespRpc(int64_t rid, uint64_t index, SRpcMsg* msg);
 int32_t syncGetAndDelRespRpc(int64_t rid, uint64_t index, SRpcMsg* msg);
-void    syncSetQ(int64_t rid, void* queueHandle);
-void    syncSetRpc(int64_t rid, void* rpcHandle);
+void    syncSetMsgCb(int64_t rid, const SMsgCb* msgcb);
 char*   sync2SimpleStr(int64_t rid);
 
 // set timer ms

include/libs/transport/trpc.h

@@ -38,7 +38,7 @@ typedef struct {
 
 typedef struct SRpcHandleInfo {
   // rpc info
-  void   *handle;         // rpc handle returned to app
+  void *  handle;         // rpc handle returned to app
   int64_t refId;          // refid, used by server
   int32_t noResp;         // has response or not(default 0, 0: resp, 1: no resp);
   int32_t persistHandle;  // persist handle or not
@@ -49,13 +49,13 @@ typedef struct SRpcHandleInfo {
   void *node;     // node mgmt handle
 
   // resp info
-  void   *rsp;
+  void *  rsp;
   int32_t rspLen;
 } SRpcHandleInfo;
 
 typedef struct SRpcMsg {
   tmsg_t         msgType;
-  void          *pCont;
+  void *         pCont;
   int32_t        contLen;
   int32_t        code;
   SRpcHandleInfo info;
@@ -63,11 +63,6 @@ typedef struct SRpcMsg {
 } SRpcMsg;
 
 typedef void (*RpcCfp)(void *parent, SRpcMsg *, SEpSet *rf);
-typedef int (*RpcAfp)(void *parent, char *tableId, char *spi, char *encrypt, char *secret, char *ckey);
-///
-// // SRpcMsg code
-// REDIERE,
-// NOT READY, EpSet
 typedef bool (*RpcRfp)(int32_t code);
 
 typedef struct SRpcInit {
@@ -81,17 +76,10 @@ typedef struct SRpcInit {
 
   // the following is for client app ecurity only
   char *user;  // user name
-  char  spi;      // security parameter index
-  char  encrypt;  // encrypt algorithm
-  char *secret;   // key for authentication
-  char *ckey;     // ciphering key
 
   // call back to process incoming msg, code shall be ignored by server app
   RpcCfp cfp;
 
-  // call back to retrieve the client auth info, for server app only
-  RpcAfp afp;
-
   // user defined retry func
   RpcRfp rfp;
 

include/util/taoserror.h

@@ -649,6 +649,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_INVALID_TBNAME            TAOS_DEF_ERROR_CODE(0, 0x264C)
 #define TSDB_CODE_PAR_INVALID_FUNCTION_NAME     TAOS_DEF_ERROR_CODE(0, 0x264D)
 #define TSDB_CODE_PAR_COMMENT_TOO_LONG          TAOS_DEF_ERROR_CODE(0, 0x264E)
+#define TSDB_CODE_PAR_NOT_ALLOWED_FUNC          TAOS_DEF_ERROR_CODE(0, 0x264F)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)

packaging/deb/makedeb.sh

@@ -67,9 +67,9 @@ fi
 
 cp ${compile_dir}/build/bin/taos                    ${pkg_dir}${install_home_path}/bin
 cp ${compile_dir}/build/lib/${libfile}              ${pkg_dir}${install_home_path}/driver
-cp ${compile_dir}/../src/inc/taos.h                 ${pkg_dir}${install_home_path}/include
-cp ${compile_dir}/../src/inc/taosdef.h              ${pkg_dir}${install_home_path}/include
-cp ${compile_dir}/../src/inc/taoserror.h            ${pkg_dir}${install_home_path}/include
+cp ${compile_dir}/../include/client/taos.h          ${pkg_dir}${install_home_path}/include
+cp ${compile_dir}/../include/common/taosdef.h       ${pkg_dir}${install_home_path}/include
+cp ${compile_dir}/../include/util/taoserror.h       ${pkg_dir}${install_home_path}/include
 cp -r ${top_dir}/examples/*                         ${pkg_dir}${install_home_path}/examples
 #cp -r ${top_dir}/src/connector/python               ${pkg_dir}${install_home_path}/connector
 #cp -r ${top_dir}/src/connector/go                   ${pkg_dir}${install_home_path}/connector

packaging/release.sh

 #!/bin/bash
 #
-# Generate the tar.gz package for linux os
+# Generate the deb package for ubuntu, or rpm package for centos, or tar.gz package for other linux os
 
 set -e
 #set -x
 
+# release.sh  -v [cluster | edge]
+#             -c [aarch32 | aarch64 | x64 | x86 | mips64 ...]
+#             -o [Linux | Kylin | Alpine | Raspberrypi | Darwin | Windows | Ningsi60 | Ningsi80 |...]
+#             -V [stable | beta]
+#             -l [full | lite]
+#             -s [static | dynamic]
+#             -d [taos | ...]
+#             -n [2.0.0.3]
+#             -m [2.0.0.0]
+#             -H [ false | true]
+
 # set parameters by default value
-version="3.0.0.0"
+verMode=edge    # [cluster, edge]
+verType=stable  # [stable, beta]
+cpuType=x64     # [aarch32 | aarch64 | x64 | x86 | mips64 ...]
+osType=Linux    # [Linux | Kylin | Alpine | Raspberrypi | Darwin | Windows | Ningsi60 | Ningsi80 |...]
+pagMode=full    # [full | lite]
+soMode=dynamic  # [static | dynamic]
+dbName=taos     # [taos | ...]
+allocator=glibc # [glibc | jemalloc]
+verNumber=""
+verNumberComp="2.0.0.0"
+httpdBuild=false
+
+while getopts "hv:V:c:o:l:s:d:a:n:m:H:" arg; do
+  case $arg in
+  v)
+    #echo "verMode=$OPTARG"
+    verMode=$(echo $OPTARG)
+    ;;
+  V)
+    #echo "verType=$OPTARG"
+    verType=$(echo $OPTARG)
+    ;;
+  c)
+    #echo "cpuType=$OPTARG"
+    cpuType=$(echo $OPTARG)
+    ;;
+  l)
+    #echo "pagMode=$OPTARG"
+    pagMode=$(echo $OPTARG)
+    ;;
+  s)
+    #echo "soMode=$OPTARG"
+    soMode=$(echo $OPTARG)
+    ;;
+  d)
+    #echo "dbName=$OPTARG"
+    dbName=$(echo $OPTARG)
+    ;;
+  a)
+    #echo "allocator=$OPTARG"
+    allocator=$(echo $OPTARG)
+    ;;
+  n)
+    #echo "verNumber=$OPTARG"
+    verNumber=$(echo $OPTARG)
+    ;;
+  m)
+    #echo "verNumberComp=$OPTARG"
+    verNumberComp=$(echo $OPTARG)
+    ;;
+  o)
+    #echo "osType=$OPTARG"
+    osType=$(echo $OPTARG)
+    ;;
+  H)
+    #echo "httpdBuild=$OPTARG"
+    httpdBuild=$(echo $OPTARG)
+    ;;
+  h)
+    echo "Usage: $(basename $0) -v [cluster | edge] "
+    echo "                  -c [aarch32 | aarch64 | x64 | x86 | mips64 ...] "
+    echo "                  -o [Linux | Kylin | Alpine | Raspberrypi | Darwin | Windows | Ningsi60 | Ningsi80 |...] "
+    echo "                  -V [stable | beta] "
+    echo "                  -l [full | lite] "
+    echo "                  -a [glibc | jemalloc] "
+    echo "                  -s [static | dynamic] "
+    echo "                  -d [taos | ...] "
+    echo "                  -n [version number] "
+    echo "                  -m [compatible version number] "
+    echo "                  -H [false | true] "
+    exit 0
+    ;;
+  ?) #unknow option
+    echo "unkonw argument"
+    exit 1
+    ;;
+  esac
+done
+
+echo "verMode=${verMode} verType=${verType} cpuType=${cpuType} osType=${osType} pagMode=${pagMode} soMode=${soMode} dbName=${dbName} allocator=${allocator} verNumber=${verNumber} verNumberComp=${verNumberComp} httpdBuild=${httpdBuild}"
 
 curr_dir=$(pwd)
 
-script_dir="$(dirname $(readlink -f $0))"
-top_dir="$(readlink -f ${script_dir}/..)"
+if [ "$osType" == "Darwin" ]; then
+  script_dir=$(dirname $0)
+  cd ${script_dir}
+  script_dir="$(pwd)"
+  top_dir=${script_dir}/..
+else
+  script_dir="$(dirname $(readlink -f $0))"
+  top_dir="$(readlink -f ${script_dir}/..)"
+fi
+
+csudo=""
+if command -v sudo > /dev/null; then
+ csudo="sudo "
+fi
+
+function is_valid_version() {
+  [ -z $1 ] && return 1 || :
+
+  rx='^([0-9]+\.){3}(\*|[0-9]+)$'
+  if [[ $1 =~ $rx ]]; then
+    return 0
+  fi
+  return 1
+}
+
+function vercomp() {
+  if [[ $1 == $2 ]]; then
+    echo 0
+    exit 0
+  fi
+
+  local IFS=.
+  local i ver1=($1) ver2=($2)
+
+  # fill empty fields in ver1 with zeros
+  for ((i = ${#ver1[@]}; i < ${#ver2[@]}; i++)); do
+    ver1[i]=0
+  done
+
+  for ((i = 0; i < ${#ver1[@]}; i++)); do
+    if [[ -z ${ver2[i]} ]]; then
+      # fill empty fields in ver2 with zeros
+      ver2[i]=0
+    fi
+    if ((10#${ver1[i]} > 10#${ver2[i]})); then
+      echo 1
+      exit 0
+    fi
+    if ((10#${ver1[i]} < 10#${ver2[i]})); then
+      echo 2
+      exit 0
+    fi
+  done
+  echo 0
+}
+
+# 1. check version information
+if ( (! is_valid_version $verNumber) || (! is_valid_version $verNumberComp) || [[ "$(vercomp $verNumber $verNumberComp)" == '2' ]]); then
+  echo "please enter correct version"
+  exit 0
+fi
 
-echo "=======================new version number: ${verNumber}======================================"
+echo "=======================new version number: ${verNumber}, compatible version: ${verNumberComp}======================================"
 
 build_time=$(date +"%F %R")
 
-echo "script_dir: ${script_dir}"
-echo "top_dir: ${top_dir}"
+# get commint id from git
+gitinfo=$(git rev-parse --verify HEAD)
 
-cd ${top_dir}
-# git checkout -- .
-# git checkout 3.0
-# git pull || :
+if [[ "$verMode" == "cluster" ]]; then
+  enterprise_dir="${top_dir}/../enterprise"
+  cd ${enterprise_dir}
+  gitinfoOfInternal=$(git rev-parse --verify HEAD)
+else
+  gitinfoOfInternal=NULL
+fi
 
-echo "curr_dir: ${curr_dir}"
+cd "${curr_dir}"
 
 # 2. cmake executable file
 compile_dir="${top_dir}/debug"
-# if [ -d ${compile_dir} ]; then
-#   rm -rf ${compile_dir}
-# fi
-
-mkdir -p ${compile_dir}
+if [ -d ${compile_dir} ]; then
+  ${csudo}rm -rf ${compile_dir}
+fi
 
+if [ "$osType" != "Darwin" ]; then
+  ${csudo}mkdir -p ${compile_dir}
+else
+  mkdir -p ${compile_dir}
+fi
 cd ${compile_dir}
 
-echo "compile_dir: ${compile_dir}"
+if [[ "$allocator" == "jemalloc" ]]; then
+  allocator_macro="-DJEMALLOC_ENABLED=true"
+else
+  allocator_macro=""
+fi
 
-cmake .. -DBUILD_TOOLS=true
-make -j32
+if [[ "$dbName" != "taos" ]]; then
+  source ${enterprise_dir}/packaging/oem/sed_$dbName.sh
+  replace_community_$dbName
+fi
 
-release_dir="${top_dir}/release"
-if [ -d ${release_dir} ]; then
-  rm -rf ${release_dir}
+if [[ "$httpdBuild" == "true" ]]; then
+  BUILD_HTTP=true
+else
+  BUILD_HTTP=false
 fi
 
-mkdir -p ${release_dir}
-cd ${release_dir}
+if [[ "$verMode" == "cluster" ]]; then
+  BUILD_HTTP=internal
+fi
 
-install_dir="${release_dir}/TDengine-server-${version}"
-mkdir -p ${install_dir}
-mkdir -p ${install_dir}/bin
-mkdir -p ${install_dir}/lib
-mkdir -p ${install_dir}/inc
+if [[ "$pagMode" == "full" ]]; then
+  BUILD_TOOLS=true
+else
+  BUILD_TOOLS=false
+fi
+
+# check support cpu type
+if [[ "$cpuType" == "x64" ]] || [[ "$cpuType" == "aarch64" ]] || [[ "$cpuType" == "aarch32" ]] || [[ "$cpuType" == "mips64" ]]; then
+  if [ "$verMode" != "cluster" ]; then
+    # community-version compile
+    cmake ../ -DCPUTYPE=${cpuType} -DOSTYPE=${osType} -DSOMODE=${soMode} -DDBNAME=${dbName} -DVERTYPE=${verType} -DVERDATE="${build_time}" -DGITINFO=${gitinfo} -DGITINFOI=${gitinfoOfInternal} -DVERNUMBER=${verNumber} -DVERCOMPATIBLE=${verNumberComp} -DPAGMODE=${pagMode} -DBUILD_HTTP=${BUILD_HTTP} -DBUILD_TOOLS=${BUILD_TOOLS} ${allocator_macro}
+  else
+    if [[ "$dbName" != "taos" ]]; then
+      replace_enterprise_$dbName
+    fi
+    cmake ../../ -DCPUTYPE=${cpuType} -DOSTYPE=${osType} -DSOMODE=${soMode} -DDBNAME=${dbName} -DVERTYPE=${verType} -DVERDATE="${build_time}" -DGITINFO=${gitinfo} -DGITINFOI=${gitinfoOfInternal} -DVERNUMBER=${verNumber} -DVERCOMPATIBLE=${verNumberComp} -DBUILD_HTTP=${BUILD_HTTP} -DBUILD_TOOLS=${BUILD_TOOLS} ${allocator_macro}
+  fi
+else
+  echo "input cpuType=${cpuType} error!!!"
+  exit 1
+fi
 
-install_files="${script_dir}/tools/install.sh"
-chmod a+x ${script_dir}/tools/install.sh || :
-cp ${install_files} ${install_dir} 
+CORES=$(grep -c ^processor /proc/cpuinfo)
 
-header_files="${top_dir}/include/client/taos.h ${top_dir}/include/util/taoserror.h"
-cp ${header_files} ${install_dir}/inc
+if [[ "$allocator" == "jemalloc" ]]; then
+  # jemalloc need compile first, so disable parallel build
+  make -j ${CORES} && ${csudo}make install
+else
+  make -j ${CORES} && ${csudo}make install
+fi
+
+cd ${curr_dir}
 
-bin_files="${compile_dir}/build/bin/taosd ${compile_dir}/build/bin/taos  ${compile_dir}/build/bin/create_table ${compile_dir}/build/bin/tmq_sim ${script_dir}/tools/remove.sh ${compile_dir}/build/bin/taosBenchmark   ${compile_dir}/build/bin/taosdump"
-cp -rf ${bin_files} ${install_dir}/bin && chmod a+x ${install_dir}/bin/* || :
+# 3. Call the corresponding script for packaging
+if [ "$osType" != "Darwin" ]; then
+  if [[ "$verMode" != "cluster" ]] && [[ "$pagMode" == "full" ]] && [[ "$cpuType" == "x64" ]] && [[ "$dbName" == "taos" ]]; then
+    ret='0'
+    command -v dpkg >/dev/null 2>&1 || { ret='1'; }
+    if [ "$ret" -eq 0 ]; then
+      echo "====do deb package for the ubuntu system===="
+      output_dir="${top_dir}/debs"
+      if [ -d ${output_dir} ]; then
+        ${csudo}rm -rf ${output_dir}
+      fi
+      ${csudo}mkdir -p ${output_dir}
+      cd ${script_dir}/deb
+      ${csudo}./makedeb.sh ${compile_dir} ${output_dir} ${verNumber} ${cpuType} ${osType} ${verMode} ${verType}
 
-cp ${compile_dir}/build/lib/libtaos.so  ${install_dir}/lib/
-cp ${compile_dir}/build/lib/libavro* ${install_dir}/lib/ > /dev/null || echo -e "failed to copy avro libraries"
-cp -rf ${compile_dir}/build/lib/pkgconfig  ${install_dir}/lib/ > /dev/null || echo -e "failed to copy pkgconfig directory"
+      if [[ "$pagMode" == "full" ]]; then
+        if [ -d ${top_dir}/tools/taos-tools/packaging/deb ]; then
+          cd ${top_dir}/tools/taos-tools/packaging/deb
+          [ -z "$taos_tools_ver" ] && taos_tools_ver="0.1.0"
 
+          taos_tools_ver=$(git describe --tags | sed -e 's/ver-//g' | awk -F '-' '{print $1}')
+          ${csudo}./make-taos-tools-deb.sh ${top_dir} \
+            ${compile_dir} ${output_dir} ${taos_tools_ver} ${cpuType} ${osType} ${verMode} ${verType}
+        fi
+      fi
+    else
+      echo "==========dpkg command not exist, so not release deb package!!!"
+    fi
+    ret='0'
+    command -v rpmbuild >/dev/null 2>&1 || { ret='1'; }
+    if [ "$ret" -eq 0 ]; then
+      echo "====do rpm package for the centos system===="
+      output_dir="${top_dir}/rpms"
+      if [ -d ${output_dir} ]; then
+        ${csudo}rm -rf ${output_dir}
+      fi
+      ${csudo}mkdir -p ${output_dir}
+      cd ${script_dir}/rpm
+      ${csudo}./makerpm.sh ${compile_dir} ${output_dir} ${verNumber} ${cpuType} ${osType} ${verMode} ${verType}
 
-#cp ${compile_dir}/source/dnode/mnode/impl/libmnode.so ${install_dir}/lib/
-#cp ${compile_dir}/source/dnode/qnode/libqnode.so ${install_dir}/lib/
-#cp ${compile_dir}/source/dnode/snode/libsnode.so ${install_dir}/lib/
-#cp ${compile_dir}/source/dnode/bnode/libbnode.so ${install_dir}/lib/
-#cp ${compile_dir}/source/libs/wal/libwal.so ${install_dir}/lib/
-#cp ${compile_dir}/source/libs/scheduler/libscheduler.so ${install_dir}/lib/
-#cp ${compile_dir}/source/libs/planner/libplanner.so ${install_dir}/lib/
-#cp ${compile_dir}/source/libs/parser/libparser.so ${install_dir}/lib/
-#cp ${compile_dir}/source/libs/qcom/libqcom.so ${install_dir}/lib/
-#cp ${compile_dir}/source/libs/transport/libtransport.so ${install_dir}/lib/
-#cp ${compile_dir}/source/libs/function/libfunction.so ${install_dir}/lib/
-#cp ${compile_dir}/source/common/libcommon.so ${install_dir}/lib/
-#cp ${compile_dir}/source/os/libos.so ${install_dir}/lib/
-#cp ${compile_dir}/source/dnode/mnode/sdb/libsdb.so ${install_dir}/lib/
-#cp ${compile_dir}/source/libs/catalog/libcatalog.so ${install_dir}/lib/
+      if [[ "$pagMode" == "full" ]]; then
+        if [ -d ${top_dir}/tools/taos-tools/packaging/rpm ]; then
+          cd ${top_dir}/tools/taos-tools/packaging/rpm
+          [ -z "$taos_tools_ver" ] && taos_tools_ver="0.1.0"
 
-pkg_name=${install_dir}-Linux-x64
+          taos_tools_ver=$(git describe --tags | sed -e 's/ver-//g' | awk -F '-' '{print $1}' | sed -e 's/-/_/g')
+          ${csudo}./make-taos-tools-rpm.sh ${top_dir} \
+            ${compile_dir} ${output_dir} ${taos_tools_ver} ${cpuType} ${osType} ${verMode} ${verType}
+        fi
+      fi
+    else
+      echo "==========rpmbuild command not exist, so not release rpm package!!!"
+    fi
+  fi
 
-tar -zcv -f "$(basename ${pkg_name}).tar.gz" $(basename ${install_dir}) --remove-files || :
+  echo "====do tar.gz package for all systems===="
+  cd ${script_dir}/tools
 
+  ${csudo}./makepkg.sh ${compile_dir} ${verNumber} "${build_time}" ${cpuType} ${osType} ${verMode} ${verType} ${pagMode} ${verNumberComp} ${dbName}
+  ${csudo}./makeclient.sh ${compile_dir} ${verNumber} "${build_time}" ${cpuType} ${osType} ${verMode} ${verType} ${pagMode} ${dbName}
+  # ${csudo}./makearbi.sh ${compile_dir} ${verNumber} "${build_time}" ${cpuType} ${osType} ${verMode} ${verType} ${pagMode}
 
+else
+  # only make client for Darwin
+  cd ${script_dir}/tools
+  ./makeclient.sh ${compile_dir} ${verNumber} "${build_time}" ${cpuType} ${osType} ${verMode} ${verType} ${pagMode} ${dbName}
+fi

packaging/rpm/tdengine.spec

@@ -74,9 +74,9 @@ if [ -f %{_compiledir}/build/bin/taosadapter ]; then
     cp %{_compiledir}/build/bin/taosadapter                    %{buildroot}%{homepath}/bin ||:
 fi
 cp %{_compiledir}/build/lib/${libfile}              %{buildroot}%{homepath}/driver
-cp %{_compiledir}/../src/inc/taos.h                 %{buildroot}%{homepath}/include
-cp %{_compiledir}/../src/inc/taosdef.h              %{buildroot}%{homepath}/include
-cp %{_compiledir}/../src/inc/taoserror.h            %{buildroot}%{homepath}/include
+cp %{_compiledir}/../include/client/taos.h          %{buildroot}%{homepath}/include
+cp %{_compiledir}/../include/common/taosdef.h       %{buildroot}%{homepath}/include
+cp %{_compiledir}/../include/util/taoserror.h       %{buildroot}%{homepath}/include
 #cp -r %{_compiledir}/../src/connector/python        %{buildroot}%{homepath}/connector
 #cp -r %{_compiledir}/../src/connector/go            %{buildroot}%{homepath}/connector
 #cp -r %{_compiledir}/../src/connector/nodejs        %{buildroot}%{homepath}/connector

packaging/tools/install.sh

@@ -485,6 +485,17 @@ function install_service() {
     # fi
 }
 
+function install_config() {
+    if [ ! -f ${cfg_install_dir}/${configFile} ]; then
+        ${csudo}mkdir -p ${cfg_install_dir}
+        [ -f ${script_dir}/cfg/${configFile} ] && ${csudo}cp ${script_dir}/cfg/${configFile} ${cfg_install_dir}
+        ${csudo}chmod 644 ${cfg_install_dir}/*
+    fi
+
+    ${csudo}cp -f ${script_dir}/cfg/${configFile} ${install_main_dir}/cfg/${configFile}.org
+    ${csudo}ln -s ${cfg_install_dir}/${configFile} ${install_main_dir}/cfg
+}
+
 function install_TDengine() {
     # Start to install
     echo -e "${GREEN}Start to install TDengine...${NC}"
@@ -500,7 +511,7 @@ function install_TDengine() {
         # For installing new
         install_bin
         install_service
-        #install_config
+        install_config
 
         # Ask if to start the service
         #echo
@@ -539,7 +550,7 @@ function install_TDengine() {
         echo
     else # Only install client
         install_bin
-        #install_config
+        install_config
         echo
         echo -e "\033[44;32;1mTDengine client is installed successfully!${NC}"
     fi

packaging/tools/install_client.sh

@@ -17,7 +17,6 @@ serverName="taosd"
 clientName="taos"
 uninstallScript="rmtaos"
 configFile="taos.cfg"
-tarName="taos.tar.gz"
 
 osType=Linux
 pagMode=full
@@ -243,12 +242,6 @@ function install_examples() {
 
 function update_TDengine() {
     # Start to update
-    if [ ! -e ${tarName} ]; then
-        echo "File ${tarName} does not exist"
-        exit 1
-    fi
-    tar -zxf ${tarName}
-
     echo -e "${GREEN}Start to update ${productName} client...${NC}"
     # Stop the client shell if running
     if pidof ${clientName} &> /dev/null; then
@@ -271,18 +264,10 @@ function update_TDengine() {
 
     echo
     echo -e "\033[44;32;1m${productName} client is updated successfully!${NC}"
-
-    rm -rf $(tar -tf ${tarName})
 }
 
 function install_TDengine() {
   # Start to install
-  if [ ! -e ${tarName} ]; then
-    echo "File ${tarName} does not exist"
-    exit 1
-  fi
-  tar -zxf ${tarName}
-
   echo -e "${GREEN}Start to install ${productName} client...${NC}"
 
   install_main_path

packaging/tools/makearbi.sh

@@ -22,7 +22,7 @@ productName="TDengine"
 
 # create compressed install file.
 build_dir="${compile_dir}/build"
-code_dir="${top_dir}/src"
+code_dir="${top_dir}"
 release_dir="${top_dir}/release"
 
 #package_name='linux'
@@ -36,7 +36,7 @@ fi
 bin_files="${build_dir}/bin/tarbitrator ${script_dir}/remove_arbi.sh"
 install_files="${script_dir}/install_arbi.sh"
 
-#header_files="${code_dir}/inc/taos.h ${code_dir}/inc/taosdef.h ${code_dir}/inc/taoserror.h"
+#header_files="${code_dir}/include/client/taos.h ${code_dir}/include/common/taosdef.h ${code_dir}/include/util/taoserror.h"
 init_file_tarbitrator_deb=${script_dir}/../deb/tarbitratord
 init_file_tarbitrator_rpm=${script_dir}/../rpm/tarbitratord
 

packaging/tools/makeclient.sh

@@ -32,7 +32,7 @@ fi
 
 # create compressed install file.
 build_dir="${compile_dir}/build"
-code_dir="${top_dir}/src"
+code_dir="${top_dir}"
 release_dir="${top_dir}/release"
 
 #package_name='linux'
@@ -62,7 +62,7 @@ else
   lib_files="${build_dir}/lib/libtaos.${version}.dylib"
 fi
 
-header_files="${code_dir}/inc/taos.h ${code_dir}/inc/taosdef.h ${code_dir}/inc/taoserror.h"
+header_files="${code_dir}/include/client/taos.h ${code_dir}/include/common/taosdef.h ${code_dir}/include/util/taoserror.h"
 if [ "$dbName" != "taos" ]; then
   cfg_dir="${top_dir}/../enterprise/packaging/cfg"
 else
@@ -152,7 +152,7 @@ if [[ $productName == "TDengine" ]]; then
     cp -r ${examples_dir}/go ${install_dir}/examples
     cp -r ${examples_dir}/nodejs ${install_dir}/examples
     cp -r ${examples_dir}/C# ${install_dir}/examples
-    mkdir -p ${install_dir}/examples/taosbenchmark-json && cp ${examples_dir}/../src/kit/taos-tools/example/* ${install_dir}/examples/taosbenchmark-json
+    mkdir -p ${install_dir}/examples/taosbenchmark-json && cp ${examples_dir}/../tools/taos-tools/example/* ${install_dir}/examples/taosbenchmark-json
   fi
 
   if [ "$verMode" == "cluster" ]; then
@@ -199,8 +199,8 @@ if [[ "$pagMode" != "lite" ]] && [[ "$cpuType" != "aarch32" ]]; then
   else
     echo "WARNING: go connector not found, please check if want to use it!"
   fi
-  cp -r ${connector_dir}/python ${install_dir}/connector
-  cp -r ${connector_dir}/nodejs ${install_dir}/connector
+  cp -r ${connector_dir}/python ${install_dir}/connector || :
+  cp -r ${connector_dir}/nodejs ${install_dir}/connector || :
 fi
 # Copy release note
 # cp ${script_dir}/release_note ${install_dir}

packaging/tools/makepkg.sh

@@ -33,7 +33,7 @@ defaultPasswd="taosdata"
 
 # create compressed install file.
 build_dir="${compile_dir}/build"
-code_dir="${top_dir}/src"
+code_dir="${top_dir}"
 release_dir="${top_dir}/release"
 
 #package_name='linux'
@@ -43,8 +43,8 @@ else
   install_dir="${release_dir}/${productName}-server-${version}"
 fi
 
-if [ -d ${top_dir}/src/kit/taos-tools/packaging/deb ]; then
-  cd ${top_dir}/src/kit/taos-tools/packaging/deb
+if [ -d ${top_dir}/tools/taos-tools/packaging/deb ]; then
+  cd ${top_dir}/tools/taos-tools/packaging/deb
   [ -z "$taos_tools_ver" ] && taos_tools_ver="0.1.0"
 
   taostools_ver=$(git describe --tags | sed -e 's/ver-//g' | awk -F '-' '{print $1}')
@@ -94,7 +94,7 @@ else
 fi
 
 lib_files="${build_dir}/lib/libtaos.so.${version}"
-header_files="${code_dir}/inc/taos.h ${code_dir}/inc/taosdef.h ${code_dir}/inc/taoserror.h"
+header_files="${code_dir}/include/client/taos.h ${code_dir}/include/common/taosdef.h ${code_dir}/include/util/taoserror.h"
 
 if [ "$dbName" != "taos" ]; then
   cfg_dir="${top_dir}/../enterprise/packaging/cfg"
@@ -103,7 +103,7 @@ else
 fi
 
 install_files="${script_dir}/install.sh"
-nginx_dir="${code_dir}/../../enterprise/src/plugins/web"
+nginx_dir="${top_dir}/../enterprise/src/plugins/web"
 
 init_file_deb=${script_dir}/../deb/taosd
 init_file_rpm=${script_dir}/../rpm/taosd
@@ -162,8 +162,8 @@ if [ -n "${taostools_bin_files}" ]; then
         && cp ${taostools_bin_files} ${taostools_install_dir}/bin \
         && chmod a+x ${taostools_install_dir}/bin/* || :
 
-    if [ -f ${top_dir}/src/kit/taos-tools/packaging/tools/install-taostools.sh ]; then
-        cp ${top_dir}/src/kit/taos-tools/packaging/tools/install-taostools.sh \
+    if [ -f ${top_dir}/tools/taos-tools/packaging/tools/install-taostools.sh ]; then
+        cp ${top_dir}/tools/taos-tools/packaging/tools/install-taostools.sh \
             ${taostools_install_dir}/ > /dev/null \
             && chmod a+x ${taostools_install_dir}/install-taostools.sh \
             || echo -e "failed to copy install-taostools.sh"
@@ -171,8 +171,8 @@ if [ -n "${taostools_bin_files}" ]; then
         echo -e "install-taostools.sh not found"
     fi
 
-    if [ -f ${top_dir}/src/kit/taos-tools/packaging/tools/uninstall-taostools.sh ]; then
-        cp ${top_dir}/src/kit/taos-tools/packaging/tools/uninstall-taostools.sh \
+    if [ -f ${top_dir}/tools/taos-tools/packaging/tools/uninstall-taostools.sh ]; then
+        cp ${top_dir}/tools/taos-tools/packaging/tools/uninstall-taostools.sh \
             ${taostools_install_dir}/ > /dev/null \
             && chmod a+x ${taostools_install_dir}/uninstall-taostools.sh \
             || echo -e "failed to copy uninstall-taostools.sh"
@@ -288,7 +288,7 @@ if [[ $dbName == "taos" ]]; then
     cp -r ${examples_dir}/go ${install_dir}/examples
     cp -r ${examples_dir}/nodejs ${install_dir}/examples
     cp -r ${examples_dir}/C# ${install_dir}/examples
-    mkdir -p ${install_dir}/examples/taosbenchmark-json && cp ${examples_dir}/../src/kit/taos-tools/example/* ${install_dir}/examples/taosbenchmark-json
+    mkdir -p ${install_dir}/examples/taosbenchmark-json && cp ${examples_dir}/../tools/taos-tools/example/* ${install_dir}/examples/taosbenchmark-json
   fi
 fi
 

source/client/src/clientEnv.c

@@ -60,7 +60,7 @@ static void registerRequest(SRequestObj *pRequest) {
 static void deregisterRequest(SRequestObj *pRequest) {
   assert(pRequest != NULL);
 
-  STscObj          *pTscObj = pRequest->pTscObj;
+  STscObj *         pTscObj = pRequest->pTscObj;
   SInstanceSummary *pActivity = &pTscObj->pAppInfo->summary;
 
   int32_t currentInst = atomic_sub_fetch_64((int64_t *)&pActivity->currentRequests, 1);
@@ -91,7 +91,6 @@ static bool clientRpcRfp(int32_t code) {
   }
 }
 
-
 // TODO refactor
 void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
   SRpcInit rpcInit;
@@ -105,10 +104,6 @@ void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
   rpcInit.connType = TAOS_CONN_CLIENT;
   rpcInit.user = (char *)user;
   rpcInit.idleTime = tsShellActivityTimer * 1000;
-  rpcInit.ckey = "key";
-  rpcInit.spi = 1;
-  rpcInit.secret = (char *)auth;
-
   void *pDnodeConn = rpcOpen(&rpcInit);
   if (pDnodeConn == NULL) {
     tscError("failed to init connection to server");
@@ -318,7 +313,7 @@ int taos_options_imp(TSDB_OPTION option, const char *str) {
     return 0;
   }
 
-  SConfig     *pCfg = taosGetCfg();
+  SConfig *    pCfg = taosGetCfg();
   SConfigItem *pItem = NULL;
 
   switch (option) {

source/client/src/clientImpl.c

@@ -339,12 +339,11 @@ int32_t validateSversion(SRequestObj* pRequest, void* res) {
     }
 
     for (int32_t i = 0; i < pRsp->nBlocks; ++i) {
-      SSubmitBlkRsp *blk = pRsp->pBlocks + i;
+      SSubmitBlkRsp* blk = pRsp->pBlocks + i;
       STbSVersion    tbSver = {.tbFName = blk->tblFName, .sver = blk->sver};
       taosArrayPush(pArray, &tbSver);
     }
   } else if (TDMT_VND_QUERY == pRequest->type) {
-
   }
 
   SCatalog* pCatalog = NULL;
@@ -369,7 +368,6 @@ void freeRequestRes(SRequestObj* pRequest, void* res) {
   if (TDMT_VND_SUBMIT == pRequest->type) {
     tFreeSSubmitRsp((SSubmitRsp*)res);
   } else if (TDMT_VND_QUERY == pRequest->type) {
-
   }
 }
 
@@ -1022,7 +1020,6 @@ TSDB_SERVER_STATUS taos_check_server_status(const char* fqdn, int port, char* de
   SRpcInit           rpcInit = {0};
   char               pass[TSDB_PASSWORD_LEN + 1] = {0};
 
-  taosEncryptPass_c((uint8_t*)("_pwd"), strlen("_pwd"), pass);
   rpcInit.label = "CHK";
   rpcInit.numOfThreads = 1;
   rpcInit.cfp = NULL;
@@ -1030,9 +1027,6 @@ TSDB_SERVER_STATUS taos_check_server_status(const char* fqdn, int port, char* de
   rpcInit.connType = TAOS_CONN_CLIENT;
   rpcInit.idleTime = tsShellActivityTimer * 1000;
   rpcInit.user = "_dnd";
-  rpcInit.ckey = "_key";
-  rpcInit.spi = 1;
-  rpcInit.secret = pass;
 
   clientRpc = rpcOpen(&rpcInit);
   if (clientRpc == NULL) {

source/dnode/mgmt/mgmt_bnode/inc/bmInt.h

@@ -36,9 +36,9 @@ typedef struct SBnodeMgmt {
 
 // bmHandle.c
 SArray *bmGetMsgHandles();
-int32_t bmProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pReq);
-int32_t bmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pReq);
-int32_t bmProcessGetMonBmInfoReq(SBnodeMgmt *pMgmt, SRpcMsg *pReq);
+int32_t bmProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg);
+int32_t bmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg);
+int32_t bmProcessGetMonBmInfoReq(SBnodeMgmt *pMgmt, SRpcMsg *pMsg);
 
 // bmWorker.c
 int32_t bmStartWorker(SBnodeMgmt *pMgmt);

source/dnode/mgmt/mgmt_bnode/src/bmHandle.c

@@ -18,7 +18,7 @@
 
 void bmGetMonitorInfo(SBnodeMgmt *pMgmt, SMonBmInfo *bmInfo) {}
 
-int32_t bmProcessGetMonBmInfoReq(SBnodeMgmt *pMgmt, SRpcMsg *pReq) {
+int32_t bmProcessGetMonBmInfoReq(SBnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   SMonBmInfo bmInfo = {0};
   bmGetMonitorInfo(pMgmt, &bmInfo);
   dmGetMonitorSystemInfo(&bmInfo.sys);
@@ -37,17 +37,15 @@ int32_t bmProcessGetMonBmInfoReq(SBnodeMgmt *pMgmt, SRpcMsg *pReq) {
   }
 
   tSerializeSMonBmInfo(pRsp, rspLen, &bmInfo);
-  pReq->info.rsp = pRsp;
-  pReq->info.rspLen = rspLen;
+  pMsg->info.rsp = pRsp;
+  pMsg->info.rspLen = rspLen;
   tFreeSMonBmInfo(&bmInfo);
   return 0;
 }
 
 int32_t bmProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg) {
-  SRpcMsg *pReq = pMsg;
-
   SDCreateBnodeReq createReq = {0};
-  if (tDeserializeSCreateDropMQSBNodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
+  if (tDeserializeSCreateDropMQSBNodeReq(pMsg->pCont, pMsg->contLen, &createReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }
@@ -68,10 +66,8 @@ int32_t bmProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg) {
 }
 
 int32_t bmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg) {
-  SRpcMsg *pReq = pMsg;
-
   SDDropBnodeReq dropReq = {0};
-  if (tDeserializeSCreateDropMQSBNodeReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
+  if (tDeserializeSCreateDropMQSBNodeReq(pMsg->pCont, pMsg->contLen, &dropReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }

source/dnode/mgmt/mgmt_mnode/inc/mmInt.h

@@ -41,18 +41,18 @@ typedef struct SMnodeMgmt {
 
 // mmFile.c
 int32_t mmReadFile(SMnodeMgmt *pMgmt, bool *pDeployed);
-int32_t mmWriteFile(SMnodeMgmt *pMgmt, SDCreateMnodeReq *pReq, bool deployed);
+int32_t mmWriteFile(SMnodeMgmt *pMgmt, SDCreateMnodeReq *pMsg, bool deployed);
 
 // mmInt.c
-int32_t mmAlter(SMnodeMgmt *pMgmt, SDAlterMnodeReq *pReq);
+int32_t mmAlter(SMnodeMgmt *pMgmt, SDAlterMnodeReq *pMsg);
 
 // mmHandle.c
 SArray *mmGetMsgHandles();
 int32_t mmProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg);
 int32_t mmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg);
 int32_t mmProcessAlterReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t mmProcessGetMonitorInfoReq(SMnodeMgmt *pMgmt, SRpcMsg *pReq);
-int32_t mmProcessGetLoadsReq(SMnodeMgmt *pMgmt, SRpcMsg *pReq);
+int32_t mmProcessGetMonitorInfoReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t mmProcessGetLoadsReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 
 // mmWorker.c
 int32_t mmStartWorker(SMnodeMgmt *pMgmt);
@@ -62,10 +62,10 @@ int32_t mmPutNodeMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmPutNodeMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmPutNodeMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 int32_t mmPutNodeMsgToMonitorQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
-int32_t mmPutRpcMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pRpc);
-int32_t mmPutRpcMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pRpc);
-int32_t mmPutRpcMsgToWriteQueue(SMnodeMgmt *pMgmt, SRpcMsg *pRpc);
-int32_t mmPutRpcMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pRpc);
+int32_t mmPutRpcMsgToQueryQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t mmPutRpcMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t mmPutRpcMsgToWriteQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t mmPutRpcMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg);
 
 #ifdef __cplusplus
 }

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

@@ -104,7 +104,7 @@ _OVER:
   return code;
 }
 
-int32_t mmWriteFile(SMnodeMgmt *pMgmt, SDCreateMnodeReq *pReq, bool deployed) {
+int32_t mmWriteFile(SMnodeMgmt *pMgmt, SDCreateMnodeReq *pMsg, bool deployed) {
   char file[PATH_MAX] = {0};
   char realfile[PATH_MAX] = {0};
   snprintf(file, sizeof(file), "%s%smnode.json.bak", pMgmt->path, TD_DIRSEP);
@@ -124,11 +124,11 @@ int32_t mmWriteFile(SMnodeMgmt *pMgmt, SDCreateMnodeReq *pReq, bool deployed) {
   len += snprintf(content + len, maxLen - len, "{\n");
   len += snprintf(content + len, maxLen - len, "  \"mnodes\": [{\n");
 
-  int8_t replica = (pReq != NULL ? pReq->replica : pMgmt->replica);
+  int8_t replica = (pMsg != NULL ? pMsg->replica : pMgmt->replica);
   for (int32_t i = 0; i < replica; ++i) {
     SReplica *pReplica = &pMgmt->replicas[i];
-    if (pReq != NULL) {
-      pReplica = &pReq->replicas[i];
+    if (pMsg != NULL) {
+      pReplica = &pMsg->replicas[i];
     }
     len += snprintf(content + len, maxLen - len, "    \"id\": %d,\n", pReplica->id);
     len += snprintf(content + len, maxLen - len, "    \"fqdn\": \"%s\",\n", pReplica->fqdn);

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

@@ -25,7 +25,7 @@ void mmGetMnodeLoads(SMnodeMgmt *pMgmt, SMonMloadInfo *pInfo) {
   mndGetLoad(pMgmt->pMnode, &pInfo->load);
 }
 
-int32_t mmProcessGetMonitorInfoReq(SMnodeMgmt *pMgmt, SRpcMsg *pReq) {
+int32_t mmProcessGetMonitorInfoReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   SMonMmInfo mmInfo = {0};
   mmGetMonitorInfo(pMgmt, &mmInfo);
   dmGetMonitorSystemInfo(&mmInfo.sys);
@@ -44,13 +44,13 @@ int32_t mmProcessGetMonitorInfoReq(SMnodeMgmt *pMgmt, SRpcMsg *pReq) {
   }
 
   tSerializeSMonMmInfo(pRsp, rspLen, &mmInfo);
-  pReq->info.rsp = pRsp;
-  pReq->info.rspLen = rspLen;
+  pMsg->info.rsp = pRsp;
+  pMsg->info.rspLen = rspLen;
   tFreeSMonMmInfo(&mmInfo);
   return 0;
 }
 
-int32_t mmProcessGetLoadsReq(SMnodeMgmt *pMgmt, SRpcMsg *pReq) {
+int32_t mmProcessGetLoadsReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   SMonMloadInfo mloads = {0};
   mmGetMnodeLoads(pMgmt, &mloads);
 
@@ -67,16 +67,14 @@ int32_t mmProcessGetLoadsReq(SMnodeMgmt *pMgmt, SRpcMsg *pReq) {
   }
 
   tSerializeSMonMloadInfo(pRsp, rspLen, &mloads);
-  pReq->info.rsp = pRsp;
-  pReq->info.rspLen = rspLen;
+  pMsg->info.rsp = pRsp;
+  pMsg->info.rspLen = rspLen;
   return 0;
 }
 
 int32_t mmProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg) {
-  SRpcMsg *pReq = pMsg;
-
   SDCreateMnodeReq createReq = {0};
-  if (tDeserializeSDCreateMnodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
+  if (tDeserializeSDCreateMnodeReq(pMsg->pCont, pMsg->contLen, &createReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }
@@ -101,10 +99,8 @@ int32_t mmProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg) {
 }
 
 int32_t mmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg) {
-  SRpcMsg *pReq = pMsg;
-
   SDDropMnodeReq dropReq = {0};
-  if (tDeserializeSCreateDropMQSBNodeReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
+  if (tDeserializeSCreateDropMQSBNodeReq(pMsg->pCont, pMsg->contLen, &dropReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }
@@ -129,10 +125,8 @@ int32_t mmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg) {
 }
 
 int32_t mmProcessAlterReq(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  SRpcMsg *pReq = pMsg;
-
   SDAlterMnodeReq alterReq = {0};
-  if (tDeserializeSDCreateMnodeReq(pReq->pCont, pReq->contLen, &alterReq) != 0) {
+  if (tDeserializeSDCreateMnodeReq(pMsg->pCont, pMsg->contLen, &alterReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }

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

@@ -87,9 +87,9 @@ static int32_t mmBuildOptionFromReq(SMnodeMgmt *pMgmt, SMnodeOpt *pOption, SDCre
   return 0;
 }
 
-int32_t mmAlter(SMnodeMgmt *pMgmt, SDAlterMnodeReq *pReq) {
+int32_t mmAlter(SMnodeMgmt *pMgmt, SDAlterMnodeReq *pMsg) {
   SMnodeOpt option = {0};
-  if (mmBuildOptionFromReq(pMgmt, &option, pReq) != 0) {
+  if (mmBuildOptionFromReq(pMgmt, &option, pMsg) != 0) {
     return -1;
   }
 
@@ -98,7 +98,7 @@ int32_t mmAlter(SMnodeMgmt *pMgmt, SDAlterMnodeReq *pReq) {
   }
 
   bool deployed = true;
-  if (mmWriteFile(pMgmt, pReq, deployed) != 0) {
+  if (mmWriteFile(pMgmt, pMsg, deployed) != 0) {
     dError("failed to write mnode file since %s", terrstr());
     return -1;
   }
@@ -135,7 +135,7 @@ static int32_t mmOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
   pMgmt->msgCb.queueFps[QUERY_QUEUE] = (PutToQueueFp)mmPutRpcMsgToQueryQueue;
   pMgmt->msgCb.queueFps[READ_QUEUE] = (PutToQueueFp)mmPutRpcMsgToReadQueue;
   pMgmt->msgCb.queueFps[WRITE_QUEUE] = (PutToQueueFp)mmPutRpcMsgToWriteQueue;
-  pMgmt->msgCb.queueFps[SYNC_QUEUE] = (PutToQueueFp)mmPutRpcMsgToWriteQueue;
+  pMgmt->msgCb.queueFps[SYNC_QUEUE] = (PutToQueueFp)mmPutRpcMsgToSyncQueue;
   pMgmt->msgCb.mgmt = pMgmt;
 
   bool deployed = false;

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

@@ -126,7 +126,7 @@ int32_t mmPutRpcMsgToReadQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   return mmPutRpcMsgToWorker(&pMgmt->readWorker, pMsg);
 }
 
-int32_t mmPutMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { return mmPutRpcMsgToWorker(&pMgmt->syncWorker, pMsg); }
+int32_t mmPutRpcMsgToSyncQueue(SMnodeMgmt *pMgmt, SRpcMsg *pMsg) { return mmPutRpcMsgToWorker(&pMgmt->syncWorker, pMsg); }
 
 int32_t mmStartWorker(SMnodeMgmt *pMgmt) {
   SSingleWorkerCfg qCfg = {

source/dnode/mgmt/mgmt_qnode/inc/qmInt.h

@@ -39,7 +39,7 @@ typedef struct SQnodeMgmt {
 SArray *qmGetMsgHandles();
 int32_t qmProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg);
 int32_t qmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg);
-int32_t qmProcessGetMonitorInfoReq(SQnodeMgmt *pMgmt, SRpcMsg *pReq);
+int32_t qmProcessGetMonitorInfoReq(SQnodeMgmt *pMgmt, SRpcMsg *pMsg);
 
 // qmWorker.c
 int32_t qmPutRpcMsgToQueryQueue(SQnodeMgmt *pMgmt, SRpcMsg *pMsg);

source/dnode/mgmt/mgmt_qnode/src/qmHandle.c

@@ -18,7 +18,7 @@
 
 void qmGetMonitorInfo(SQnodeMgmt *pMgmt, SMonQmInfo *qmInfo) {}
 
-int32_t qmProcessGetMonitorInfoReq(SQnodeMgmt *pMgmt, SRpcMsg *pReq) {
+int32_t qmProcessGetMonitorInfoReq(SQnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   SMonQmInfo qmInfo = {0};
   qmGetMonitorInfo(pMgmt, &qmInfo);
   dmGetMonitorSystemInfo(&qmInfo.sys);
@@ -37,17 +37,15 @@ int32_t qmProcessGetMonitorInfoReq(SQnodeMgmt *pMgmt, SRpcMsg *pReq) {
   }
 
   tSerializeSMonQmInfo(pRsp, rspLen, &qmInfo);
-  pReq->info.rsp = pRsp;
-  pReq->info.rspLen = rspLen;
+  pMsg->info.rsp = pRsp;
+  pMsg->info.rspLen = rspLen;
   tFreeSMonQmInfo(&qmInfo);
   return 0;
 }
 
 int32_t qmProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg) {
-  SRpcMsg *pReq = pMsg;
-
   SDCreateQnodeReq createReq = {0};
-  if (tDeserializeSCreateDropMQSBNodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
+  if (tDeserializeSCreateDropMQSBNodeReq(pMsg->pCont, pMsg->contLen, &createReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }
@@ -68,10 +66,8 @@ int32_t qmProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg) {
 }
 
 int32_t qmProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg) {
-  SRpcMsg *pReq = pMsg;
-
   SDDropQnodeReq dropReq = {0};
-  if (tDeserializeSCreateDropMQSBNodeReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
+  if (tDeserializeSCreateDropMQSBNodeReq(pMsg->pCont, pMsg->contLen, &dropReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }

source/dnode/mgmt/mgmt_snode/inc/smInt.h

@@ -40,7 +40,7 @@ typedef struct SSnodeMgmt {
 SArray *smGetMsgHandles();
 int32_t smProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg);
 int32_t smProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg);
-int32_t smProcessGetMonitorInfoReq(SSnodeMgmt *pMgmt, SRpcMsg *pReq);
+int32_t smProcessGetMonitorInfoReq(SSnodeMgmt *pMgmt, SRpcMsg *pMsg);
 
 // smWorker.c
 int32_t smStartWorker(SSnodeMgmt *pMgmt);

source/dnode/mgmt/mgmt_snode/src/smHandle.c

@@ -18,7 +18,7 @@
 
 void smGetMonitorInfo(SSnodeMgmt *pMgmt, SMonSmInfo *smInfo) {}
 
-int32_t smProcessGetMonitorInfoReq(SSnodeMgmt *pMgmt, SRpcMsg *pReq) {
+int32_t smProcessGetMonitorInfoReq(SSnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   SMonSmInfo smInfo = {0};
   smGetMonitorInfo(pMgmt, &smInfo);
   dmGetMonitorSystemInfo(&smInfo.sys);
@@ -37,17 +37,15 @@ int32_t smProcessGetMonitorInfoReq(SSnodeMgmt *pMgmt, SRpcMsg *pReq) {
   }
 
   tSerializeSMonSmInfo(pRsp, rspLen, &smInfo);
-  pReq->info.rsp = pRsp;
-  pReq->info.rspLen = rspLen;
+  pMsg->info.rsp = pRsp;
+  pMsg->info.rspLen = rspLen;
   tFreeSMonSmInfo(&smInfo);
   return 0;
 }
 
 int32_t smProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg) {
-  SRpcMsg *pReq = pMsg;
-
   SDCreateSnodeReq createReq = {0};
-  if (tDeserializeSCreateDropMQSBNodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
+  if (tDeserializeSCreateDropMQSBNodeReq(pMsg->pCont, pMsg->contLen, &createReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }
@@ -68,10 +66,8 @@ int32_t smProcessCreateReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg) {
 }
 
 int32_t smProcessDropReq(const SMgmtInputOpt *pInput, SRpcMsg *pMsg) {
-  SRpcMsg *pReq = pMsg;
-
   SDDropSnodeReq dropReq = {0};
-  if (tDeserializeSCreateDropMQSBNodeReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
+  if (tDeserializeSCreateDropMQSBNodeReq(pMsg->pCont, pMsg->contLen, &dropReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }

source/dnode/mgmt/mgmt_vnode/inc/vmInt.h

@@ -84,10 +84,10 @@ void       vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode);
 
 // vmHandle.c
 SArray *vmGetMsgHandles();
-int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pReq);
-int32_t vmProcessDropVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pReq);
-int32_t vmProcessGetMonitorInfoReq(SVnodeMgmt *pMgmt, SRpcMsg *pReq);
-int32_t vmProcessGetLoadsReq(SVnodeMgmt *pMgmt, SRpcMsg *pReq);
+int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t vmProcessDropVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t vmProcessGetMonitorInfoReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
+int32_t vmProcessGetLoadsReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg);
 
 // vmFile.c
 int32_t     vmGetVnodeListFromFile(SVnodeMgmt *pMgmt, SWrapperCfg **ppCfgs, int32_t *numOfVnodes);

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

@@ -82,7 +82,7 @@ void vmGetMonitorInfo(SVnodeMgmt *pMgmt, SMonVmInfo *pInfo) {
   taosArrayDestroy(pVloads);
 }
 
-int32_t vmProcessGetMonitorInfoReq(SVnodeMgmt *pMgmt, SRpcMsg *pReq) {
+int32_t vmProcessGetMonitorInfoReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   SMonVmInfo vmInfo = {0};
   vmGetMonitorInfo(pMgmt, &vmInfo);
   dmGetMonitorSystemInfo(&vmInfo.sys);
@@ -101,13 +101,13 @@ int32_t vmProcessGetMonitorInfoReq(SVnodeMgmt *pMgmt, SRpcMsg *pReq) {
   }
 
   tSerializeSMonVmInfo(pRsp, rspLen, &vmInfo);
-  pReq->info.rsp = pRsp;
-  pReq->info.rspLen = rspLen;
+  pMsg->info.rsp = pRsp;
+  pMsg->info.rspLen = rspLen;
   tFreeSMonVmInfo(&vmInfo);
   return 0;
 }
 
-int32_t vmProcessGetLoadsReq(SVnodeMgmt *pMgmt, SRpcMsg *pReq) {
+int32_t vmProcessGetLoadsReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   SMonVloadInfo vloads = {0};
   vmGetVnodeLoads(pMgmt, &vloads);
 
@@ -124,8 +124,8 @@ int32_t vmProcessGetLoadsReq(SVnodeMgmt *pMgmt, SRpcMsg *pReq) {
   }
 
   tSerializeSMonVloadInfo(pRsp, rspLen, &vloads);
-  pReq->info.rsp = pRsp;
-  pReq->info.rspLen = rspLen;
+  pMsg->info.rsp = pRsp;
+  pMsg->info.rspLen = rspLen;
   tFreeSMonVloadInfo(&vloads);
   return 0;
 }
@@ -174,12 +174,11 @@ static void vmGenerateWrapperCfg(SVnodeMgmt *pMgmt, SCreateVnodeReq *pCreate, SW
 }
 
 int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  SRpcMsg        *pReq = pMsg;
   SCreateVnodeReq createReq = {0};
   int32_t         code = -1;
   char            path[TSDB_FILENAME_LEN] = {0};
 
-  if (tDeserializeSCreateVnodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
+  if (tDeserializeSCreateVnodeReq(pMsg->pCont, pMsg->contLen, &createReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }
@@ -242,9 +241,8 @@ _OVER:
 }
 
 int32_t vmProcessDropVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  SRpcMsg      *pReq = pMsg;
   SDropVnodeReq dropReq = {0};
-  if (tDeserializeSDropVnodeReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
+  if (tDeserializeSDropVnodeReq(pMsg->pCont, pMsg->contLen, &dropReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }

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

@@ -126,7 +126,6 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
       syncGetEpSet(vnodeGetSyncHandle(pVnode->pImpl), &newEpSet);
       newEpSet.inUse = (newEpSet.inUse + 1) % newEpSet.numOfEps;
       tmsgSendRedirectRsp(&rsp, &newEpSet);
-
     } else if (ret == TAOS_SYNC_PROPOSE_OTHER_ERROR) {
       rsp.code = TSDB_CODE_SYN_INTERNAL_ERROR;
       tmsgSendRsp(&rsp);

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

@@ -103,7 +103,7 @@ static inline int32_t dmPushToProcQueue(SProc *proc, SProcQueue *queue, SRpcMsg
     return -1;
   }
 
-  if (ftype == DND_FUNC_REQ && IsReq(pMsg) && pMsg->code == 0 && handle != 0) {
+  if (ftype == DND_FUNC_REQ && IsReq(pMsg) && pMsg->code == 0 && handle != 0 && pMsg->info.noResp == 0) {
     if (taosHashPut(proc->hash, &handle, sizeof(int64_t), &pMsg->info, sizeof(SRpcConnInfo)) != 0) {
       taosThreadMutexUnlock(&queue->mutex);
       return -1;

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

@@ -49,9 +49,9 @@ int32_t dmProcessNodeMsg(SMgmtWrapper *pWrapper, SRpcMsg *pMsg) {
 }
 
 static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) {
-  SDnodeTrans  *pTrans = &pDnode->trans;
+  SDnodeTrans * pTrans = &pDnode->trans;
   int32_t       code = -1;
-  SRpcMsg      *pMsg = NULL;
+  SRpcMsg *     pMsg = NULL;
   bool          needRelease = false;
   SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pRpc->msgType)];
   SMgmtWrapper *pWrapper = NULL;
@@ -179,11 +179,11 @@ int32_t dmInitMsgHandle(SDnode *pDnode) {
 
   for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {
     SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
-    SArray       *pArray = (*pWrapper->func.getHandlesFp)();
+    SArray *      pArray = (*pWrapper->func.getHandlesFp)();
     if (pArray == NULL) return -1;
 
     for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
-      SMgmtHandle  *pMgmt = taosArrayGet(pArray, i);
+      SMgmtHandle * pMgmt = taosArrayGet(pArray, i);
       SDnodeHandle *pHandle = &pTrans->msgHandles[TMSG_INDEX(pMgmt->msgType)];
       if (pMgmt->needCheckVgId) {
         pHandle->needCheckVgId = pMgmt->needCheckVgId;
@@ -200,94 +200,54 @@ int32_t dmInitMsgHandle(SDnode *pDnode) {
   return 0;
 }
 
-static void dmSendRpcRedirectRsp(const SRpcMsg *pReq) {
-  SDnode *pDnode = dmInstance();
-  SEpSet  epSet = {0};
-  dmGetMnodeEpSet(&pDnode->data, &epSet);
-
-  dDebug("RPC %p, req is redirected, num:%d use:%d", pReq->info.handle, epSet.numOfEps, epSet.inUse);
-  for (int32_t i = 0; i < epSet.numOfEps; ++i) {
-    dDebug("mnode index:%d %s:%u", i, epSet.eps[i].fqdn, epSet.eps[i].port);
-    if (strcmp(epSet.eps[i].fqdn, tsLocalFqdn) == 0 && epSet.eps[i].port == tsServerPort) {
-      epSet.inUse = (i + 1) % epSet.numOfEps;
-    }
-
-    epSet.eps[i].port = htons(epSet.eps[i].port);
-  }
-
-  SMEpSet msg = {.epSet = epSet};
-  int32_t len = tSerializeSMEpSet(NULL, 0, &msg);
-
-  SRpcMsg rsp = {
-      .code = TSDB_CODE_RPC_REDIRECT,
-      .info = pReq->info,
-      .contLen = len,
-  };
-  rsp.pCont = rpcMallocCont(len);
-  tSerializeSMEpSet(rsp.pCont, len, &msg);
-  rpcSendResponse(&rsp);
-}
-
-static inline void dmSendRecv(SEpSet *pEpSet, SRpcMsg *pReq, SRpcMsg *pRsp) {
+static inline int32_t dmSendReq(const SEpSet *pEpSet, SRpcMsg *pMsg) {
   SDnode *pDnode = dmInstance();
   if (pDnode->status != DND_STAT_RUNNING) {
-    pRsp->code = TSDB_CODE_NODE_OFFLINE;
-    rpcFreeCont(pReq->pCont);
-    pReq->pCont = NULL;
-  } else {
-    rpcSendRecv(pDnode->trans.clientRpc, pEpSet, pReq, pRsp);
-  }
-}
-
-static inline int32_t dmSendReq(const SEpSet *pEpSet, SRpcMsg *pReq) {
-  SDnode *pDnode = dmInstance();
-  if (pDnode->status != DND_STAT_RUNNING) {
-    rpcFreeCont(pReq->pCont);
-    pReq->pCont = NULL;
+    rpcFreeCont(pMsg->pCont);
+    pMsg->pCont = NULL;
     terrno = TSDB_CODE_NODE_OFFLINE;
-    dError("failed to send rpc msg since %s, handle:%p", terrstr(), pReq->info.handle);
+    dError("failed to send rpc msg since %s, handle:%p", terrstr(), pMsg->info.handle);
     return -1;
   } else {
-    rpcSendRequest(pDnode->trans.clientRpc, pEpSet, pReq, NULL);
+    rpcSendRequest(pDnode->trans.clientRpc, pEpSet, pMsg, NULL);
     return 0;
   }
 }
 
 static inline void dmSendRsp(SRpcMsg *pMsg) {
   SMgmtWrapper *pWrapper = pMsg->info.wrapper;
-  if (pMsg->code == TSDB_CODE_NODE_REDIRECT) {
-    dmSendRpcRedirectRsp(pMsg);
-  } else {
   if (InChildProc(pWrapper)) {
     dmPutToProcPQueue(&pWrapper->proc, pMsg, DND_FUNC_RSP);
+    rpcFreeCont(pMsg->pCont);
+    pMsg->pCont = NULL;
   } else {
     rpcSendResponse(pMsg);
   }
-  }
 }
 
 static inline void dmSendRedirectRsp(SRpcMsg *pMsg, const SEpSet *pNewEpSet) {
-  SMgmtWrapper *pWrapper = pMsg->info.wrapper;
-  if (InChildProc(pWrapper)) {
-    dmPutToProcPQueue(&pWrapper->proc, pMsg, DND_FUNC_RSP);
-  } else {
-    SRpcMsg rsp = {0};
+  SRpcMsg rsp = {.code = TSDB_CODE_RPC_REDIRECT, .info = pMsg->info};
   SMEpSet msg = {.epSet = *pNewEpSet};
-    int32_t len = tSerializeSMEpSet(NULL, 0, &msg);
-    rsp.pCont = rpcMallocCont(len);
-    rsp.contLen = len;
-    tSerializeSMEpSet(rsp.pCont, len, &msg);
+  int32_t contLen = tSerializeSMEpSet(NULL, 0, &msg);
 
-    rsp.code = TSDB_CODE_RPC_REDIRECT;
-    rsp.info = pMsg->info;
-    rpcSendResponse(&rsp);
+  rsp.pCont = rpcMallocCont(contLen);
+  if (rsp.pCont == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+  } else {
+    tSerializeSMEpSet(rsp.pCont, contLen, &msg);
+    rsp.contLen = contLen;
   }
+  dmSendRsp(&rsp);
+  rpcFreeCont(pMsg->pCont);
+  pMsg->pCont = NULL;
 }
 
 static inline void dmRegisterBrokenLinkArg(SRpcMsg *pMsg) {
   SMgmtWrapper *pWrapper = pMsg->info.wrapper;
   if (InChildProc(pWrapper)) {
     dmPutToProcPQueue(&pWrapper->proc, pMsg, DND_FUNC_REGIST);
+    rpcFreeCont(pMsg->pCont);
+    pMsg->pCont = NULL;
   } else {
     rpcRegisterBrokenLinkArg(pMsg);
   }
@@ -316,15 +276,9 @@ int32_t dmInitClient(SDnode *pDnode) {
   rpcInit.connType = TAOS_CONN_CLIENT;
   rpcInit.idleTime = tsShellActivityTimer * 1000;
   rpcInit.user = INTERNAL_USER;
-  rpcInit.ckey = INTERNAL_CKEY;
-  rpcInit.spi = 1;
   rpcInit.parent = pDnode;
   rpcInit.rfp = rpcRfp;
 
-  char pass[TSDB_PASSWORD_LEN + 1] = {0};
-  taosEncryptPass_c((uint8_t *)(INTERNAL_SECRET), strlen(INTERNAL_SECRET), pass);
-  rpcInit.secret = pass;
-
   pTrans->clientRpc = rpcOpen(&rpcInit);
   if (pTrans->clientRpc == NULL) {
     dError("failed to init dnode rpc client");
@@ -389,3 +343,34 @@ SMsgCb dmGetMsgcb(SDnode *pDnode) {
   };
   return msgCb;
 }
+
+static void dmSendMnodeRedirectRsp(SRpcMsg *pMsg) {
+  SDnode *pDnode = dmInstance();
+  SEpSet  epSet = {0};
+  dmGetMnodeEpSet(&pDnode->data, &epSet);
+
+  dDebug("msg:%p, is redirected, num:%d use:%d", pMsg, epSet.numOfEps, epSet.inUse);
+  for (int32_t i = 0; i < epSet.numOfEps; ++i) {
+    dDebug("mnode index:%d %s:%u", i, epSet.eps[i].fqdn, epSet.eps[i].port);
+    if (strcmp(epSet.eps[i].fqdn, tsLocalFqdn) == 0 && epSet.eps[i].port == tsServerPort) {
+      epSet.inUse = (i + 1) % epSet.numOfEps;
+    }
+
+    epSet.eps[i].port = htons(epSet.eps[i].port);
+  }
+
+  SRpcMsg rsp = {.code = TSDB_CODE_RPC_REDIRECT, .info = pMsg->info};
+  SMEpSet msg = {.epSet = epSet};
+  int32_t contLen = tSerializeSMEpSet(NULL, 0, &msg);
+  rsp.pCont = rpcMallocCont(contLen);
+  if (rsp.pCont == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+  } else {
+    tSerializeSMEpSet(rsp.pCont, contLen, &msg);
+    rsp.contLen = contLen;
+  }
+
+  dmSendRsp(&rsp);
+  rpcFreeCont(pMsg->pCont);
+  pMsg->pCont = NULL;
+}

source/dnode/mgmt/test/sut/src/client.cpp

@@ -48,10 +48,10 @@ void TestClient::DoInit() {
   rpcInit.connType = TAOS_CONN_CLIENT;
   rpcInit.idleTime = 30 * 1000;
   rpcInit.user = (char*)this->user;
-  rpcInit.ckey = (char*)"key";
+  // rpcInit.ckey = (char*)"key";
   rpcInit.parent = this;
-  rpcInit.secret = (char*)secretEncrypt;
-  rpcInit.spi = 1;
+  // rpcInit.secret = (char*)secretEncrypt;
+  // rpcInit.spi = 1;
 
   clientRpc = rpcOpen(&rpcInit);
   ASSERT(clientRpc);

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

@@ -624,14 +624,12 @@ static int32_t mndProcessConfigDnodeReq(SRpcMsg *pReq) {
 
   SRpcMsg rpcMsg = {.msgType = TDMT_DND_CONFIG_DNODE, .pCont = pBuf, .contLen = bufLen, .info = pReq->info};
 
-  mInfo("dnode:%d, app:%p config:%s req send to dnode", cfgReq.dnodeId, rpcMsg.info.ahandle, cfgReq.config);
-  tmsgSendReq(&epSet, &rpcMsg);
-
-  return 0;
+  mDebug("dnode:%d, send config req to dnode, app:%p", cfgReq.dnodeId, rpcMsg.info.ahandle);
+  return tmsgSendReq(&epSet, &rpcMsg);
 }
 
 static int32_t mndProcessConfigDnodeRsp(SRpcMsg *pRsp) {
-  mInfo("app:%p config rsp from dnode", pRsp->info.ahandle);
+  mDebug("config rsp from dnode, app:%p", pRsp->info.ahandle);
   return TSDB_CODE_SUCCESS;
 }
 

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

@@ -85,10 +85,9 @@ int vnodeAsyncCommit(SVnode* pVnode);
 int32_t   vnodeSyncOpen(SVnode* pVnode, char* path);
 int32_t   vnodeSyncStart(SVnode* pVnode);
 void      vnodeSyncClose(SVnode* pVnode);
-void      vnodeSyncSetQ(SVnode* pVnode, void* qHandle);
-void      vnodeSyncSetRpc(SVnode* pVnode, void* rpcHandle);
-int32_t   vnodeSyncEqMsg(void* qHandle, SRpcMsg* pMsg);
-int32_t   vnodeSendMsg(void* rpcHandle, const SEpSet* pEpSet, SRpcMsg* pMsg);
+void      vnodeSyncSetMsgCb(SVnode* pVnode);
+int32_t   vnodeSyncEqMsg(const SMsgCb* msgcb, SRpcMsg* pMsg);
+int32_t   vnodeSyncSendMsg(const SEpSet* pEpSet, SRpcMsg* pMsg);
 void      vnodeSyncCommitCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta);
 void      vnodeSyncPreCommitCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta);
 void      vnodeSyncRollBackCb(struct SSyncFSM* pFsm, const SRpcMsg* pMsg, SFsmCbMeta cbMeta);

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

@@ -420,6 +420,11 @@ static STsdbReadHandle* tsdbQueryTablesImpl(SVnode* pVnode, SQueryTableDataCond*
   setQueryTimewindow(pReadHandle, pCond);
 
   if (pCond->numOfCols > 0) {
+    int32_t rowLen = 0;
+    for(int32_t i = 0; i < pCond->numOfCols; ++i) {
+      rowLen += pCond->colList[i].bytes;
+    }
+
     // allocate buffer in order to load data blocks from file
     pReadHandle->suppInfo.pstatis = taosMemoryCalloc(pCond->numOfCols, sizeof(SColumnDataAgg));
     if (pReadHandle->suppInfo.pstatis == NULL) {

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

@@ -180,8 +180,7 @@ void vnodeClose(SVnode *pVnode) {
 
 // start the sync timer after the queue is ready
 int32_t vnodeStart(SVnode *pVnode) {
-  vnodeSyncSetQ(pVnode, NULL);
-  vnodeSyncSetRpc(pVnode, NULL);
+  vnodeSyncSetMsgCb(pVnode);
   vnodeSyncStart(pVnode);
   return 0;
 }

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

@@ -682,7 +682,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
 
       submitBlkRsp.uid = createTbReq.uid;
       submitBlkRsp.tblFName = taosMemoryMalloc(strlen(pVnode->config.dbname) + strlen(createTbReq.name) + 2);
-      sprintf(submitBlkRsp.tblFName, "%s.%s", pVnode->config.dbname, createTbReq.name);
+      sprintf(submitBlkRsp.tblFName, "%s.", pVnode->config.dbname);
 
       msgIter.uid = createTbReq.uid;
       if (createTbReq.type == TSDB_CHILD_TABLE) {

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

@@ -27,9 +27,8 @@ int32_t vnodeSyncOpen(SVnode *pVnode, char *path) {
   syncInfo.pWal = pVnode->pWal;
 
   syncInfo.pFsm = syncVnodeMakeFsm(pVnode);
-  syncInfo.rpcClient = NULL;
-  syncInfo.FpSendMsg = vnodeSendMsg;
-  syncInfo.queue = NULL;
+  syncInfo.msgcb = NULL;
+  syncInfo.FpSendMsg = vnodeSyncSendMsg;
   syncInfo.FpEqMsg = vnodeSyncEqMsg;
 
   pVnode->sync = syncOpen(&syncInfo);
@@ -53,31 +52,13 @@ void vnodeSyncClose(SVnode *pVnode) {
   syncStop(pVnode->sync);
 }
 
-void vnodeSyncSetQ(SVnode *pVnode, void *qHandle) { syncSetQ(pVnode->sync, (void *)(&(pVnode->msgCb))); }
+void vnodeSyncSetMsgCb(SVnode *pVnode) { syncSetMsgCb(pVnode->sync, &pVnode->msgCb); }
 
-void vnodeSyncSetRpc(SVnode *pVnode, void *rpcHandle) { syncSetRpc(pVnode->sync, (void *)(&(pVnode->msgCb))); }
+int32_t vnodeSyncEqMsg(const SMsgCb *msgcb, SRpcMsg *pMsg) { return tmsgPutToQueue(msgcb, SYNC_QUEUE, pMsg); }
 
-int32_t vnodeSyncEqMsg(void *qHandle, SRpcMsg *pMsg) {
-  int32_t ret = 0;
-  SMsgCb *pMsgCb = qHandle;
-  if (pMsgCb->queueFps[SYNC_QUEUE] != NULL) {
-    tmsgPutToQueue(qHandle, SYNC_QUEUE, pMsg);
-  } else {
-    vError("vnodeSyncEqMsg queue is NULL, SYNC_QUEUE:%d", SYNC_QUEUE);
-  }
-  return ret;
-}
-
-int32_t vnodeSendMsg(void *rpcHandle, const SEpSet *pEpSet, SRpcMsg *pMsg) {
-  int32_t ret = 0;
-  SMsgCb *pMsgCb = rpcHandle;
-  if (pMsgCb->queueFps[SYNC_QUEUE] != NULL) {
+int32_t vnodeSyncSendMsg(const SEpSet *pEpSet, SRpcMsg *pMsg) {
   pMsg->info.noResp = 1;
-    tmsgSendReq(pEpSet, pMsg);
-  } else {
-    vError("vnodeSendMsg queue is NULL, SYNC_QUEUE:%d", SYNC_QUEUE);
-  }
-  return ret;
+  return tmsgSendReq(pEpSet, pMsg);
 }
 
 int32_t vnodeSyncGetSnapshotCb(struct SSyncFSM *pFsm, SSnapshot *pSnapshot) {

source/libs/executor/src/executor.c

@@ -14,8 +14,6 @@
  */
 
 #include "executor.h"
-#include <executorimpl.h>
-#include <vnode.h>
 #include "executorimpl.h"
 #include "planner.h"
 #include "tdatablock.h"

source/libs/executor/src/executorimpl.c

@@ -586,8 +586,7 @@ void doApplyFunctions(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, STimeWindow
     int32_t numOfRows = pCtx[k].input.numOfRows;
     int32_t startOffset = pCtx[k].input.startRowIndex;
 
-    int32_t pos = (order == TSDB_ORDER_ASC) ? offset : offset - (forwardStep - 1);
-    pCtx[k].input.startRowIndex = pos;
+    pCtx[k].input.startRowIndex = offset;
     pCtx[k].input.numOfRows = forwardStep;
 
     if (tsCol != NULL) {
@@ -758,45 +757,6 @@ static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCt
         }
       }
     }
-
-    //    setBlockStatisInfo(&pCtx[i], pBlock, pOperator->pExpr[i].base.pColumns);
-    //      uint32_t flag = pOperator->pExpr[i].base.pParam[0].pCol->flag;
-    //      if (TSDB_COL_IS_NORMAL_COL(flag) /*|| (pCtx[i].functionId == FUNCTION_BLKINFO) ||
-    //          (TSDB_COL_IS_TAG(flag) && pOperator->pRuntimeEnv->scanFlag == MERGE_STAGE)*/) {
-
-    //        SColumn* pCol = pOperator->pExpr[i].base.pParam[0].pCol;
-    //        if (pCtx[i].columnIndex == -1) {
-    //          for(int32_t j = 0; j < pBlock->info.numOfCols; ++j) {
-    //            SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, j);
-    //            if (pColData->info.colId == pCol->colId) {
-    //              pCtx[i].columnIndex = j;
-    //              break;
-    //            }
-    //          }
-    //        }
-
-    //        uint32_t status = aAggs[pCtx[i].functionId].status;
-    //        if ((status & (FUNCSTATE_SELECTIVITY | FUNCSTATE_NEED_TS)) != 0) {
-    //          SColumnInfoData* tsInfo = taosArrayGet(pBlock->pDataBlock, 0);
-    // In case of the top/bottom query again the nest query result, which has no timestamp column
-    // don't set the ptsList attribute.
-    //          if (tsInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
-    //            pCtx[i].ptsList = (int64_t*) tsInfo->pData;
-    //          } else {
-    //            pCtx[i].ptsList = NULL;
-    //          }
-    //        }
-    //      } else if (TSDB_COL_IS_UD_COL(pCol->flag) && (pOperator->pRuntimeEnv->scanFlag == MERGE_STAGE)) {
-    //        SColIndex*       pColIndex = &pOperator->pExpr[i].base.colInfo;
-    //        SColumnInfoData* p = taosArrayGet(pBlock->pDataBlock, pColIndex->colIndex);
-    //
-    //        pCtx[i].pInput = p->pData;
-    //        assert(p->info.colId == pColIndex->info.colId && pCtx[i].inputType == p->info.type);
-    //        for(int32_t j = 0; j < pBlock->info.rows; ++j) {
-    //          char* dst = p->pData + j * p->info.bytes;
-    //          taosVariantDump(&pOperator->pExpr[i].base.param[1], dst, p->info.type, true);
-    //        }
-    //      }
   }
 
   return code;
@@ -2842,9 +2802,11 @@ int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadI
 
     for (int32_t i = 0; i < numOfCols; ++i) {
       SColumnInfoData idata = {0};
+
       idata.info.type  = pSchema[i].type;
       idata.info.bytes = pSchema[i].bytes;
       idata.info.colId = pSchema[i].colId;
+      idata.hasNull    = true;
 
       taosArrayPush(pBlock->pDataBlock, &idata);
       if (IS_VAR_DATA_TYPE(idata.info.type)) {

source/libs/executor/src/scanoperator.c

@@ -898,11 +898,11 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
       SSDataBlock* upRes = getUpdateDataBlock(pInfo, true); //TODO(liuyao) get invertible from plan
       if (upRes) {
         pInfo->pUpdateRes = upRes;
-        if (upRes->info.type = STREAM_REPROCESS) {
+        if (upRes->info.type == STREAM_REPROCESS) {
           pInfo->updateResIndex = 0;
           prepareDataScan(pInfo);
           pInfo->scanMode = STREAM_SCAN_FROM_UPDATERES;
-        } else if (upRes->info.type = STREAM_INVERT) {
+        } else if (upRes->info.type == STREAM_INVERT) {
           pInfo->scanMode = STREAM_SCAN_FROM_RES;
           return upRes;
         }

source/libs/executor/src/timewindowoperator.c

@@ -1072,7 +1072,9 @@ static void doClearWindows(SAggSupporter* pSup, SOptrBasicInfo* pBinfo,
 
 static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
   SIntervalAggOperatorInfo* pInfo = pOperator->info;
-  int32_t                   order = TSDB_ORDER_ASC;
+  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
+
+  pInfo->order = TSDB_ORDER_ASC;
 
   if (pOperator->status == OP_EXEC_DONE) {
     return NULL;
@@ -1086,11 +1088,9 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
     return pInfo->binfo.pRes->info.rows == 0 ? NULL : pInfo->binfo.pRes;
   }
 
-  //  STimeWindow win = {0};
   SOperatorInfo* downstream = pOperator->pDownstream[0];
 
   SArray* pUpdated = NULL;
-
   while (1) {
     publishOperatorProfEvent(downstream, QUERY_PROF_BEFORE_OPERATOR_EXEC);
     SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
@@ -1103,16 +1103,18 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
     // The timewindows that overlaps the timestamps of the input pBlock need to be recalculated and return to the
     // caller. Note that all the time window are not close till now.
     // the pDataBlock are always the same one, no need to call this again
-    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order, MAIN_SCAN, true);
+    setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, pInfo->order, MAIN_SCAN, true);
     if (pInfo->invertible) {
       setInverFunction(pInfo->binfo.pCtx, pOperator->numOfExprs, pBlock->info.type);
     }
+
     if (pBlock->info.type == STREAM_REPROCESS) {
       doClearWindows(&pInfo->aggSup, &pInfo->binfo, &pInfo->interval,
           pInfo->primaryTsIndex, pOperator->numOfExprs, pBlock);
+      qDebug("%s clear existed time window results for updates checked", GET_TASKID(pTaskInfo));
       continue;
     }
-    pInfo->order = TSDB_ORDER_ASC;
+
     pUpdated = hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, 0);
   }
 

source/libs/function/src/functionMgt.c

@@ -207,6 +207,9 @@ bool fmIsInvertible(int32_t funcId) {
     case FUNCTION_TYPE_SUM:
     case FUNCTION_TYPE_STDDEV:
     case FUNCTION_TYPE_AVG:
+    case FUNCTION_TYPE_WSTARTTS:
+    case FUNCTION_TYPE_WENDTS:
+    case FUNCTION_TYPE_WDURATION:
       res = true;
       break;
     default:

source/libs/function/src/udfd.c

@@ -27,16 +27,16 @@
 #include "trpc.h"
 
 typedef struct SUdfdContext {
-  uv_loop_t  *loop;
+  uv_loop_t * loop;
   uv_pipe_t   ctrlPipe;
   uv_signal_t intrSignal;
   char        listenPipeName[PATH_MAX + UDF_LISTEN_PIPE_NAME_LEN + 2];
   uv_pipe_t   listeningPipe;
 
-  void       *clientRpc;
+  void *     clientRpc;
   SCorEpSet  mgmtEp;
   uv_mutex_t udfsMutex;
-  SHashObj  *udfsHash;
+  SHashObj * udfsHash;
 
   bool printVersion;
 } SUdfdContext;
@@ -45,7 +45,7 @@ SUdfdContext global;
 
 typedef struct SUdfdUvConn {
   uv_stream_t *client;
-  char        *inputBuf;
+  char *       inputBuf;
   int32_t      inputLen;
   int32_t      inputCap;
   int32_t      inputTotal;
@@ -99,7 +99,7 @@ typedef enum EUdfdRpcReqRspType {
 typedef struct SUdfdRpcSendRecvInfo {
   EUdfdRpcReqRspType rpcType;
   int32_t            code;
-  void* param;
+  void *             param;
   uv_sem_t           resultSem;
 } SUdfdRpcSendRecvInfo;
 
@@ -136,7 +136,7 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
     tDeserializeSRetrieveFuncRsp(pMsg->pCont, pMsg->contLen, &retrieveRsp);
 
     SFuncInfo *pFuncInfo = (SFuncInfo *)taosArrayGet(retrieveRsp.pFuncInfos, 0);
-    SUdf* udf = msgInfo->param;
+    SUdf *     udf = msgInfo->param;
     udf->funcType = pFuncInfo->funcType;
     udf->scriptType = pFuncInfo->scriptType;
     udf->outputType = pFuncInfo->outputType;
@@ -145,7 +145,8 @@ void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) {
 
     char path[PATH_MAX] = {0};
     snprintf(path, sizeof(path), "%s/lib%s.so", "/tmp", pFuncInfo->name);
-    TdFilePtr file = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC | TD_FILE_AUTO_DEL);
+    TdFilePtr file =
+        taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC | TD_FILE_AUTO_DEL);
     // TODO check for failure of flush to disk
     taosWriteFile(file, pFuncInfo->pCode, pFuncInfo->codeSize);
     taosCloseFile(&file);
@@ -168,11 +169,11 @@ int32_t udfdFillUdfInfoFromMNode(void *clientRpc, char *udfName, SUdf *udf) {
   taosArrayPush(retrieveReq.pFuncNames, udfName);
 
   int32_t contLen = tSerializeSRetrieveFuncReq(NULL, 0, &retrieveReq);
-  void   *pReq = rpcMallocCont(contLen);
+  void *  pReq = rpcMallocCont(contLen);
   tSerializeSRetrieveFuncReq(pReq, contLen, &retrieveReq);
   taosArrayDestroy(retrieveReq.pFuncNames);
 
-  SUdfdRpcSendRecvInfo* msgInfo = taosMemoryCalloc(1, sizeof(SUdfdRpcSendRecvInfo));
+  SUdfdRpcSendRecvInfo *msgInfo = taosMemoryCalloc(1, sizeof(SUdfdRpcSendRecvInfo));
   msgInfo->rpcType = UDFD_RPC_RETRIVE_FUNC;
   msgInfo->param = udf;
   uv_sem_init(&msgInfo->resultSem, 0);
@@ -194,7 +195,7 @@ int32_t udfdFillUdfInfoFromMNode(void *clientRpc, char *udfName, SUdf *udf) {
 int32_t udfdConnectToMnode() {
   SConnectReq connReq = {0};
   connReq.connType = CONN_TYPE__UDFD;
-  tstrncpy(connReq.app, "udfd",sizeof(connReq.app));
+  tstrncpy(connReq.app, "udfd", sizeof(connReq.app));
   tstrncpy(connReq.user, TSDB_DEFAULT_USER, sizeof(connReq.user));
   char pass[TSDB_PASSWORD_LEN + 1] = {0};
   taosEncryptPass_c((uint8_t *)(TSDB_DEFAULT_PASS), strlen(TSDB_DEFAULT_PASS), pass);
@@ -203,7 +204,7 @@ int32_t udfdConnectToMnode() {
   connReq.startTime = htobe64(taosGetTimestampMs());
 
   int32_t contLen = tSerializeSConnectReq(NULL, 0, &connReq);
-  void*   pReq = rpcMallocCont(contLen);
+  void *  pReq = rpcMallocCont(contLen);
   tSerializeSConnectReq(pReq, contLen, &connReq);
 
   SUdfdRpcSendRecvInfo *msgInfo = taosMemoryCalloc(1, sizeof(SUdfdRpcSendRecvInfo));
@@ -240,17 +241,17 @@ int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
     return TSDB_CODE_UDF_LOAD_UDF_FAILURE;
   }
 
-  char initFuncName[TSDB_FUNC_NAME_LEN+5] = {0};
+  char  initFuncName[TSDB_FUNC_NAME_LEN + 5] = {0};
   char *initSuffix = "_init";
   strcpy(initFuncName, udfName);
   strncat(initFuncName, initSuffix, strlen(initSuffix));
-  uv_dlsym(&udf->lib, initFuncName, (void**)(&udf->initFunc));
+  uv_dlsym(&udf->lib, initFuncName, (void **)(&udf->initFunc));
 
-  char destroyFuncName[TSDB_FUNC_NAME_LEN+5] = {0};
+  char  destroyFuncName[TSDB_FUNC_NAME_LEN + 5] = {0};
   char *destroySuffix = "_destroy";
   strcpy(destroyFuncName, udfName);
   strncat(destroyFuncName, destroySuffix, strlen(destroySuffix));
-  uv_dlsym(&udf->lib, destroyFuncName, (void**)(&udf->destroyFunc));
+  uv_dlsym(&udf->lib, destroyFuncName, (void **)(&udf->destroyFunc));
 
   if (udf->funcType == TSDB_FUNC_TYPE_SCALAR) {
     char processFuncName[TSDB_FUNC_NAME_LEN] = {0};
@@ -270,17 +271,17 @@ int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
     strncpy(finishFuncName, processFuncName, strlen(processFuncName));
     strncat(finishFuncName, finishSuffix, strlen(finishSuffix));
     uv_dlsym(&udf->lib, finishFuncName, (void **)(&udf->aggFinishFunc));
-    //TODO: merge
+    // TODO: merge
   }
   return 0;
 }
 
-void udfdProcessSetupRequest(SUvUdfWork* uvUdf, SUdfRequest* request) {
+void udfdProcessSetupRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
   // TODO: tracable id from client. connect, setup, call, teardown
-    fnInfo( "setup request. seq num: %" PRId64 ", udf name: %s", request->seqNum, request->setup.udfName);
+  fnInfo("setup request. seq num: %" PRId64 ", udf name: %s", request->seqNum, request->setup.udfName);
   SUdfSetupRequest *setup = &request->setup;
   int32_t           code = TSDB_CODE_SUCCESS;
-    SUdf *udf = NULL;
+  SUdf *            udf = NULL;
   uv_mutex_lock(&global.udfsMutex);
   SUdf **udfInHash = taosHashGet(global.udfsHash, request->setup.udfName, strlen(request->setup.udfName));
   if (udfInHash) {
@@ -341,16 +342,15 @@ void udfdProcessSetupRequest(SUvUdfWork* uvUdf, SUdfRequest* request) {
 
 void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
   SUdfCallRequest *call = &request->call;
-  fnDebug("%" PRId64 "call request. call type %d, handle: %" PRIx64, request->seqNum, call->callType,
-          call->udfHandle);
-  SUdfcFuncHandle *handle = (SUdfcFuncHandle *)(call->udfHandle);
-  SUdf            *udf = handle->udf;
+  fnDebug("%" PRId64 "call request. call type %d, handle: %" PRIx64, request->seqNum, call->callType, call->udfHandle);
+  SUdfcFuncHandle * handle = (SUdfcFuncHandle *)(call->udfHandle);
+  SUdf *            udf = handle->udf;
   SUdfResponse      response = {0};
-  SUdfResponse *rsp = &response;
+  SUdfResponse *    rsp = &response;
   SUdfCallResponse *subRsp = &rsp->callRsp;
 
   int32_t code = TSDB_CODE_SUCCESS;
-  switch(call->callType) {
+  switch (call->callType) {
     case TSDB_UDF_CALL_SCALA_PROC: {
       SUdfColumn output = {0};
 
@@ -363,9 +363,7 @@ void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
       break;
     }
     case TSDB_UDF_CALL_AGG_INIT: {
-      SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize),
-                             .bufLen= udf->bufSize,
-                             .numOfResult = 0};
+      SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize), .bufLen = udf->bufSize, .numOfResult = 0};
       udf->aggStartFunc(&outBuf);
       subRsp->resultBuf = outBuf;
       break;
@@ -373,9 +371,7 @@ void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
     case TSDB_UDF_CALL_AGG_PROC: {
       SUdfDataBlock input = {0};
       convertDataBlockToUdfDataBlock(&call->block, &input);
-      SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize),
-                             .bufLen= udf->bufSize,
-                             .numOfResult = 0};
+      SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize), .bufLen = udf->bufSize, .numOfResult = 0};
       code = udf->aggProcFunc(&input, &call->interBuf, &outBuf);
       freeUdfInterBuf(&call->interBuf);
       freeUdfDataDataBlock(&input);
@@ -384,9 +380,7 @@ void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
       break;
     }
     case TSDB_UDF_CALL_AGG_FIN: {
-      SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize),
-                             .bufLen= udf->bufSize,
-                             .numOfResult = 0};
+      SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize), .bufLen = udf->bufSize, .numOfResult = 0};
       code = udf->aggFinishFunc(&call->interBuf, &outBuf);
       freeUdfInterBuf(&call->interBuf);
       subRsp->resultBuf = outBuf;
@@ -429,18 +423,17 @@ void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
     }
     default:
       break;
-
   }
 
   taosMemoryFree(uvUdf->input.base);
   return;
 }
 
-void udfdProcessTeardownRequest(SUvUdfWork* uvUdf, SUdfRequest* request) {
+void udfdProcessTeardownRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
   SUdfTeardownRequest *teardown = &request->teardown;
   fnInfo("teardown. seq number: %" PRId64 ", handle:%" PRIx64, request->seqNum, teardown->udfHandle);
   SUdfcFuncHandle *handle = (SUdfcFuncHandle *)(teardown->udfHandle);
-  SUdf *udf = handle->udf;
+  SUdf *           udf = handle->udf;
   bool             unloadUdf = false;
   int32_t          code = TSDB_CODE_SUCCESS;
 
@@ -568,7 +561,7 @@ bool isUdfdUvMsgComplete(SUdfdUvConn *pipe) {
 }
 
 void udfdHandleRequest(SUdfdUvConn *conn) {
-  uv_work_t  *work = taosMemoryMalloc(sizeof(uv_work_t));
+  uv_work_t * work = taosMemoryMalloc(sizeof(uv_work_t));
   SUvUdfWork *udfWork = taosMemoryMalloc(sizeof(SUvUdfWork));
   udfWork->client = conn->client;
   udfWork->input = uv_buf_init(conn->inputBuf, conn->inputLen);
@@ -653,11 +646,11 @@ static bool udfdRpcRfp(int32_t code) {
   }
 }
 
-int initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet) {
+int initEpSetFromCfg(const char *firstEp, const char *secondEp, SCorEpSet *pEpSet) {
   pEpSet->version = 0;
 
   // init mnode ip set
-  SEpSet* mgmtEpSet = &(pEpSet->epSet);
+  SEpSet *mgmtEpSet = &(pEpSet->epSet);
   mgmtEpSet->numOfEps = 0;
   mgmtEpSet->inUse = 0;
 
@@ -694,7 +687,6 @@ int initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSe
   return 0;
 }
 
-
 int32_t udfdOpenClientRpc() {
   SRpcInit rpcInit = {0};
   rpcInit.label = "UDFD";
@@ -704,15 +696,9 @@ int32_t udfdOpenClientRpc() {
   rpcInit.connType = TAOS_CONN_CLIENT;
   rpcInit.idleTime = tsShellActivityTimer * 1000;
   rpcInit.user = TSDB_DEFAULT_USER;
-  rpcInit.ckey = "key";
-  rpcInit.spi = 1;
   rpcInit.parent = &global;
   rpcInit.rfp = udfdRpcRfp;
 
-  char pass[TSDB_PASSWORD_LEN + 1] = {0};
-  taosEncryptPass_c((uint8_t *)(TSDB_DEFAULT_PASS), strlen(TSDB_DEFAULT_PASS), pass);
-  rpcInit.secret = pass;
-
   global.clientRpc = rpcOpen(&rpcInit);
   if (global.clientRpc == NULL) {
     fnError("failed to init dnode rpc client");
@@ -823,7 +809,7 @@ static int32_t udfdUvInit() {
   return 0;
 }
 
-static void udfdCloseWalkCb(uv_handle_t* handle, void* arg) {
+static void udfdCloseWalkCb(uv_handle_t *handle, void *arg) {
   if (!uv_is_closing(handle)) {
     uv_close(handle, NULL);
   }
@@ -883,7 +869,7 @@ int main(int argc, char *argv[]) {
   int32_t retryMnodeTimes = 0;
   int32_t code = 0;
   while (retryMnodeTimes++ < TSDB_MAX_REPLICA) {
-    uv_sleep(500 * ( 1 << retryMnodeTimes));
+    uv_sleep(500 * (1 << retryMnodeTimes));
     code = udfdConnectToMnode();
     if (code == 0) {
       break;

source/libs/function/test/udf2.c

@@ -26,7 +26,7 @@ int32_t udf2_start(SUdfInterBuf *buf) {
 
 int32_t udf2(SUdfDataBlock* block, SUdfInterBuf *interBuf, SUdfInterBuf *newInterBuf) {
   double sumSquares = *(double*)interBuf->buf;
-  int8_t numOutput = 0;
+  int8_t numNotNull = 0;
   for (int32_t i = 0; i < block->numOfCols; ++i) {
     SUdfColumn* col = block->udfCols[i];
     if (!(col->colMeta.type == TSDB_DATA_TYPE_INT || 
@@ -56,15 +56,18 @@ int32_t udf2(SUdfDataBlock* block, SUdfInterBuf *interBuf, SUdfInterBuf *newInte
         default: 
           break;
       }
-      numOutput = 1;
+      ++numNotNull;
     }
   }
 
-  if (numOutput == 1) {
   *(double*)(newInterBuf->buf) = sumSquares;
   newInterBuf->bufLen = sizeof(double);
+
+  if (interBuf->numOfResult == 0 && numNotNull == 0) {
+    newInterBuf->numOfResult = 0;
+  } else {
+    newInterBuf->numOfResult = 1;
   }
-  newInterBuf->numOfResult = numOutput;
   return 0;
 }
 

source/libs/index/src/indexCache.c

@@ -22,7 +22,7 @@
 #define MAX_INDEX_KEY_LEN 256  // test only, change later
 
 #define MEM_TERM_LIMIT     10 * 10000
-#define MEM_THRESHOLD      1024 * 1024
+#define MEM_THRESHOLD      64 * 1024
 #define MEM_ESTIMATE_RADIO 1.5
 
 static void indexMemRef(MemTable* tbl);

source/libs/index/src/indexFst.c

@@ -99,7 +99,7 @@ void fstUnFinishedNodesAddSuffix(FstUnFinishedNodes* nodes, FstSlice bs, Output
   if (fstSliceIsEmpty(s)) {
     return;
   }
-  size_t                    sz = taosArrayGetSize(nodes->stack) - 1;
+  int32_t                   sz = taosArrayGetSize(nodes->stack) - 1;
   FstBuilderNodeUnfinished* un = taosArrayGet(nodes->stack, sz);
   assert(un->last == NULL);
 
@@ -130,11 +130,11 @@ void fstUnFinishedNodesAddSuffix(FstUnFinishedNodes* nodes, FstSlice bs, Output
 uint64_t fstUnFinishedNodesFindCommPrefix(FstUnFinishedNodes* node, FstSlice bs) {
   FstSlice* s = &bs;
 
-  size_t   ssz = taosArrayGetSize(node->stack);  // stack size
+  int32_t  ssz = taosArrayGetSize(node->stack);  // stack size
   uint64_t count = 0;
   int32_t  lsz;  // data len
   uint8_t* data = fstSliceData(s, &lsz);
-  for (size_t i = 0; i < ssz && i < lsz; i++) {
+  for (int32_t i = 0; i < ssz && i < lsz; i++) {
     FstBuilderNodeUnfinished* un = taosArrayGet(node->stack, i);
     if (un->last->inp == data[i]) {
       count++;
@@ -147,8 +147,8 @@ uint64_t fstUnFinishedNodesFindCommPrefix(FstUnFinishedNodes* node, FstSlice bs)
 uint64_t fstUnFinishedNodesFindCommPrefixAndSetOutput(FstUnFinishedNodes* node, FstSlice bs, Output in, Output* out) {
   FstSlice* s = &bs;
 
-  size_t lsz = (size_t)(s->end - s->start + 1);  // data len
-  size_t ssz = taosArrayGetSize(node->stack);    // stack size
+  int32_t lsz = (size_t)(s->end - s->start + 1);  // data len
+  int32_t ssz = taosArrayGetSize(node->stack);    // stack size
   *out = in;
   uint64_t i = 0;
   for (i = 0; i < lsz && i < ssz; i++) {
@@ -245,7 +245,7 @@ void fstStateCompileForOneTrans(FstCountingWriter* w, CompiledAddr addr, FstTran
   return;
 }
 void fstStateCompileForAnyTrans(FstCountingWriter* w, CompiledAddr addr, FstBuilderNode* node) {
-  size_t sz = taosArrayGetSize(node->trans);
+  int32_t sz = taosArrayGetSize(node->trans);
   assert(sz <= 256);
 
   uint8_t tSize = 0;
@@ -253,7 +253,7 @@ void fstStateCompileForAnyTrans(FstCountingWriter* w, CompiledAddr addr, FstBuil
 
   // finalOutput.is_zero()
   bool anyOuts = (node->finalOutput != 0);
-  for (size_t i = 0; i < sz; i++) {
+  for (int32_t i = 0; i < sz; i++) {
     FstTransition* t = taosArrayGet(node->trans, i);
     tSize = TMAX(tSize, packDeltaSize(addr, t->addr));
     oSize = TMAX(oSize, packSize(t->out));
@@ -301,7 +301,7 @@ void fstStateCompileForAnyTrans(FstCountingWriter* w, CompiledAddr addr, FstBuil
     /// for (uint8_t i = 0; i < 256; i++) {
     //  index[i] = 255;
     ///}
-    for (size_t i = 0; i < sz; i++) {
+    for (int32_t i = 0; i < sz; i++) {
       FstTransition* t = taosArrayGet(node->trans, i);
       index[t->inp] = i;
       // fstPackDeltaIn(w, addr, t->addr, tSize);
@@ -731,7 +731,7 @@ bool fstNodeFindInput(FstNode* node, uint8_t b, uint64_t* res) {
 }
 
 bool fstNodeCompile(FstNode* node, void* w, CompiledAddr lastAddr, CompiledAddr addr, FstBuilderNode* builderNode) {
-  size_t sz = taosArrayGetSize(builderNode->trans);
+  int32_t sz = taosArrayGetSize(builderNode->trans);
   assert(sz < 256);
   if (sz == 0 && builderNode->isFinal && builderNode->finalOutput == 0) {
     return true;
@@ -959,8 +959,8 @@ void fstBuilderNodeUnfinishedAddOutputPrefix(FstBuilderNodeUnfinished* unNode, O
   if (FST_BUILDER_NODE_IS_FINAL(unNode->node)) {
     unNode->node->finalOutput += out;
   }
-  size_t sz = taosArrayGetSize(unNode->node->trans);
-  for (size_t i = 0; i < sz; i++) {
+  int32_t sz = taosArrayGetSize(unNode->node->trans);
+  for (int32_t i = 0; i < sz; i++) {
     FstTransition* trn = taosArrayGet(unNode->node->trans, i);
     trn->out += out;
   }
@@ -1077,7 +1077,7 @@ bool fstGet(Fst* fst, FstSlice* b, Output* out) {
     tOut = tOut + FST_NODE_FINAL_OUTPUT(root);
   }
 
-  for (size_t i = 0; i < taosArrayGetSize(nodes); i++) {
+  for (int32_t i = 0; i < taosArrayGetSize(nodes); i++) {
     FstNode** node = (FstNode**)taosArrayGet(nodes, i);
     fstNodeDestroy(*node);
   }
@@ -1352,7 +1352,7 @@ StreamWithStateResult* streamWithStateNextWith(StreamWithState* sws, StreamCallb
     StreamState s2 = {.node = nextNode, .trans = 0, .out = {.null = false, .out = out}, .autState = nextState};
     taosArrayPush(sws->stack, &s2);
 
-    size_t   isz = taosArrayGetSize(sws->inp);
+    int32_t  isz = taosArrayGetSize(sws->inp);
     uint8_t* buf = (uint8_t*)taosMemoryMalloc(isz * sizeof(uint8_t));
     for (uint32_t i = 0; i < isz; i++) {
       buf[i] = *(uint8_t*)taosArrayGet(sws->inp, i);

source/libs/index/src/indexTfile.c

@@ -116,7 +116,7 @@ TFileCache* tfileCacheCreate(const char* path) {
       continue;
     }
     TFileHeader* header = &reader->header;
-    ICacheKey    key = {.suid = header->suid, .colName = header->colName, .nColName = strlen(header->colName)};
+    ICacheKey    key = {.suid = header->suid, .colName = header->colName, .nColName = (int32_t)strlen(header->colName)};
 
     char    buf[128] = {0};
     int32_t sz = indexSerialCacheKey(&key, buf);
@@ -230,7 +230,7 @@ static int32_t tfSearchTerm(void* reader, SIndexTerm* tem, SIdxTempResult* tr) {
     indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, found table info in tindex, time cost: %" PRIu64 "us",
               tem->suid, tem->colName, tem->colVal, cost);
 
-    ret = tfileReaderLoadTableIds((TFileReader*)reader, offset, tr->total);
+    ret = tfileReaderLoadTableIds((TFileReader*)reader, (int32_t)offset, tr->total);
     cost = taosGetTimestampUs() - et;
     indexInfo("index: %" PRIu64 ", col: %s, colVal: %s, load all table info, time cost: %" PRIu64 "us", tem->suid,
               tem->colName, tem->colVal, cost);
@@ -890,7 +890,7 @@ static int tfileWriteFooter(TFileWriter* write) {
   char  buf[sizeof(tfileMagicNumber) + 1] = {0};
   void* pBuf = (void*)buf;
   taosEncodeFixedU64((void**)(void*)&pBuf, tfileMagicNumber);
-  int nwrite = write->ctx->write(write->ctx, buf, strlen(buf));
+  int nwrite = write->ctx->write(write->ctx, buf, (int32_t)strlen(buf));
 
   indexInfo("tfile write footer size: %d", write->ctx->size(write->ctx));
   assert(nwrite == sizeof(tfileMagicNumber));

source/libs/index/src/indexUtil.c

@@ -37,14 +37,14 @@ static int iBinarySearch(SArray *arr, int s, int e, uint64_t k) {
 }
 
 void iIntersection(SArray *inters, SArray *final) {
-  int32_t sz = taosArrayGetSize(inters);
+  int32_t sz = (int32_t)taosArrayGetSize(inters);
   if (sz <= 0) {
     return;
   }
   MergeIndex *mi = taosMemoryCalloc(sz, sizeof(MergeIndex));
   for (int i = 0; i < sz; i++) {
     SArray *t = taosArrayGetP(inters, i);
-    mi[i].len = taosArrayGetSize(t);
+    mi[i].len = (int32_t)taosArrayGetSize(t);
     mi[i].idx = 0;
   }
 
@@ -70,7 +70,7 @@ void iIntersection(SArray *inters, SArray *final) {
   taosMemoryFreeClear(mi);
 }
 void iUnion(SArray *inters, SArray *final) {
-  int32_t sz = taosArrayGetSize(inters);
+  int32_t sz = (int32_t)taosArrayGetSize(inters);
   if (sz <= 0) {
     return;
   }
@@ -82,7 +82,7 @@ void iUnion(SArray *inters, SArray *final) {
   MergeIndex *mi = taosMemoryCalloc(sz, sizeof(MergeIndex));
   for (int i = 0; i < sz; i++) {
     SArray *t = taosArrayGetP(inters, i);
-    mi[i].len = taosArrayGetSize(t);
+    mi[i].len = (int32_t)taosArrayGetSize(t);
     mi[i].idx = 0;
   }
   while (1) {
@@ -117,8 +117,8 @@ void iUnion(SArray *inters, SArray *final) {
 }
 
 void iExcept(SArray *total, SArray *except) {
-  int32_t tsz = taosArrayGetSize(total);
-  int32_t esz = taosArrayGetSize(except);
+  int32_t tsz = (int32_t)taosArrayGetSize(total);
+  int32_t esz = (int32_t)taosArrayGetSize(except);
   if (esz == 0 || tsz == 0) {
     return;
   }
@@ -141,7 +141,10 @@ int uidCompare(const void *a, const void *b) {
   // add more version compare
   uint64_t u1 = *(uint64_t *)a;
   uint64_t u2 = *(uint64_t *)b;
-  return u1 - u2;
+  if (u1 == u2) {
+    return 0;
+  }
+  return u1 < u2 ? -1 : 1;
 }
 int verdataCompare(const void *a, const void *b) {
   SIdxVerdata *va = (SIdxVerdata *)a;

source/libs/index/test/CMakeLists.txt

@@ -92,7 +92,19 @@ target_link_libraries (jsonUT
   index
 )
 
-#add_test(
-#  NAME index_test
-#  COMMAND indexTest
-#)
+add_test(
+  NAME idxtest
+  COMMAND indexTest
+)
+add_test(
+  NAME idxJsonUT
+  COMMAND jsonUT 
+)
+add_test(
+  NAME idxUtilUT 
+  COMMAND UtilUT 
+)
+add_test(
+  NAME idxFstUT 
+  COMMAND fstUT 
+)

source/libs/index/test/fstTest.cc

@@ -48,7 +48,7 @@ class FstWriter {
 
 class FstReadMemory {
  public:
-  FstReadMemory(size_t size, const std::string& fileName = "/tmp/tindex.tindex") {
+  FstReadMemory(int32_t size, const std::string& fileName = "/tmp/tindex.tindex") {
     _wc = writerCtxCreate(TFile, fileName.c_str(), true, 64 * 1024);
     _w = fstCountingWriterCreate(_wc);
     _size = size;
@@ -152,7 +152,7 @@ class FstReadMemory {
   Fst*               _fst;
   FstSlice           _s;
   WriterCtx*         _wc;
-  size_t             _size;
+  int32_t            _size;
 };
 
 #define L 100

source/libs/index/test/indexTests.cc

@@ -714,7 +714,7 @@ class IndexObj {
     return numOfTable;
   }
   int ReadMultiMillonData(const std::string& colName, const std::string& colVal = "Hello world",
-                          size_t numOfTable = 100 * 10000) {
+                          size_t numOfTable = 100) {
     std::string tColVal = colVal;
 
     int colValSize = tColVal.size();
@@ -896,7 +896,7 @@ TEST_F(IndexEnv2, testIndex_TrigeFlush) {
     // r
     std::cout << "failed to init" << std::endl;
   }
-  int numOfTable = 100 * 10000;
+  int numOfTable = 100 * 100;
   index->WriteMillonData("tag1", "Hello Wolrd", numOfTable);
   int target = index->SearchOne("tag1", "Hello Wolrd");
   std::cout << "Get Index: " << target << std::endl;
@@ -910,8 +910,8 @@ static void single_write_and_search(IndexObj* idx) {
 static void multi_write_and_search(IndexObj* idx) {
   int target = idx->SearchOne("tag1", "Hello");
   target = idx->SearchOne("tag2", "Test");
-  idx->WriteMultiMillonData("tag1", "hello world test", 100 * 10000);
-  idx->WriteMultiMillonData("tag2", "world test nothing", 100 * 10000);
+  idx->WriteMultiMillonData("tag1", "hello world test", 100 * 100);
+  idx->WriteMultiMillonData("tag2", "world test nothing", 100 * 10);
 }
 TEST_F(IndexEnv2, testIndex_serarch_cache_and_tfile) {
   std::string path = "/tmp/cache_and_tfile";
@@ -920,8 +920,8 @@ TEST_F(IndexEnv2, testIndex_serarch_cache_and_tfile) {
   }
   index->PutOne("tag1", "Hello");
   index->PutOne("tag2", "Test");
-  index->WriteMultiMillonData("tag1", "Hello", 100 * 10000);
-  index->WriteMultiMillonData("tag2", "Test", 100 * 10000);
+  index->WriteMultiMillonData("tag1", "Hello", 100 * 100);
+  index->WriteMultiMillonData("tag2", "Test", 100 * 100);
   std::thread threads[NUM_OF_THREAD];
 
   for (int i = 0; i < NUM_OF_THREAD; i++) {
@@ -949,49 +949,49 @@ TEST_F(IndexEnv2, testIndex_MultiWrite_and_MultiRead) {
   }
 }
 
-TEST_F(IndexEnv2, testIndex_restart) {
-  std::string path = "/tmp/cache_and_tfile";
-  if (index->Init(path) != 0) {
-  }
-  index->SearchOneTarget("tag1", "Hello", 10);
-  index->SearchOneTarget("tag2", "Test", 10);
-}
-TEST_F(IndexEnv2, testIndex_restart1) {
-  std::string path = "/tmp/cache_and_tfile";
-  if (index->Init(path) != 0) {
-  }
-  index->ReadMultiMillonData("tag1", "coding");
-  index->SearchOneTarget("tag1", "Hello", 10);
-  index->SearchOneTarget("tag2", "Test", 10);
-}
+// TEST_F(IndexEnv2, testIndex_restart) {
+//  std::string path = "/tmp/cache_and_tfile";
+//  if (index->Init(path) != 0) {
+//  }
+//  index->SearchOneTarget("tag1", "Hello", 10);
+//  index->SearchOneTarget("tag2", "Test", 10);
+//}
+// TEST_F(IndexEnv2, testIndex_restart1) {
+//  std::string path = "/tmp/cache_and_tfile";
+//  if (index->Init(path) != 0) {
+//  }
+//  index->ReadMultiMillonData("tag1", "coding");
+//  index->SearchOneTarget("tag1", "Hello", 10);
+//  index->SearchOneTarget("tag2", "Test", 10);
+//}
 
-TEST_F(IndexEnv2, testIndex_read_performance) {
-  std::string path = "/tmp/cache_and_tfile";
-  if (index->Init(path) != 0) {
-  }
-  index->PutOneTarge("tag1", "Hello", 12);
-  index->PutOneTarge("tag1", "Hello", 15);
-  index->ReadMultiMillonData("tag1", "Hello");
-  std::cout << "reader sz: " << index->SearchOne("tag1", "Hello") << std::endl;
-  assert(3 == index->SearchOne("tag1", "Hello"));
-}
-TEST_F(IndexEnv2, testIndexMultiTag) {
-  std::string path = "/tmp/multi_tag";
-  if (index->Init(path) != 0) {
-  }
-  int64_t st = taosGetTimestampUs();
-  int32_t num = 1000 * 10000;
-  index->WriteMultiMillonData("tag1", "xxxxxxxxxxxxxxx", num);
-  std::cout << "numOfRow: " << num << "\ttime cost:" << taosGetTimestampUs() - st << std::endl;
-  // index->WriteMultiMillonData("tag2", "xxxxxxxxxxxxxxxxxxxxxxxxx", 100 * 10000);
-}
+// TEST_F(IndexEnv2, testIndex_read_performance) {
+//  std::string path = "/tmp/cache_and_tfile";
+//  if (index->Init(path) != 0) {
+//  }
+//  index->PutOneTarge("tag1", "Hello", 12);
+//  index->PutOneTarge("tag1", "Hello", 15);
+//  index->ReadMultiMillonData("tag1", "Hello");
+//  std::cout << "reader sz: " << index->SearchOne("tag1", "Hello") << std::endl;
+//  assert(3 == index->SearchOne("tag1", "Hello"));
+//}
+// TEST_F(IndexEnv2, testIndexMultiTag) {
+//  std::string path = "/tmp/multi_tag";
+//  if (index->Init(path) != 0) {
+//  }
+//  int64_t st = taosGetTimestampUs();
+//  int32_t num = 1000 * 10000;
+//  index->WriteMultiMillonData("tag1", "xxxxxxxxxxxxxxx", num);
+//  std::cout << "numOfRow: " << num << "\ttime cost:" << taosGetTimestampUs() - st << std::endl;
+//  // index->WriteMultiMillonData("tag2", "xxxxxxxxxxxxxxxxxxxxxxxxx", 100 * 10000);
+//}
 TEST_F(IndexEnv2, testLongComVal1) {
   std::string path = "/tmp/long_colVal";
   if (index->Init(path) != 0) {
   }
   // gen colVal by randstr
   std::string randstr = "xxxxxxxxxxxxxxxxx";
-  index->WriteMultiMillonData("tag1", randstr, 100 * 10000);
+  index->WriteMultiMillonData("tag1", randstr, 100 * 1000);
 }
 
 TEST_F(IndexEnv2, testLongComVal2) {
@@ -1000,7 +1000,7 @@ TEST_F(IndexEnv2, testLongComVal2) {
   }
   // gen colVal by randstr
   std::string randstr = "abcccc fdadfafdafda";
-  index->WriteMultiMillonData("tag1", randstr, 100 * 10000);
+  index->WriteMultiMillonData("tag1", randstr, 100 * 1000);
 }
 TEST_F(IndexEnv2, testLongComVal3) {
   std::string path = "/tmp/long_colVal";
@@ -1008,7 +1008,7 @@ TEST_F(IndexEnv2, testLongComVal3) {
   }
   // gen colVal by randstr
   std::string randstr = "Yes, coding and coding and coding";
-  index->WriteMultiMillonData("tag1", randstr, 100 * 10000);
+  index->WriteMultiMillonData("tag1", randstr, 100 * 1000);
 }
 TEST_F(IndexEnv2, testLongComVal4) {
   std::string path = "/tmp/long_colVal";
@@ -1016,7 +1016,7 @@ TEST_F(IndexEnv2, testLongComVal4) {
   }
   // gen colVal by randstr
   std::string randstr = "111111 bac fdadfa";
-  index->WriteMultiMillonData("tag1", randstr, 100 * 10000);
+  index->WriteMultiMillonData("tag1", randstr, 100 * 100);
 }
 TEST_F(IndexEnv2, testIndex_read_performance1) {
   std::string path = "/tmp/cache_and_tfile";
@@ -1026,7 +1026,7 @@ TEST_F(IndexEnv2, testIndex_read_performance1) {
   index->PutOneTarge("tag1", "Hello", 15);
   index->ReadMultiMillonData("tag1", "Hello", 1000);
   std::cout << "reader sz: " << index->SearchOne("tag1", "Hello") << std::endl;
-  assert(3 == index->SearchOne("tag1", "Hello"));
+  EXPECT_EQ(2, index->SearchOne("tag1", "Hello"));
 }
 TEST_F(IndexEnv2, testIndex_read_performance2) {
   std::string path = "/tmp/cache_and_tfile";
@@ -1034,9 +1034,9 @@ TEST_F(IndexEnv2, testIndex_read_performance2) {
   }
   index->PutOneTarge("tag1", "Hello", 12);
   index->PutOneTarge("tag1", "Hello", 15);
-  index->ReadMultiMillonData("tag1", "Hello", 1000 * 10);
+  index->ReadMultiMillonData("tag1", "Hello", 1000);
   std::cout << "reader sz: " << index->SearchOne("tag1", "Hello") << std::endl;
-  assert(3 == index->SearchOne("tag1", "Hello"));
+  EXPECT_EQ(2, index->SearchOne("tag1", "Hello"));
 }
 TEST_F(IndexEnv2, testIndex_read_performance3) {
   std::string path = "/tmp/cache_and_tfile";
@@ -1044,9 +1044,9 @@ TEST_F(IndexEnv2, testIndex_read_performance3) {
   }
   index->PutOneTarge("tag1", "Hello", 12);
   index->PutOneTarge("tag1", "Hello", 15);
-  index->ReadMultiMillonData("tag1", "Hello", 1000 * 100);
+  index->ReadMultiMillonData("tag1", "Hello", 1000);
   std::cout << "reader sz: " << index->SearchOne("tag1", "Hello") << std::endl;
-  assert(3 == index->SearchOne("tag1", "Hello"));
+  EXPECT_EQ(2, index->SearchOne("tag1", "Hello"));
 }
 TEST_F(IndexEnv2, testIndex_read_performance4) {
   std::string path = "/tmp/cache_and_tfile";
@@ -1054,9 +1054,9 @@ TEST_F(IndexEnv2, testIndex_read_performance4) {
   }
   index->PutOneTarge("tag10", "Hello", 12);
   index->PutOneTarge("tag12", "Hello", 15);
-  index->ReadMultiMillonData("tag10", "Hello", 1000 * 100);
+  index->ReadMultiMillonData("tag10", "Hello", 1000);
   std::cout << "reader sz: " << index->SearchOne("tag1", "Hello") << std::endl;
-  assert(3 == index->SearchOne("tag10", "Hello"));
+  EXPECT_EQ(1, index->SearchOne("tag10", "Hello"));
 }
 TEST_F(IndexEnv2, testIndex_cache_del) {
   std::string path = "/tmp/cache_and_tfile";
@@ -1108,7 +1108,7 @@ TEST_F(IndexEnv2, testIndex_del) {
   index->Del("tag10", "Hello", 11);
   EXPECT_EQ(98, index->SearchOne("tag10", "Hello"));
 
-  index->WriteMultiMillonData("tag10", "xxxxxxxxxxxxxx", 100 * 10000);
+  index->WriteMultiMillonData("tag10", "xxxxxxxxxxxxxx", 100 * 100);
   index->Del("tag10", "Hello", 17);
   EXPECT_EQ(97, index->SearchOne("tag10", "Hello"));
 }

source/libs/index/test/jsonUT.cc

@@ -154,7 +154,7 @@ TEST_F(JsonEnv, testWriteMillonData) {
 
     SIndexMultiTerm* terms = indexMultiTermCreate();
     indexMultiTermAdd(terms, term);
-    for (size_t i = 0; i < 100; i++) {
+    for (size_t i = 0; i < 10; i++) {
       tIndexJsonPut(index, terms, i);
     }
     indexMultiTermDestroy(terms);
@@ -162,14 +162,14 @@ TEST_F(JsonEnv, testWriteMillonData) {
   {
     std::string colName("voltagefdadfa");
     std::string colVal("abxxxxxxxxxxxx");
-    for (int i = 0; i < 1000; i++) {
+    for (int i = 0; i < 10; i++) {
       colVal[i % colVal.size()] = '0' + i % 128;
       SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(),
                                          colVal.c_str(), colVal.size());
 
       SIndexMultiTerm* terms = indexMultiTermCreate();
       indexMultiTermAdd(terms, term);
-      for (size_t i = 0; i < 1000; i++) {
+      for (size_t i = 0; i < 100; i++) {
         tIndexJsonPut(index, terms, i);
       }
       indexMultiTermDestroy(terms);
@@ -199,7 +199,7 @@ TEST_F(JsonEnv, testWriteMillonData) {
     SArray* result = taosArrayInit(1, sizeof(uint64_t));
     indexMultiTermQueryAdd(mq, q, QUERY_TERM);
     tIndexJsonSearch(index, mq, result);
-    assert(100 == taosArrayGetSize(result));
+    EXPECT_EQ(10, taosArrayGetSize(result));
     indexMultiTermQueryDestroy(mq);
   }
   {
@@ -229,7 +229,7 @@ TEST_F(JsonEnv, testWriteMillonData) {
         SArray* result = taosArrayInit(1, sizeof(uint64_t));
         indexMultiTermQueryAdd(mq, q, QUERY_GREATER_EQUAL);
         tIndexJsonSearch(index, mq, result);
-        assert(100 == taosArrayGetSize(result));
+        EXPECT_EQ(10, taosArrayGetSize(result));
         indexMultiTermQueryDestroy(mq);
       }
     }
@@ -385,7 +385,7 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) {
 
     SIndexMultiTerm* terms = indexMultiTermCreate();
     indexMultiTermAdd(terms, term);
-    for (size_t i = 0; i < 100000; i++) {
+    for (size_t i = 0; i < 1000; i++) {
       tIndexJsonPut(index, terms, i);
     }
     indexMultiTermDestroy(terms);
@@ -523,7 +523,7 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT2) {
   {
     int         val = 10;
     std::string colName("test1");
-    for (int i = 0; i < 10000; i++) {
+    for (int i = 0; i < 1000; i++) {
       val += 1;
       WriteData(index, colName, TSDB_DATA_TYPE_INT, &val, sizeof(val), i);
     }
@@ -532,7 +532,7 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT2) {
     int         val = 10;
     std::string colName("test2xxx");
     std::string colVal("xxxxxxxxxxxxxxx");
-    for (int i = 0; i < 100000; i++) {
+    for (int i = 0; i < 1000; i++) {
       val += 1;
       WriteData(index, colName, TSDB_DATA_TYPE_BINARY, (void*)(colVal.c_str()), colVal.size(), i);
     }
@@ -542,14 +542,14 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT2) {
     std::string colName("test1");
     int         val = 9;
     Search(index, colName, TSDB_DATA_TYPE_INT, &val, sizeof(val), QUERY_GREATER_EQUAL, &res);
-    EXPECT_EQ(10000, taosArrayGetSize(res));
+    EXPECT_EQ(1000, taosArrayGetSize(res));
   }
   {
     SArray*     res = NULL;
     std::string colName("test2xxx");
     std::string colVal("xxxxxxxxxxxxxxx");
     Search(index, colName, TSDB_DATA_TYPE_BINARY, (void*)(colVal.c_str()), colVal.size(), QUERY_TERM, &res);
-    EXPECT_EQ(100000, taosArrayGetSize(res));
+    EXPECT_EQ(1000, taosArrayGetSize(res));
   }
 }
 TEST_F(JsonEnv, testWriteJsonTfileAndCache_FLOAT) {

source/libs/parser/src/parInsert.c

@@ -1110,7 +1110,9 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
       NEXT_TOKEN(pCxt->pSql, sToken);
       autoCreateTbl = true;
     } else {
-      CHECK_CODE(getTableMeta(pCxt, &name, tbFName));
+      char dbFName[TSDB_DB_FNAME_LEN];
+      tNameGetFullDbName(&name, dbFName);
+      CHECK_CODE(getTableMeta(pCxt, &name, dbFName));
     }
 
     STableDataBlocks* dataBuf = NULL;

source/libs/parser/src/parTranslater.c

@@ -53,6 +53,8 @@ static bool afterGroupBy(ESqlClause clause) { return clause > SQL_CLAUSE_GROUP_B
 
 static bool beforeHaving(ESqlClause clause) { return clause < SQL_CLAUSE_HAVING; }
 
+static bool afterHaving(ESqlClause clause) { return clause > SQL_CLAUSE_HAVING; }
+
 static int32_t addNamespace(STranslateContext* pCxt, void* pTable) {
   size_t currTotalLevel = taosArrayGetSize(pCxt->pNsLevel);
   if (currTotalLevel > pCxt->currLevel) {
@@ -276,6 +278,10 @@ static bool isScanPseudoColumnFunc(const SNode* pNode) {
   return (QUERY_NODE_FUNCTION == nodeType(pNode) && fmIsScanPseudoColumnFunc(((SFunctionNode*)pNode)->funcId));
 }
 
+static bool isNonstandardSQLFunc(const SNode* pNode) {
+  return (QUERY_NODE_FUNCTION == nodeType(pNode) && fmIsNonstandardSQLFunc(((SFunctionNode*)pNode)->funcId));
+}
+
 static bool isDistinctOrderBy(STranslateContext* pCxt) {
   return (SQL_CLAUSE_ORDER_BY == pCxt->currClause && pCxt->pCurrStmt->isDistinct);
 }
@@ -433,6 +439,7 @@ static EDealRes translateColumnWithoutPrefix(STranslateContext* pCxt, SColumnNod
   SArray* pTables = taosArrayGetP(pCxt->pNsLevel, pCxt->currLevel);
   size_t  nums = taosArrayGetSize(pTables);
   bool    found = false;
+  bool    isInternalPk = isInternalPrimaryKey(pCol);
   for (size_t i = 0; i < nums; ++i) {
     STableNode* pTable = taosArrayGetP(pTables, i);
     if (findAndSetColumn(pCol, pTable)) {
@@ -440,10 +447,13 @@ static EDealRes translateColumnWithoutPrefix(STranslateContext* pCxt, SColumnNod
         return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_AMBIGUOUS_COLUMN, pCol->colName);
       }
       found = true;
+      if (isInternalPk) {
+        break;
+      }
     }
   }
   if (!found) {
-    if (isInternalPrimaryKey(pCol)) {
+    if (isInternalPk) {
       return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_INVALID_INTERNAL_PK);
     } else {
       return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_INVALID_COLUMN, pCol->colName);
@@ -703,10 +713,13 @@ static EDealRes translateOperator(STranslateContext* pCxt, SOperatorNode* pOp) {
   return DEAL_RES_CONTINUE;
 }
 
-static EDealRes haveAggFunction(SNode* pNode, void* pContext) {
+static EDealRes haveAggOrNonstdFunction(SNode* pNode, void* pContext) {
   if (isAggFunc(pNode)) {
     *((bool*)pContext) = true;
     return DEAL_RES_END;
+  } else if (isNonstandardSQLFunc(pNode)) {
+    *((bool*)pContext) = true;
+    return DEAL_RES_END;
   }
   return DEAL_RES_CONTINUE;
 }
@@ -743,6 +756,12 @@ static int32_t rewriteCountStar(STranslateContext* pCxt, SFunctionNode* pCount)
   return code;
 }
 
+static bool hasInvalidFuncNesting(SNodeList* pParameterList) {
+  bool hasInvalidFunc = false;
+  nodesWalkExprs(pParameterList, haveAggOrNonstdFunction, &hasInvalidFunc);
+  return hasInvalidFunc;
+}
+
 static EDealRes translateFunction(STranslateContext* pCxt, SFunctionNode* pFunc) {
   SFmGetFuncInfoParam param = {.pCtg = pCxt->pParseCxt->pCatalog,
                                .pRpc = pCxt->pParseCxt->pTransporter,
@@ -754,11 +773,12 @@ static EDealRes translateFunction(STranslateContext* pCxt, SFunctionNode* pFunc)
     if (beforeHaving(pCxt->currClause)) {
       return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_ILLEGAL_USE_AGG_FUNCTION);
     }
-    bool haveAggFunc = false;
-    nodesWalkExprs(pFunc->pParameterList, haveAggFunction, &haveAggFunc);
-    if (haveAggFunc) {
+    if (hasInvalidFuncNesting(pFunc->pParameterList)) {
       return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_AGG_FUNC_NESTING);
     }
+    if (pCxt->pCurrStmt->hasNonstdSQLFunc) {
+      return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_NOT_ALLOWED_FUNC);
+    }
 
     pCxt->pCurrStmt->hasAggFuncs = true;
     pCxt->pCurrStmt->isTimeOrderQuery = false;
@@ -784,6 +804,15 @@ static EDealRes translateFunction(STranslateContext* pCxt, SFunctionNode* pFunc)
       }
     }
   }
+  if (TSDB_CODE_SUCCESS == pCxt->errCode && fmIsNonstandardSQLFunc(pFunc->funcId)) {
+    if (SQL_CLAUSE_SELECT != pCxt->currClause || pCxt->pCurrStmt->hasNonstdSQLFunc || pCxt->pCurrStmt->hasAggFuncs) {
+      return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_NOT_ALLOWED_FUNC);
+    }
+    if (hasInvalidFuncNesting(pFunc->pParameterList)) {
+      return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_AGG_FUNC_NESTING);
+    }
+    pCxt->pCurrStmt->hasNonstdSQLFunc = true;
+  }
   return TSDB_CODE_SUCCESS == pCxt->errCode ? DEAL_RES_CONTINUE : DEAL_RES_ERROR;
 }
 
@@ -952,6 +981,7 @@ typedef struct CheckAggColCoexistCxt {
   STranslateContext* pTranslateCxt;
   bool               existAggFunc;
   bool               existCol;
+  bool               existNonstdFunc;
   int32_t            selectFuncNum;
 } CheckAggColCoexistCxt;
 
@@ -962,6 +992,10 @@ static EDealRes doCheckAggColCoexist(SNode* pNode, void* pContext) {
     pCxt->existAggFunc = true;
     return DEAL_RES_IGNORE_CHILD;
   }
+  if (isNonstandardSQLFunc(pNode)) {
+    pCxt->existNonstdFunc = true;
+    return DEAL_RES_IGNORE_CHILD;
+  }
   if (isScanPseudoColumnFunc(pNode) || QUERY_NODE_COLUMN == nodeType(pNode)) {
     pCxt->existCol = true;
   }
@@ -972,16 +1006,21 @@ static int32_t checkAggColCoexist(STranslateContext* pCxt, SSelectStmt* pSelect)
   if (NULL != pSelect->pGroupByList) {
     return TSDB_CODE_SUCCESS;
   }
-  CheckAggColCoexistCxt cxt = {.pTranslateCxt = pCxt, .existAggFunc = false, .existCol = false};
+  CheckAggColCoexistCxt cxt = {
+      .pTranslateCxt = pCxt, .existAggFunc = false, .existCol = false, .existNonstdFunc = false};
   nodesWalkExprs(pSelect->pProjectionList, doCheckAggColCoexist, &cxt);
   if (!pSelect->isDistinct) {
     nodesWalkExprs(pSelect->pOrderByList, doCheckAggColCoexist, &cxt);
   }
   if (1 == cxt.selectFuncNum) {
     return rewriteColsToSelectValFunc(pCxt, pSelect);
-  } else if ((cxt.selectFuncNum > 1 || cxt.existAggFunc || NULL != pSelect->pWindow) && cxt.existCol) {
+  }
+  if ((cxt.selectFuncNum > 1 || cxt.existAggFunc || NULL != pSelect->pWindow) && cxt.existCol) {
     return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_NOT_SINGLE_GROUP);
   }
+  if (cxt.existNonstdFunc && cxt.existCol) {
+    return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_NOT_ALLOWED_FUNC);
+  }
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/parser/src/parUtil.c

@@ -164,6 +164,9 @@ static char* getSyntaxErrFormat(int32_t errCode) {
       return "Invalid function name";
     case TSDB_CODE_PAR_COMMENT_TOO_LONG:
       return "Comment too long";
+    case TSDB_CODE_PAR_NOT_ALLOWED_FUNC:
+      return "Some functions are allowed only in the SELECT list of a query. "
+             "And, cannot be mixed with other non scalar functions or columns.";
     case TSDB_CODE_OUT_OF_MEMORY:
       return "Out of memory";
     default:

source/libs/parser/test/parSelectTest.cpp

@@ -121,6 +121,26 @@ TEST_F(ParserSelectTest, selectFunc) {
   run("SELECT MAX(c1), c2 FROM t1 STATE_WINDOW(c3)");
 }
 
+TEST_F(ParserSelectTest, nonstdFunc) {
+  useDb("root", "test");
+
+  run("SELECT DIFF(c1) FROM t1");
+
+  // run("SELECT DIFF(c1) FROM t1 INTERVAL(10s)");
+}
+
+TEST_F(ParserSelectTest, nonstdFuncSemanticCheck) {
+  useDb("root", "test");
+
+  run("SELECT DIFF(c1), c2 FROM t1", TSDB_CODE_PAR_NOT_ALLOWED_FUNC, PARSER_STAGE_TRANSLATE);
+
+  run("SELECT DIFF(c1), tbname FROM t1", TSDB_CODE_PAR_NOT_ALLOWED_FUNC, PARSER_STAGE_TRANSLATE);
+
+  run("SELECT DIFF(c1), count(*) FROM t1", TSDB_CODE_PAR_NOT_ALLOWED_FUNC, PARSER_STAGE_TRANSLATE);
+
+  run("SELECT DIFF(c1), CSUM(c1) FROM t1", TSDB_CODE_PAR_NOT_ALLOWED_FUNC, PARSER_STAGE_TRANSLATE);
+}
+
 TEST_F(ParserSelectTest, clause) {
   useDb("root", "test");
 

source/libs/planner/src/planSpliter.c

@@ -24,7 +24,7 @@
 #define SPLIT_FLAG_TEST_MASK(val, mask) (((val) & (mask)) != 0)
 
 typedef struct SSplitContext {
-  int32_t queryId;
+  uint64_t queryId;
   int32_t  groupId;
   bool     split;
 } SSplitContext;

source/libs/qworker/src/qworkerMsg.c

@@ -287,7 +287,7 @@ int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SRpcHandleInfo *pConn) {
   req->taskId = htobe64(tId);
   req->refId = htobe64(rId);
 
-  SRpcMsg pMsg = {
+  SRpcMsg brokenMsg = {
       .msgType = TDMT_VND_DROP_TASK,
       .pCont = req,
       .contLen = sizeof(STaskDropReq),
@@ -295,7 +295,7 @@ int32_t qwRegisterQueryBrokenLinkArg(QW_FPARAMS_DEF, SRpcHandleInfo *pConn) {
       .info = *pConn,
   };
 
-  tmsgRegisterBrokenLinkArg(&pMsg);
+  tmsgRegisterBrokenLinkArg(&brokenMsg);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -321,7 +321,7 @@ int32_t qwRegisterHbBrokenLinkArg(SQWorker *mgmt, uint64_t sId, SRpcHandleInfo *
     QW_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
   }
 
-  SRpcMsg pMsg = {
+  SRpcMsg brokenMsg = {
       .msgType = TDMT_VND_QUERY_HEARTBEAT,
       .pCont = msg,
       .contLen = msgSize,
@@ -329,7 +329,7 @@ int32_t qwRegisterHbBrokenLinkArg(SQWorker *mgmt, uint64_t sId, SRpcHandleInfo *
       .info = *pConn,
   };
 
-  tmsgRegisterBrokenLinkArg(&pMsg);
+  tmsgRegisterBrokenLinkArg(&brokenMsg);
 
   return TSDB_CODE_SUCCESS;
 }