Merge branch '3.0' into feature/TD-13066-3.0

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/common/tmsg.h

@@ -1648,6 +1648,15 @@ typedef struct {
 int32_t tSerializeSMDropTopicReq(void* buf, int32_t bufLen, SMDropTopicReq* pReq);
 int32_t tDeserializeSMDropTopicReq(void* buf, int32_t bufLen, SMDropTopicReq* pReq);
 
+typedef struct {
+  char   topic[TSDB_TOPIC_FNAME_LEN];
+  char   cgroup[TSDB_CGROUP_LEN];
+  int8_t igNotExists;
+} SMDropCgroupReq;
+
+int32_t tSerializeSMDropCgroupReq(void* buf, int32_t bufLen, SMDropCgroupReq* pReq);
+int32_t tDeserializeSMDropCgroupReq(void* buf, int32_t bufLen, SMDropCgroupReq* pReq);
+
 typedef struct {
   char    name[TSDB_TABLE_FNAME_LEN];
   int8_t  alterType;

include/dnode/mnode/sdb/sdb.h

@@ -343,7 +343,7 @@ typedef struct SSdb {
   int64_t        maxId[SDB_MAX];
   EKeyType       keyTypes[SDB_MAX];
   SHashObj      *hashObjs[SDB_MAX];
-  SRWLatch    locks[SDB_MAX];
+  TdThreadRwlock locks[SDB_MAX];
   SdbInsertFp    insertFps[SDB_MAX];
   SdbUpdateFp    updateFps[SDB_MAX];
   SdbDeleteFp    deleteFps[SDB_MAX];

include/libs/executor/executor.h

@@ -95,6 +95,15 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo
 int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, struct SSubplan* pPlan,
                         qTaskInfo_t* pTaskInfo, DataSinkHandle* handle, EOPTR_EXEC_MODEL model);
 
+/**
+ *
+ * @param tinfo
+ * @param sversion
+ * @param tversion
+ * @return
+ */
+int32_t qGetQueriedTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* tableName, int32_t* sversion, int32_t* tversion);
+
 /**
  * The main task execution function, including query on both table and multiple tables,
  * which are decided according to the tag or table name query conditions

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/client/src/tmq.c

@@ -1435,7 +1435,7 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t wait_time) {
 
   while (1) {
     tmqHandleAllDelayedTask(tmq);
-    tmqPollImpl(tmq, wait_time);
+    if (tmqPollImpl(tmq, wait_time) < 0) return NULL;
 
     rspObj = tmqHandleAllRsp(tmq, wait_time, false);
     if (rspObj) {

source/common/src/tmsg.c

@@ -2627,6 +2627,35 @@ int32_t tDeserializeSMDropTopicReq(void *buf, int32_t bufLen, SMDropTopicReq *pR
   return 0;
 }
 
+int32_t tSerializeSMDropCgroupReq(void *buf, int32_t bufLen, SMDropCgroupReq *pReq) {
+  SEncoder encoder = {0};
+  tEncoderInit(&encoder, buf, bufLen);
+
+  if (tStartEncode(&encoder) < 0) return -1;
+  if (tEncodeCStr(&encoder, pReq->topic) < 0) return -1;
+  if (tEncodeCStr(&encoder, pReq->cgroup) < 0) return -1;
+  if (tEncodeI8(&encoder, pReq->igNotExists) < 0) return -1;
+  tEndEncode(&encoder);
+
+  int32_t tlen = encoder.pos;
+  tEncoderClear(&encoder);
+  return tlen;
+}
+
+int32_t tDeserializeSMDropCgroupReq(void *buf, int32_t bufLen, SMDropCgroupReq *pReq) {
+  SDecoder decoder = {0};
+  tDecoderInit(&decoder, buf, bufLen);
+
+  if (tStartDecode(&decoder) < 0) return -1;
+  if (tDecodeCStrTo(&decoder, pReq->topic) < 0) return -1;
+  if (tDecodeCStrTo(&decoder, pReq->cgroup) < 0) return -1;
+  if (tDecodeI8(&decoder, &pReq->igNotExists) < 0) return -1;
+  tEndDecode(&decoder);
+
+  tDecoderClear(&decoder);
+  return 0;
+}
+
 int32_t tSerializeSCMCreateTopicReq(void *buf, int32_t bufLen, const SCMCreateTopicReq *pReq) {
   int32_t sqlLen = 0;
   int32_t astLen = 0;

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_dnode/src/dmHandle.c

@@ -19,11 +19,11 @@
 static void dmUpdateDnodeCfg(SDnodeMgmt *pMgmt, SDnodeCfg *pCfg) {
   if (pMgmt->pData->dnodeId == 0 || pMgmt->pData->clusterId == 0) {
     dInfo("set dnodeId:%d clusterId:%" PRId64, pCfg->dnodeId, pCfg->clusterId);
-    taosWLockLatch(&pMgmt->pData->latch);
+    taosThreadRwlockWrlock(&pMgmt->pData->lock);
     pMgmt->pData->dnodeId = pCfg->dnodeId;
     pMgmt->pData->clusterId = pCfg->clusterId;
     dmWriteEps(pMgmt->pData);
-    taosWUnLockLatch(&pMgmt->pData->latch);
+    taosThreadRwlockUnlock(&pMgmt->pData->lock);
   }
 }
 
@@ -50,7 +50,7 @@ static void dmProcessStatusRsp(SDnodeMgmt *pMgmt, SRpcMsg *pRsp) {
 void dmSendStatusReq(SDnodeMgmt *pMgmt) {
   SStatusReq req = {0};
 
-  taosRLockLatch(&pMgmt->pData->latch);
+  taosThreadRwlockRdlock(&pMgmt->pData->lock);
   req.sver = tsVersion;
   req.dnodeVer = pMgmt->pData->dnodeVer;
   req.dnodeId = pMgmt->pData->dnodeId;
@@ -69,7 +69,7 @@ void dmSendStatusReq(SDnodeMgmt *pMgmt) {
   memcpy(req.clusterCfg.timezone, tsTimezoneStr, TD_TIMEZONE_LEN);
   memcpy(req.clusterCfg.locale, tsLocale, TD_LOCALE_LEN);
   memcpy(req.clusterCfg.charset, tsCharset, TD_LOCALE_LEN);
-  taosRUnLockLatch(&pMgmt->pData->latch);
+  taosThreadRwlockUnlock(&pMgmt->pData->lock);
 
   SMonVloadInfo vinfo = {0};
   (*pMgmt->getVnodeLoadsFp)(&vinfo);

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

@@ -30,7 +30,6 @@ typedef struct SSnodeMgmt {
   SMsgCb        msgCb;
   const char   *path;
   const char   *name;
-  SRWLatch      latch;
   int8_t        uniqueWorkerInUse;
   SArray       *uniqueWorkers;  // SArray<SMultiWorker*>
   SSingleWorker sharedWorker;
@@ -41,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

@@ -38,7 +38,7 @@ typedef struct SVnodeMgmt {
   SSingleWorker  mgmtWorker;
   SSingleWorker  monitorWorker;
   SHashObj      *hash;
-  SRWLatch      latch;
+  TdThreadRwlock lock;
   SVnodesStat    state;
   STfs          *pTfs;
 } SVnodeMgmt;
@@ -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/vmFile.c

@@ -17,7 +17,7 @@
 #include "vmInt.h"
 
 SVnodeObj **vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes) {
-  taosRLockLatch(&pMgmt->latch);
+  taosThreadRwlockRdlock(&pMgmt->lock);
 
   int32_t     num = 0;
   int32_t     size = taosHashGetSize(pMgmt->hash);
@@ -38,7 +38,7 @@ SVnodeObj **vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes) {
     }
   }
 
-  taosRUnLockLatch(&pMgmt->latch);
+  taosThreadRwlockUnlock(&pMgmt->lock);
   *numOfVnodes = num;
 
   return pVnodes;

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

@@ -20,7 +20,7 @@ void vmGetVnodeLoads(SVnodeMgmt *pMgmt, SMonVloadInfo *pInfo) {
   pInfo->pVloads = taosArrayInit(pMgmt->state.totalVnodes, sizeof(SVnodeLoad));
   if (pInfo->pVloads == NULL) return;
 
-  taosRLockLatch(&pMgmt->latch);
+  taosThreadRwlockRdlock(&pMgmt->lock);
 
   void *pIter = taosHashIterate(pMgmt->hash, NULL);
   while (pIter) {
@@ -34,7 +34,7 @@ void vmGetVnodeLoads(SVnodeMgmt *pMgmt, SMonVloadInfo *pInfo) {
     pIter = taosHashIterate(pMgmt->hash, pIter);
   }
 
-  taosRUnLockLatch(&pMgmt->latch);
+  taosThreadRwlockUnlock(&pMgmt->lock);
 }
 
 void vmGetMonitorInfo(SVnodeMgmt *pMgmt, SMonVmInfo *pInfo) {
@@ -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/vmInt.c

@@ -20,14 +20,14 @@ SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) {
   SVnodeObj *pVnode = NULL;
   int32_t    refCount = 0;
 
-  taosRLockLatch(&pMgmt->latch);
+  taosThreadRwlockRdlock(&pMgmt->lock);
   taosHashGetDup(pMgmt->hash, &vgId, sizeof(int32_t), (void *)&pVnode);
   if (pVnode == NULL) {
     terrno = TSDB_CODE_VND_INVALID_VGROUP_ID;
   } else {
     refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
   }
-  taosRUnLockLatch(&pMgmt->latch);
+  taosThreadRwlockUnlock(&pMgmt->lock);
 
   if (pVnode != NULL) {
     dTrace("vgId:%d, acquire vnode, refCount:%d", pVnode->vgId, refCount);
@@ -39,9 +39,9 @@ SVnodeObj *vmAcquireVnode(SVnodeMgmt *pMgmt, int32_t vgId) {
 void vmReleaseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
   if (pVnode == NULL) return;
 
-  taosRLockLatch(&pMgmt->latch);
+  taosThreadRwlockRdlock(&pMgmt->lock);
   int32_t refCount = atomic_sub_fetch_32(&pVnode->refCount, 1);
-  taosRUnLockLatch(&pMgmt->latch);
+  taosThreadRwlockUnlock(&pMgmt->lock);
   dTrace("vgId:%d, release vnode, refCount:%d", pVnode->vgId, refCount);
 }
 
@@ -70,9 +70,9 @@ int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
     return -1;
   }
 
-  taosWLockLatch(&pMgmt->latch);
+  taosThreadRwlockWrlock(&pMgmt->lock);
   int32_t code = taosHashPut(pMgmt->hash, &pVnode->vgId, sizeof(int32_t), &pVnode, sizeof(SVnodeObj *));
-  taosWUnLockLatch(&pMgmt->latch);
+  taosThreadRwlockUnlock(&pMgmt->lock);
 
   return code;
 }
@@ -80,9 +80,9 @@ int32_t vmOpenVnode(SVnodeMgmt *pMgmt, SWrapperCfg *pCfg, SVnode *pImpl) {
 void vmCloseVnode(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) {
   char path[TSDB_FILENAME_LEN] = {0};
 
-  taosWLockLatch(&pMgmt->latch);
+  taosThreadRwlockWrlock(&pMgmt->lock);
   taosHashRemove(pMgmt->hash, &pVnode->vgId, sizeof(int32_t));
-  taosWUnLockLatch(&pMgmt->latch);
+  taosThreadRwlockUnlock(&pMgmt->lock);
 
   vmReleaseVnode(pMgmt, pVnode);
   while (pVnode->refCount > 0) taosMsleep(10);
@@ -239,6 +239,7 @@ static void vmCleanup(SVnodeMgmt *pMgmt) {
   vmStopWorker(pMgmt);
   vnodeCleanup();
   tfsClose(pMgmt->pTfs);
+  taosThreadRwlockDestroy(&pMgmt->lock);
   taosMemoryFree(pMgmt);
 }
 
@@ -260,7 +261,7 @@ static int32_t vmInit(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
   pMgmt->msgCb.queueFps[MERGE_QUEUE] = (PutToQueueFp)vmPutRpcMsgToMergeQueue;
   pMgmt->msgCb.qsizeFp = (GetQueueSizeFp)vmGetQueueSize;
   pMgmt->msgCb.mgmt = pMgmt;
-  taosInitRWLatch(&pMgmt->latch);
+  taosThreadRwlockInit(&pMgmt->lock, NULL);
 
   SDiskCfg dCfg = {0};
   tstrncpy(dCfg.dir, tsDataDir, TSDB_FILENAME_LEN);
@@ -334,19 +335,23 @@ static int32_t vmRequire(const SMgmtInputOpt *pInput, bool *required) {
 }
 
 static int32_t vmStart(SVnodeMgmt *pMgmt) {
-  taosRLockLatch(&pMgmt->latch);
-
-  void *pIter = taosHashIterate(pMgmt->hash, NULL);
-  while (pIter) {
-    SVnodeObj **ppVnode = pIter;
-    if (ppVnode == NULL || *ppVnode == NULL) continue;
+  int32_t     numOfVnodes = 0;
+  SVnodeObj **pVnodes = vmGetVnodeListFromHash(pMgmt, &numOfVnodes);
 
-    SVnodeObj *pVnode = *ppVnode;
+  for (int32_t i = 0; i < numOfVnodes; ++i) {
+    SVnodeObj *pVnode = pVnodes[i];
     vnodeStart(pVnode->pImpl);
-    pIter = taosHashIterate(pMgmt->hash, pIter);
   }
 
-  taosRUnLockLatch(&pMgmt->latch);
+  for (int32_t i = 0; i < numOfVnodes; ++i) {
+    SVnodeObj *pVnode = pVnodes[i];
+    vmReleaseVnode(pMgmt, pVnode);
+  }
+
+  if (pVnodes != NULL) {
+    taosMemoryFree(pVnodes);
+  }
+
   return 0;
 }
 

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/inc/dmMgmt.h

@@ -70,7 +70,7 @@ typedef struct SMgmtWrapper {
   const char    *name;
   char          *path;
   int32_t        refCount;
-  SRWLatch       latch;
+  TdThreadRwlock lock;
   EDndNodeType   ntype;
   bool           deployed;
   bool           required;

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

@@ -91,7 +91,7 @@ static int32_t dmInitVars(SDnode *pDnode, EDndNodeType rtype) {
     return -1;
   }
 
-  taosInitRWLatch(&pData->latch);
+  taosThreadRwlockInit(&pData->lock, NULL);
   taosThreadMutexInit(&pDnode->mutex, NULL);
   return 0;
 }
@@ -100,6 +100,7 @@ static void dmClearVars(SDnode *pDnode) {
   for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {
     SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
     taosMemoryFreeClear(pWrapper->path);
+    taosThreadRwlockDestroy(&pWrapper->lock);
   }
   if (pDnode->lockfile != NULL) {
     taosUnLockFile(pDnode->lockfile);
@@ -108,7 +109,7 @@ static void dmClearVars(SDnode *pDnode) {
   }
 
   SDnodeData *pData = &pDnode->data;
-  taosWLockLatch(&pData->latch);
+  taosThreadRwlockWrlock(&pData->lock);
   if (pData->dnodeEps != NULL) {
     taosArrayDestroy(pData->dnodeEps);
     pData->dnodeEps = NULL;
@@ -117,8 +118,9 @@ static void dmClearVars(SDnode *pDnode) {
     taosHashCleanup(pData->dnodeHash);
     pData->dnodeHash = NULL;
   }
-  taosWUnLockLatch(&pData->latch);
+  taosThreadRwlockUnlock(&pData->lock);
 
+  taosThreadRwlockDestroy(&pData->lock);
   taosThreadMutexDestroy(&pDnode->mutex);
   memset(&pDnode->mutex, 0, sizeof(pDnode->mutex));
 }
@@ -151,7 +153,7 @@ int32_t dmInitDnode(SDnode *pDnode, EDndNodeType rtype) {
     if (ntype == DNODE) {
       pWrapper->proc.ptype = DND_PROC_SINGLE;
     }
-    taosInitRWLatch(&pWrapper->latch);
+    taosThreadRwlockInit(&pWrapper->lock, NULL);
 
     snprintf(path, sizeof(path), "%s%s%s", tsDataDir, TD_DIRSEP, pWrapper->name);
     pWrapper->path = strdup(path);
@@ -223,7 +225,7 @@ SMgmtWrapper *dmAcquireWrapper(SDnode *pDnode, EDndNodeType ntype) {
   SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
   SMgmtWrapper *pRetWrapper = pWrapper;
 
-  taosRLockLatch(&pWrapper->latch);
+  taosThreadRwlockRdlock(&pWrapper->lock);
   if (pWrapper->deployed) {
     int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1);
     dTrace("node:%s, is acquired, ref:%d", pWrapper->name, refCount);
@@ -231,7 +233,7 @@ SMgmtWrapper *dmAcquireWrapper(SDnode *pDnode, EDndNodeType ntype) {
     terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
     pRetWrapper = NULL;
   }
-  taosRUnLockLatch(&pWrapper->latch);
+  taosThreadRwlockUnlock(&pWrapper->lock);
 
   return pRetWrapper;
 }
@@ -239,7 +241,7 @@ SMgmtWrapper *dmAcquireWrapper(SDnode *pDnode, EDndNodeType ntype) {
 int32_t dmMarkWrapper(SMgmtWrapper *pWrapper) {
   int32_t code = 0;
 
-  taosRLockLatch(&pWrapper->latch);
+  taosThreadRwlockRdlock(&pWrapper->lock);
   if (pWrapper->deployed || (InParentProc(pWrapper) && pWrapper->required)) {
     int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1);
     dTrace("node:%s, is marked, ref:%d", pWrapper->name, refCount);
@@ -247,7 +249,7 @@ int32_t dmMarkWrapper(SMgmtWrapper *pWrapper) {
     terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
     code = -1;
   }
-  taosRUnLockLatch(&pWrapper->latch);
+  taosThreadRwlockUnlock(&pWrapper->lock);
 
   return code;
 }
@@ -255,9 +257,9 @@ int32_t dmMarkWrapper(SMgmtWrapper *pWrapper) {
 void dmReleaseWrapper(SMgmtWrapper *pWrapper) {
   if (pWrapper == NULL) return;
 
-  taosRLockLatch(&pWrapper->latch);
+  taosThreadRwlockRdlock(&pWrapper->lock);
   int32_t refCount = atomic_sub_fetch_32(&pWrapper->refCount, 1);
-  taosRUnLockLatch(&pWrapper->latch);
+  taosThreadRwlockUnlock(&pWrapper->lock);
   dTrace("node:%s, is released, ref:%d", pWrapper->name, refCount);
 }
 

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

@@ -186,12 +186,12 @@ void dmCloseNode(SMgmtWrapper *pWrapper) {
     }
   }
 
-  taosWLockLatch(&pWrapper->latch);
+  taosThreadRwlockWrlock(&pWrapper->lock);
   if (pWrapper->pMgmt != NULL) {
     (*pWrapper->func.closeFp)(pWrapper->pMgmt);
     pWrapper->pMgmt = NULL;
   }
-  taosWUnLockLatch(&pWrapper->latch);
+  taosThreadRwlockUnlock(&pWrapper->lock);
 
   if (!OnlyInSingleProc(pWrapper)) {
     dmCleanupProc(pWrapper);

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/node_util/inc/dmUtil.h

@@ -104,7 +104,7 @@ typedef struct {
   SEpSet         mnodeEps;
   SArray        *dnodeEps;
   SHashObj      *dnodeHash;
-  SRWLatch  latch;
+  TdThreadRwlock lock;
   SMsgCb         msgCb;
 } SDnodeData;
 

source/dnode/mgmt/node_util/src/dmEps.c

@@ -21,7 +21,7 @@ static bool dmIsEpChanged(SDnodeData *pData, int32_t dnodeId, const char *ep);
 static void dmResetEps(SDnodeData *pData, SArray *dnodeEps);
 
 static void dmGetDnodeEp(SDnodeData *pData, int32_t dnodeId, char *pEp, char *pFqdn, uint16_t *pPort) {
-  taosRLockLatch(&pData->latch);
+  taosThreadRwlockRdlock(&pData->lock);
 
   SDnodeEp *pDnodeEp = taosHashGet(pData->dnodeHash, &dnodeId, sizeof(int32_t));
   if (pDnodeEp != NULL) {
@@ -36,7 +36,7 @@ static void dmGetDnodeEp(SDnodeData *pData, int32_t dnodeId, char *pEp, char *pF
     }
   }
 
-  taosRUnLockLatch(&pData->latch);
+  taosThreadRwlockUnlock(&pData->lock);
 }
 
 int32_t dmReadEps(SDnodeData *pData) {
@@ -232,7 +232,7 @@ void dmUpdateEps(SDnodeData *pData, SArray *eps) {
   int32_t numOfEps = taosArrayGetSize(eps);
   if (numOfEps <= 0) return;
 
-  taosWLockLatch(&pData->latch);
+  taosThreadRwlockWrlock(&pData->lock);
 
   int32_t numOfEpsOld = (int32_t)taosArrayGetSize(pData->dnodeEps);
   if (numOfEps != numOfEpsOld) {
@@ -246,7 +246,7 @@ void dmUpdateEps(SDnodeData *pData, SArray *eps) {
     }
   }
 
-  taosWUnLockLatch(&pData->latch);
+  taosThreadRwlockUnlock(&pData->lock);
 }
 
 static void dmResetEps(SDnodeData *pData, SArray *dnodeEps) {
@@ -292,7 +292,7 @@ static void dmPrintEps(SDnodeData *pData) {
 static bool dmIsEpChanged(SDnodeData *pData, int32_t dnodeId, const char *ep) {
   bool changed = false;
   if (dnodeId == 0) return changed;
-  taosRLockLatch(&pData->latch);
+  taosThreadRwlockRdlock(&pData->lock);
 
   SDnodeEp *pDnodeEp = taosHashGet(pData->dnodeHash, &dnodeId, sizeof(int32_t));
   if (pDnodeEp != NULL) {
@@ -304,24 +304,23 @@ static bool dmIsEpChanged(SDnodeData *pData, int32_t dnodeId, const char *ep) {
     }
   }
 
-  taosRUnLockLatch(&pData->latch);
+  taosThreadRwlockUnlock(&pData->lock);
   return changed;
 }
 
 void dmGetMnodeEpSet(SDnodeData *pData, SEpSet *pEpSet) {
-  taosRLockLatch(&pData->latch);
+  taosThreadRwlockRdlock(&pData->lock);
   *pEpSet = pData->mnodeEps;
-  taosRUnLockLatch(&pData->latch);
+  taosThreadRwlockUnlock(&pData->lock);
 }
 
 void dmSetMnodeEpSet(SDnodeData *pData, SEpSet *pEpSet) {
-  dInfo("mnode is changed, num:%d use:%d", pEpSet->numOfEps, pEpSet->inUse);
-
-  taosWLockLatch(&pData->latch);
+  taosThreadRwlockWrlock(&pData->lock);
   pData->mnodeEps = *pEpSet;
+  taosThreadRwlockUnlock(&pData->lock);
+
+  dInfo("mnode is changed, num:%d use:%d", pEpSet->numOfEps, pEpSet->inUse);
   for (int32_t i = 0; i < pEpSet->numOfEps; ++i) {
     dInfo("mnode index:%d %s:%u", i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port);
   }
-
-  taosWUnLockLatch(&pData->latch);
 }

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/mndConsumer.c

@@ -848,7 +848,7 @@ static int32_t mndRetrieveConsumer(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *
       pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
       colDataAppend(pColInfo, numOfRows, (const char *)cgroup, false);
 
-      // app id
+      // client id
       char clientId[TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE] = {0};
       tstrncpy(varDataVal(clientId), pConsumer->clientId, TSDB_CGROUP_LEN);
       varDataSetLen(clientId, strlen(varDataVal(clientId)));

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/mnode/sdb/src/sdb.c

@@ -48,7 +48,7 @@ SSdb *sdbInit(SSdbOpt *pOption) {
   }
 
   for (ESdbType i = 0; i < SDB_MAX; ++i) {
-    taosInitRWLatch(&pSdb->locks[i]);
+    taosThreadRwlockInit(&pSdb->locks[i], NULL);
     pSdb->maxId[i] = 0;
     pSdb->tableVer[i] = 0;
     pSdb->keyTypes[i] = SDB_KEY_INT32;
@@ -98,7 +98,10 @@ void sdbCleanup(SSdb *pSdb) {
 
     taosHashClear(hash);
     taosHashCleanup(hash);
+    taosThreadRwlockDestroy(&pSdb->locks[i]);
     pSdb->hashObjs[i] = NULL;
+    memset(&pSdb->locks[i], 0, sizeof(pSdb->locks[i]));
+
     mDebug("sdb table:%s is cleaned up", sdbTableName(i));
   }
 
@@ -134,7 +137,6 @@ int32_t sdbSetTable(SSdb *pSdb, SSdbTable table) {
 
   pSdb->maxId[sdbType] = 0;
   pSdb->hashObjs[sdbType] = hash;
-  taosInitRWLatch(&pSdb->locks[sdbType]);
   mDebug("sdb table:%s is initialized", sdbTableName(sdbType));
 
   return 0;

source/dnode/mnode/sdb/src/sdbFile.c

@@ -257,8 +257,8 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) {
     mTrace("write %s to file, total %d rows", sdbTableName(i), sdbGetSize(pSdb, i));
 
     SHashObj *hash = pSdb->hashObjs[i];
-    SRWLatch *pLock = &pSdb->locks[i];
-    taosWLockLatch(pLock);
+    TdThreadRwlock *pLock = &pSdb->locks[i];
+    taosThreadRwlockWrlock(pLock);
 
     SSdbRow **ppRow = taosHashIterate(hash, NULL);
     while (ppRow != NULL) {
@@ -303,7 +303,7 @@ static int32_t sdbWriteFileImp(SSdb *pSdb) {
       sdbFreeRaw(pRaw);
       ppRow = taosHashIterate(hash, ppRow);
     }
-    taosWUnLockLatch(pLock);
+    taosThreadRwlockUnlock(pLock);
   }
 
   if (code == 0) {

source/dnode/mnode/sdb/src/sdbHash.c

@@ -129,12 +129,12 @@ static int32_t sdbGetkeySize(SSdb *pSdb, ESdbType type, const void *pKey) {
 }
 
 static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *pRow, int32_t keySize) {
-  SRWLatch *pLock = &pSdb->locks[pRow->type];
-  taosWLockLatch(pLock);
+  TdThreadRwlock *pLock = &pSdb->locks[pRow->type];
+  taosThreadRwlockWrlock(pLock);
 
   SSdbRow *pOldRow = taosHashGet(hash, pRow->pObj, keySize);
   if (pOldRow != NULL) {
-    taosWUnLockLatch(pLock);
+    taosThreadRwlockUnlock(pLock);
     sdbFreeRow(pSdb, pRow, false);
     terrno = TSDB_CODE_SDB_OBJ_ALREADY_THERE;
     return terrno;
@@ -145,13 +145,13 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
   sdbPrintOper(pSdb, pRow, "insert");
 
   if (taosHashPut(hash, pRow->pObj, keySize, &pRow, sizeof(void *)) != 0) {
-    taosWUnLockLatch(pLock);
+    taosThreadRwlockUnlock(pLock);
     sdbFreeRow(pSdb, pRow, false);
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return terrno;
   }
 
-  taosWUnLockLatch(pLock);
+  taosThreadRwlockUnlock(pLock);
 
   int32_t     code = 0;
   SdbInsertFp insertFp = pSdb->insertFps[pRow->type];
@@ -159,9 +159,9 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
     code = (*insertFp)(pSdb, pRow->pObj);
     if (code != 0) {
       code = terrno;
-      taosWLockLatch(pLock);
+      taosThreadRwlockWrlock(pLock);
       taosHashRemove(hash, pRow->pObj, keySize);
-      taosWUnLockLatch(pLock);
+      taosThreadRwlockUnlock(pLock);
       sdbFreeRow(pSdb, pRow, false);
       terrno = code;
       return terrno;
@@ -180,19 +180,19 @@ static int32_t sdbInsertRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
 }
 
 static int32_t sdbUpdateRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *pNewRow, int32_t keySize) {
-  SRWLatch *pLock = &pSdb->locks[pNewRow->type];
-  taosWLockLatch(pLock);
+  TdThreadRwlock *pLock = &pSdb->locks[pNewRow->type];
+  taosThreadRwlockWrlock(pLock);
 
   SSdbRow **ppOldRow = taosHashGet(hash, pNewRow->pObj, keySize);
   if (ppOldRow == NULL || *ppOldRow == NULL) {
-    taosWUnLockLatch(pLock);
+    taosThreadRwlockUnlock(pLock);
     return sdbInsertRow(pSdb, hash, pRaw, pNewRow, keySize);
   }
 
   SSdbRow *pOldRow = *ppOldRow;
   pOldRow->status = pRaw->status;
   sdbPrintOper(pSdb, pOldRow, "update");
-  taosWUnLockLatch(pLock);
+  taosThreadRwlockUnlock(pLock);
 
   int32_t     code = 0;
   SdbUpdateFp updateFp = pSdb->updateFps[pNewRow->type];
@@ -207,12 +207,12 @@ static int32_t sdbUpdateRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
 }
 
 static int32_t sdbDeleteRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *pRow, int32_t keySize) {
-  SRWLatch *pLock = &pSdb->locks[pRow->type];
-  taosWLockLatch(pLock);
+  TdThreadRwlock *pLock = &pSdb->locks[pRow->type];
+  taosThreadRwlockWrlock(pLock);
 
   SSdbRow **ppOldRow = taosHashGet(hash, pRow->pObj, keySize);
   if (ppOldRow == NULL || *ppOldRow == NULL) {
-    taosWUnLockLatch(pLock);
+    taosThreadRwlockUnlock(pLock);
     sdbFreeRow(pSdb, pRow, false);
     terrno = TSDB_CODE_SDB_OBJ_NOT_THERE;
     return terrno;
@@ -223,7 +223,7 @@ static int32_t sdbDeleteRow(SSdb *pSdb, SHashObj *hash, SSdbRaw *pRaw, SSdbRow *
   sdbPrintOper(pSdb, pOldRow, "delete");
 
   taosHashRemove(hash, pOldRow->pObj, keySize);
-  taosWUnLockLatch(pLock);
+  taosThreadRwlockUnlock(pLock);
 
   pSdb->tableVer[pOldRow->type]++;
   sdbFreeRow(pSdb, pRow, false);
@@ -278,12 +278,12 @@ void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey) {
   void   *pRet = NULL;
   int32_t keySize = sdbGetkeySize(pSdb, type, pKey);
 
-  SRWLatch *pLock = &pSdb->locks[type];
-  taosRLockLatch(pLock);
+  TdThreadRwlock *pLock = &pSdb->locks[type];
+  taosThreadRwlockRdlock(pLock);
 
   SSdbRow **ppRow = taosHashGet(hash, pKey, keySize);
   if (ppRow == NULL || *ppRow == NULL) {
-    taosRUnLockLatch(pLock);
+    taosThreadRwlockUnlock(pLock);
     terrno = TSDB_CODE_SDB_OBJ_NOT_THERE;
     return NULL;
   }
@@ -306,13 +306,13 @@ void *sdbAcquire(SSdb *pSdb, ESdbType type, const void *pKey) {
       break;
   }
 
-  taosRUnLockLatch(pLock);
+  taosThreadRwlockUnlock(pLock);
   return pRet;
 }
 
 static void sdbCheckRow(SSdb *pSdb, SSdbRow *pRow) {
-  SRWLatch *pLock = &pSdb->locks[pRow->type];
-  taosWLockLatch(pLock);
+  TdThreadRwlock *pLock = &pSdb->locks[pRow->type];
+  taosThreadRwlockWrlock(pLock);
 
   int32_t ref = atomic_load_32(&pRow->refCount);
   sdbPrintOper(pSdb, pRow, "check");
@@ -320,7 +320,7 @@ static void sdbCheckRow(SSdb *pSdb, SSdbRow *pRow) {
     sdbFreeRow(pSdb, pRow, true);
   }
 
-  taosWUnLockLatch(pLock);
+  taosThreadRwlockUnlock(pLock);
 }
 
 void sdbRelease(SSdb *pSdb, void *pObj) {
@@ -329,8 +329,8 @@ void sdbRelease(SSdb *pSdb, void *pObj) {
   SSdbRow *pRow = (SSdbRow *)((char *)pObj - sizeof(SSdbRow));
   if (pRow->type >= SDB_MAX) return;
 
-  SRWLatch *pLock = &pSdb->locks[pRow->type];
-  taosWLockLatch(pLock);
+  TdThreadRwlock *pLock = &pSdb->locks[pRow->type];
+  taosThreadRwlockWrlock(pLock);
 
   int32_t ref = atomic_sub_fetch_32(&pRow->refCount, 1);
   sdbPrintOper(pSdb, pRow, "release");
@@ -338,7 +338,7 @@ void sdbRelease(SSdb *pSdb, void *pObj) {
     sdbFreeRow(pSdb, pRow, true);
   }
 
-  taosWUnLockLatch(pLock);
+  taosThreadRwlockUnlock(pLock);
 }
 
 void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) {
@@ -347,8 +347,8 @@ void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) {
   SHashObj *hash = sdbGetHash(pSdb, type);
   if (hash == NULL) return NULL;
 
-  SRWLatch *pLock = &pSdb->locks[type];
-  taosRLockLatch(pLock);
+  TdThreadRwlock *pLock = &pSdb->locks[type];
+  taosThreadRwlockRdlock(pLock);
 
   SSdbRow **ppRow = taosHashIterate(hash, pIter);
   while (ppRow != NULL) {
@@ -363,7 +363,7 @@ void *sdbFetch(SSdb *pSdb, ESdbType type, void *pIter, void **ppObj) {
     *ppObj = pRow->pObj;
     break;
   }
-  taosRUnLockLatch(pLock);
+  taosThreadRwlockUnlock(pLock);
 
   return ppRow;
 }
@@ -374,18 +374,18 @@ void sdbCancelFetch(SSdb *pSdb, void *pIter) {
   SHashObj *hash = sdbGetHash(pSdb, pRow->type);
   if (hash == NULL) return;
 
-  SRWLatch *pLock = &pSdb->locks[pRow->type];
-  taosRLockLatch(pLock);
+  TdThreadRwlock *pLock = &pSdb->locks[pRow->type];
+  taosThreadRwlockRdlock(pLock);
   taosHashCancelIterate(hash, pIter);
-  taosRUnLockLatch(pLock);
+  taosThreadRwlockUnlock(pLock);
 }
 
 void sdbTraverse(SSdb *pSdb, ESdbType type, sdbTraverseFp fp, void *p1, void *p2, void *p3) {
   SHashObj *hash = sdbGetHash(pSdb, type);
   if (hash == NULL) return;
 
-  SRWLatch *pLock = &pSdb->locks[type];
-  taosRLockLatch(pLock);
+  TdThreadRwlock *pLock = &pSdb->locks[type];
+  taosThreadRwlockRdlock(pLock);
 
   SSdbRow **ppRow = taosHashIterate(hash, NULL);
   while (ppRow != NULL) {
@@ -401,17 +401,17 @@ void sdbTraverse(SSdb *pSdb, ESdbType type, sdbTraverseFp fp, void *p1, void *p2
     ppRow = taosHashIterate(hash, ppRow);
   }
 
-  taosRUnLockLatch(pLock);
+  taosThreadRwlockUnlock(pLock);
 }
 
 int32_t sdbGetSize(SSdb *pSdb, ESdbType type) {
   SHashObj *hash = sdbGetHash(pSdb, type);
   if (hash == NULL) return 0;
 
-  SRWLatch *pLock = &pSdb->locks[type];
-  taosRLockLatch(pLock);
+  TdThreadRwlock *pLock = &pSdb->locks[type];
+  taosThreadRwlockRdlock(pLock);
   int32_t size = taosHashGetSize(hash);
-  taosRUnLockLatch(pLock);
+  taosThreadRwlockUnlock(pLock);
 
   return size;
 }
@@ -424,8 +424,8 @@ int32_t sdbGetMaxId(SSdb *pSdb, ESdbType type) {
 
   int32_t maxId = 0;
 
-  SRWLatch *pLock = &pSdb->locks[type];
-  taosRLockLatch(pLock);
+  TdThreadRwlock *pLock = &pSdb->locks[type];
+  taosThreadRwlockRdlock(pLock);
 
   SSdbRow **ppRow = taosHashIterate(hash, NULL);
   while (ppRow != NULL) {
@@ -435,7 +435,7 @@ int32_t sdbGetMaxId(SSdb *pSdb, ESdbType type) {
     ppRow = taosHashIterate(hash, ppRow);
   }
 
-  taosRUnLockLatch(pLock);
+  taosThreadRwlockUnlock(pLock);
 
   maxId = TMAX(maxId, pSdb->maxId[type]);
   return maxId + 1;

source/dnode/vnode/inc/vnode.h

@@ -128,6 +128,7 @@ int     tqReadHandleSetTbUidList(STqReadHandle *pHandle, const SArray *tbUidList
 int     tqReadHandleAddTbUidList(STqReadHandle *pHandle, const SArray *tbUidList);
 int32_t tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitReq *pMsg, int64_t ver);
 bool    tqNextDataBlock(STqReadHandle *pHandle);
+bool    tqNextDataBlockFilterOut(STqReadHandle *pHandle, SHashObj *filterOutUids);
 int32_t tqRetrieveDataBlock(SArray **ppCols, STqReadHandle *pHandle, uint64_t *pGroupId, uint64_t *pUid,
                             int32_t *pNumOfRows, int16_t *pNumOfCols);
 

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

@@ -163,6 +163,7 @@ typedef struct {
   int8_t        withSchema;
   int8_t        withTag;
   char*         qmsg;
+  SHashObj*     pDropTbUid;
   STqPushHandle pushHandle;
   // SRWLatch        lock;
   SWalReadHandle* pWalReader;

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/inc/vnodeInt.h

@@ -82,7 +82,7 @@ int             metaCreateSTable(SMeta* pMeta, int64_t version, SVCreateStbReq*
 int             metaAlterSTable(SMeta* pMeta, int64_t version, SVCreateStbReq* pReq);
 int             metaDropSTable(SMeta* pMeta, int64_t verison, SVDropStbReq* pReq);
 int             metaCreateTable(SMeta* pMeta, int64_t version, SVCreateTbReq* pReq);
-int             metaDropTable(SMeta* pMeta, int64_t version, SVDropTbReq* pReq);
+int             metaDropTable(SMeta* pMeta, int64_t version, SVDropTbReq* pReq, SArray* tbUids);
 int             metaAlterTable(SMeta* pMeta, int64_t version, SVAlterTbReq* pReq);
 SSchemaWrapper* metaGetTableSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver, bool isinline);
 STSchema*       metaGetTbTSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver);
@@ -104,7 +104,7 @@ int          tsdbOpen(SVnode* pVnode, STsdb** ppTsdb, const char* dir, STsdbKeep
 int          tsdbClose(STsdb** pTsdb);
 int          tsdbBegin(STsdb* pTsdb);
 int          tsdbCommit(STsdb* pTsdb);
-int          tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, const SSubmitReq *pMsg);
+int          tsdbScanAndConvertSubmitMsg(STsdb* pTsdb, const SSubmitReq* pMsg);
 int          tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSubmitRsp* pRsp);
 int          tsdbInsertTableData(STsdb* pTsdb, SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, SSubmitBlkRsp* pRsp);
 tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
@@ -118,7 +118,7 @@ STQ*    tqOpen(const char* path, SVnode* pVnode, SWal* pWal);
 void    tqClose(STQ*);
 int     tqPushMsg(STQ*, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver);
 int     tqCommit(STQ*);
-int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList);
+int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd);
 int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen);
 int32_t tqProcessVgDeleteReq(STQ* pTq, char* msg, int32_t msgLen);
 int32_t tqProcessTaskExec(STQ* pTq, char* msg, int32_t msgLen, int32_t workerId);

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

@@ -44,6 +44,7 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta) {
   pMeta->path = (char *)&pMeta[1];
   sprintf(pMeta->path, "%s%s%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path, TD_DIRSEP,
           VNODE_META_DIR);
+  taosRealPath(pMeta->path, NULL, slen);
   pMeta->pVnode = pVnode;
 
   // create path if not created yet

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

@@ -255,7 +255,7 @@ _err:
   return -1;
 }
 
-int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) {
+int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUids) {
   TBC        *pTbDbc = NULL;
   TBC        *pUidIdxc = NULL;
   TBC        *pNameIdxc = NULL;
@@ -336,6 +336,7 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq) {
   if (type == TSDB_CHILD_TABLE) {
     ctime = me.ctbEntry.ctime;
     suid = me.ctbEntry.suid;
+    taosArrayPush(tbUids, &me.uid);
   } else if (type == TSDB_NORMAL_TABLE) {
     ctime = me.ntbEntry.ctime;
     suid = 0;

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

@@ -104,16 +104,26 @@ static void tdSRowDemo() {
   taosMemoryFree(pTSChema);
 }
 
-int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList) {
+int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd) {
   void*    pIter = NULL;
   STqExec* pExec = NULL;
   while (1) {
     pIter = taosHashIterate(pTq->execs, pIter);
     if (pIter == NULL) break;
     pExec = (STqExec*)pIter;
-    if (pExec->subType == TOPIC_SUB_TYPE__DB) continue;
+    if (pExec->subType == TOPIC_SUB_TYPE__DB) {
+      if (isAdd) {
+        continue;
+      } else {
+        int32_t sz = taosArrayGetSize(tbUidList);
+        for (int32_t i = 0; i < sz; i++) {
+          int64_t tbUid = *(int64_t*)taosArrayGet(tbUidList, i);
+          taosHashPut(pExec->pDropTbUid, &tbUid, sizeof(int64_t), NULL, 0);
+        }
+      }
+    }
     for (int32_t i = 0; i < 5; i++) {
-      int32_t code = qUpdateQualifiedTableId(pExec->task[i], tbUidList, true);
+      int32_t code = qUpdateQualifiedTableId(pExec->task[i], tbUidList, isAdd);
       ASSERT(code == 0);
     }
   }
@@ -582,7 +592,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
         rsp.withSchema = 1;
         STqReadHandle* pReader = pExec->pExecReader[workerId];
         tqReadHandleSetMsg(pReader, pCont, 0);
-        while (tqNextDataBlock(pReader)) {
+        while (tqNextDataBlockFilterOut(pReader, pExec->pDropTbUid)) {
           SSDataBlock block = {0};
           if (tqRetrieveDataBlock(&block.pDataBlock, pReader, &block.info.groupId, &block.info.uid, &block.info.rows,
                                   &block.info.numOfCols) < 0) {
@@ -915,9 +925,10 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
     req.qmsg = NULL;
 
     pExec->pWalReader = walOpenReadHandle(pTq->pVnode->pWal);
+    if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
       for (int32_t i = 0; i < 5; i++) {
         pExec->pExecReader[i] = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
-      if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
+
         SReadHandle handle = {
             .reader = pExec->pExecReader[i],
             .meta = pTq->pVnode->pMeta,
@@ -925,9 +936,12 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
         };
         pExec->task[i] = qCreateStreamExecTaskInfo(pExec->qmsg, &handle);
         ASSERT(pExec->task[i]);
+      }
     } else {
-        pExec->task[i] = NULL;
+      for (int32_t i = 0; i < 5; i++) {
+        pExec->pExecReader[i] = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
       }
+      pExec->pDropTbUid = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
     }
     taosHashPut(pTq->execs, req.subKey, strlen(req.subKey), pExec, sizeof(STqExec));
     return 0;

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

@@ -64,22 +64,28 @@ bool tqNextDataBlock(STqReadHandle* pHandle) {
     }
     if (pHandle->pBlock == NULL) return false;
 
-    /*pHandle->pBlock->uid = htobe64(pHandle->pBlock->uid);*/
-    /*if (pHandle->tbUid == pHandle->pBlock->uid) {*/
     if (pHandle->tbIdHash == NULL) {
       return true;
     }
     void* ret = taosHashGet(pHandle->tbIdHash, &pHandle->msgIter.uid, sizeof(int64_t));
     if (ret != NULL) {
-      /*printf("retrieve one tb %ld\n", pHandle->pBlock->uid);*/
-      /*pHandle->pBlock->tid = htonl(pHandle->pBlock->tid);*/
-      /*pHandle->pBlock->sversion = htonl(pHandle->pBlock->sversion);*/
-      /*pHandle->pBlock->dataLen = htonl(pHandle->pBlock->dataLen);*/
-      /*pHandle->pBlock->schemaLen = htonl(pHandle->pBlock->schemaLen);*/
-      /*pHandle->pBlock->numOfRows = htons(pHandle->pBlock->numOfRows);*/
       return true;
-      /*} else {*/
-      /*printf("skip one tb %ld\n", pHandle->pBlock->uid);*/
+    }
+  }
+  return false;
+}
+
+bool tqNextDataBlockFilterOut(STqReadHandle* pHandle, SHashObj* filterOutUids) {
+  while (1) {
+    if (tGetSubmitMsgNext(&pHandle->msgIter, &pHandle->pBlock) < 0) {
+      return false;
+    }
+    if (pHandle->pBlock == NULL) return false;
+
+    ASSERT(pHandle->tbIdHash == NULL);
+    void* ret = taosHashGet(filterOutUids, &pHandle->msgIter.uid, sizeof(int64_t));
+    if (ret == NULL) {
+      return true;
     }
   }
   return false;

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

@@ -55,6 +55,7 @@ int tsdbOpen(SVnode *pVnode, STsdb **ppTsdb, const char *dir, STsdbKeepCfg *pKee
   memcpy(pTsdb->dir, dir, strlen(dir));
   pTsdb->path = (char *)&pTsdb[1];
   sprintf(pTsdb->path, "%s%s%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path, TD_DIRSEP, dir);
+  taosRealPath(pTsdb->path, NULL, slen);
   pTsdb->pVnode = pVnode;
   pTsdb->repoLocked = false;
   taosThreadMutexInit(&pTsdb->mutex, NULL);