Merge branch '3.0' into test/jcy

Jenkinsfile

@@ -113,7 +113,7 @@ pipeline {
               '''
               sh'''
               cd ${WKC}/debug
-              ctest
+              ctest -VV
               '''
             }
           }

Jenkinsfile2

@@ -88,6 +88,12 @@ def pre_test(){
     cmake .. > /dev/null
     make -j4> /dev/null
     '''
+    sh'''
+    cd ${WKPY}
+    git reset --hard
+    git pull
+    pip3 install .
+    '''
     return 1
 }
 
@@ -97,6 +103,7 @@ pipeline {
   environment{
       WK = '/var/lib/jenkins/workspace/TDinternal'
       WKC= '/var/lib/jenkins/workspace/TDengine'
+      WKPY= '/var/lib/jenkins/workspace/taos-connector-python'
   }
   stages {
       stage('pre_build'){
@@ -113,12 +120,17 @@ pipeline {
             timeout(time: 45, unit: 'MINUTES'){
               pre_test()
               sh'''
-              cd ${WKC}/tests
-              ./test-all.sh b1fq
+              cd ${WKC}/debug
+              ctest -VV
               '''
               sh'''
-              cd ${WKC}/debug
-              ctest
+              export LD_LIBRARY_PATH=${WKC}/debug/build/lib
+              cd ${WKC}/tests/system-test
+              ./fulltest.sh
+              '''
+              sh'''
+              cd ${WKC}/tests
+              ./test-all.sh b1fq
               '''
             }
           }

cmake/cmake.define

 cmake_minimum_required(VERSION 3.16)
 
-set(CMAKE_VERBOSE_MAKEFILE ON)
+set(CMAKE_VERBOSE_MAKEFILE OFF)
 
 #set output directory
 SET(LIBRARY_OUTPUT_PATH ${PROJECT_BINARY_DIR}/build/lib)

examples/c/asyncdemo.c

@@ -45,7 +45,7 @@ typedef struct {
 
 void taos_insert_call_back(void *param, TAOS_RES *tres, int code);
 void taos_select_call_back(void *param, TAOS_RES *tres, int code);
-void taos_error(TAOS *taos);
+void shellPrintError(TAOS *taos);
 
 static void queryDB(TAOS *taos, char *command) {
   int i;
@@ -102,7 +102,7 @@ int main(int argc, char *argv[])
 
   taos = taos_connect(argv[1], "root", "taosdata", NULL, 0);
   if (taos == NULL)
-    taos_error(taos);
+    shellPrintError(taos);
 
   printf("success to connect to server\n");
 
@@ -193,7 +193,7 @@ int main(int argc, char *argv[])
   return 0;
 }
 
-void taos_error(TAOS *con)
+void shellPrintError(TAOS *con)
 {
   fprintf(stderr, "TDengine error: %s\n", taos_errstr(con));
   taos_close(con);

include/common/tdatablock.h

@@ -74,8 +74,8 @@ static FORCE_INLINE bool colDataIsNull_s(const SColumnInfoData* pColumnInfoData,
     }
     char *data = colDataGetVarData(pColumnInfoData, row);
     return (*data == TSDB_DATA_TYPE_NULL);
-  } 
-  
+  }
+
   if (!pColumnInfoData->hasNull) {
     return false;
   }

include/common/tglobal.h

@@ -34,13 +34,9 @@ extern int32_t  tsVersion;
 extern int32_t  tsStatusInterval;
 
 // common
-extern int32_t tsRpcTimer;
-extern int32_t tsRpcMaxTime;
-extern bool    tsRpcForceTcp;  // all commands go to tcp protocol if this is enabled
 extern int32_t tsMaxConnections;
 extern int32_t tsMaxShellConns;
 extern int32_t tsShellActivityTimer;
-extern int32_t tsMaxTmrCtrl;
 extern int32_t tsCompressMsgSize;
 extern int32_t tsCompressColData;
 extern int32_t tsMaxNumOfDistinctResults;
@@ -98,9 +94,6 @@ extern bool    tsDeadLockKillQuery;
 extern int32_t tsQueryPolicy;
 
 // client
-extern int32_t tsMaxWildCardsLen;
-extern int32_t tsMaxRegexStringLen;
-extern int32_t tsMaxNumOfOrderedResults;
 extern int32_t tsMinSlidingTime;
 extern int32_t tsMinIntervalTime;
 extern int32_t tsMaxStreamComputDelay;

include/common/tmsgdef.h

@@ -85,6 +85,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_DND_DROP_VNODE, "dnode-drop-vnode", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_DND_CONFIG_DNODE, "dnode-config-dnode", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_DND_SERVER_STATUS, "dnode-server-status", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_DND_NET_TEST, "dnode-net-test", NULL, NULL)
 
   // Requests handled by MNODE
   TD_NEW_MSG_SEG(TDMT_MND_MSG)

include/libs/nodes/plannodes.h

@@ -46,7 +46,7 @@ typedef struct SScanLogicNode {
   struct STableMeta* pMeta;
   SVgroupsInfo* pVgroupList;
   EScanType scanType;
-  uint8_t scanFlag;         // denotes reversed scan of data or not
+  uint8_t scanSeq[2]; // first is scan count, and second is reverse scan count
   STimeWindow scanRange;
   SName tableName;
   bool showRewrite;
@@ -189,9 +189,6 @@ typedef struct SScanPhysiNode {
   SNodeList* pScanCols;
   uint64_t uid;           // unique id of the table
   int8_t tableType;
-  int32_t order;         // scan order: TSDB_ORDER_ASC|TSDB_ORDER_DESC
-  int32_t count;         // repeat count
-  int32_t reverse;       // reverse scan count
   SName tableName;
 } SScanPhysiNode;
 
@@ -207,7 +204,7 @@ typedef struct SSystemTableScanPhysiNode {
 
 typedef struct STableScanPhysiNode {
   SScanPhysiNode scan;
-  uint8_t scanFlag;         // denotes reversed scan of data or not
+  uint8_t scanSeq[2]; // first is scan count, and second is reverse scan count
   STimeWindow scanRange;
   double ratio;
   int32_t dataRequired;

include/libs/planner/planner.h

@@ -37,6 +37,8 @@ typedef struct SPlanContext {
   bool isStmtQuery;
   void* pTransporter;
   struct SCatalog* pCatalog;
+  char* pMsg;
+  int32_t msgLen;
 } SPlanContext;
 
 // Create the physical plan for the query, according to the AST.

include/libs/transport/trpc.h

@@ -38,11 +38,12 @@ typedef struct SRpcConnInfo {
 
 typedef struct SRpcMsg {
   tmsg_t  msgType;
-  void   *pCont;
+  void *  pCont;
   int     contLen;
   int32_t code;
-  void   *handle;         // rpc handle returned to app
-  void   *ahandle;        // app handle set by client
+  void *  handle;         // rpc handle returned to app
+  void *  ahandle;        // app handle set by client
+  int64_t refId;          // refid, used by server
   int     noResp;         // has response or not(default 0, 0: resp, 1: no resp);
   int     persistHandle;  // persist handle or not
 
@@ -54,8 +55,8 @@ typedef struct {
   uint16_t clientPort;
   SRpcMsg  rpcMsg;
   int32_t  rspLen;
-  void    *pRsp;
-  void    *pNode;
+  void *   pRsp;
+  void *   pNode;
 } SNodeMsg;
 
 typedef void (*RpcCfp)(void *parent, SRpcMsg *, SEpSet *);
@@ -64,7 +65,7 @@ typedef int (*RpcRfp)(void *parent, SRpcMsg *, SEpSet *);
 
 typedef struct SRpcInit {
   uint16_t localPort;     // local port
-  char    *label;         // for debug purpose
+  char *   label;         // for debug purpose
   int      numOfThreads;  // number of threads to handle connections
   int      sessions;      // number of sessions allowed
   int8_t   connType;      // TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS
@@ -97,23 +98,23 @@ typedef struct {
 
 typedef struct {
   int32_t msgType;
-  void   *val;
+  void *  val;
   int32_t (*clone)(void *src, void **dst);
   void (*freeFunc)(const void *arg);
 } SRpcBrokenlinkVal;
 
 typedef struct {
-  SHashObj         *args;
+  SHashObj *        args;
   SRpcBrokenlinkVal brokenVal;
 } SRpcCtx;
 
 int32_t rpcInit();
 void    rpcCleanup();
-void   *rpcOpen(const SRpcInit *pRpc);
+void *  rpcOpen(const SRpcInit *pRpc);
 void    rpcClose(void *);
-void   *rpcMallocCont(int contLen);
+void *  rpcMallocCont(int contLen);
 void    rpcFreeCont(void *pCont);
-void   *rpcReallocCont(void *ptr, int contLen);
+void *  rpcReallocCont(void *ptr, int contLen);
 
 // Because taosd supports multi-process mode
 // These functions should not be used on the server side

include/os/osSystem.h

@@ -31,19 +31,19 @@ extern "C" {
 
 typedef struct TdCmd *TdCmdPtr;
 
-TdCmdPtr taosOpenCmd(const char *cmd);
-int64_t taosGetLineCmd(TdCmdPtr pCmd, char ** __restrict ptrBuf);
-int32_t taosEOFCmd(TdCmdPtr pCmd);
-int64_t taosCloseCmd(TdCmdPtr *ppCmd);
+TdCmdPtr taosOpenCmd(const char* cmd);
+int64_t  taosGetLineCmd(TdCmdPtr pCmd, char** __restrict ptrBuf);
+int32_t  taosEOFCmd(TdCmdPtr pCmd);
+int64_t  taosCloseCmd(TdCmdPtr* ppCmd);
 
 void* taosLoadDll(const char* filename);
 void* taosLoadSym(void* handle, char* name);
 void  taosCloseDll(void* handle);
 
 int32_t taosSetConsoleEcho(bool on);
-void setTerminalMode();
-int32_t getOldTerminalMode();
-void resetTerminalMode();
+void    taosSetTerminalMode();
+int32_t taosGetOldTerminalMode();
+void    taosResetTerminalMode();
 
 #ifdef __cplusplus
 }

include/util/taoserror.h

@@ -623,6 +623,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_INVALID_DAYS_VALUE        TAOS_DEF_ERROR_CODE(0, 0x2636)
 #define TSDB_CODE_PAR_OFFSET_LESS_ZERO          TAOS_DEF_ERROR_CODE(0, 0x2637)
 #define TSDB_CODE_PAR_SLIMIT_LEAK_PARTITION_BY  TAOS_DEF_ERROR_CODE(0, 0x2638)
+#define TSDB_CODE_PAR_INVALID_TOPIC_QUERY       TAOS_DEF_ERROR_CODE(0, 0x2639)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)

include/util/tdef.h

@@ -76,8 +76,6 @@ extern const int32_t TYPE_BYTES[15];
 #define TSDB_DEFAULT_PASS "taosdata"
 #endif
 
-#define SHELL_MAX_PASSWORD_LEN 20
-
 #define TSDB_TRUE  1
 #define TSDB_FALSE 0
 #define TSDB_OK    0

include/util/tprocess.h

@@ -53,8 +53,8 @@ int32_t   taosProcRun(SProcObj *pProc);
 void      taosProcStop(SProcObj *pProc);
 
 int32_t taosProcPutToChildQ(SProcObj *pProc, const void *pHead, int16_t headLen, const void *pBody, int32_t bodyLen,
-                            void *handle, EProcFuncType ftype);
-void    taosProcRemoveHandle(SProcObj *pProc, void *handle);
+                            void *handle, int64_t handleRef, EProcFuncType ftype);
+int64_t taosProcRemoveHandle(SProcObj *pProc, void *handle);
 void    taosProcCloseHandles(SProcObj *pProc, void (*HandleFp)(void *handle));
 void    taosProcPutToParentQ(SProcObj *pProc, const void *pHead, int16_t headLen, const void *pBody, int32_t bodyLen,
                              EProcFuncType ftype);

source/client/src/clientImpl.c

@@ -232,7 +232,9 @@ int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArra
     .mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp),
     .pAstRoot = pQuery->pRoot,
     .showRewrite = pQuery->showRewrite,
-    .pTransporter = pRequest->pTscObj->pAppInfo->pTransporter
+    .pTransporter = pRequest->pTscObj->pAppInfo->pTransporter,
+    .pMsg = pRequest->msgBuf,
+    .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE
   };
   int32_t code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &cxt.pCatalog);
   if (TSDB_CODE_SUCCESS == code) {

source/common/src/tglobal.c

@@ -32,13 +32,9 @@ int32_t  tsVersion = 30000000;
 int32_t  tsStatusInterval = 1;  // second
 
 // common
-int32_t tsRpcTimer = 300;
-int32_t tsRpcMaxTime = 600;    // seconds;
-bool    tsRpcForceTcp = true;  // disable this, means query, show command use udp protocol as default
 int32_t tsMaxShellConns = 50000;
 int32_t tsMaxConnections = 50000;
 int32_t tsShellActivityTimer = 3;  // second
-int32_t tsMaxBinaryDisplayWidth = 30;
 bool    tsEnableSlaveQuery = true;
 bool    tsPrintAuth = false;
 
@@ -105,14 +101,6 @@ int32_t tsCompressColData = -1;
  */
 int32_t tsCompatibleModel = 1;
 
-// client
-int32_t tsMaxWildCardsLen = TSDB_PATTERN_STRING_DEFAULT_LEN;
-int32_t tsMaxRegexStringLen = TSDB_REGEX_STRING_DEFAULT_LEN;
-
-// the maximum number of results for projection query on super table that are returned from
-// one virtual node, to order according to timestamp
-int32_t tsMaxNumOfOrderedResults = 100000;
-
 // 10 ms for sliding time, the value will changed in case of time precision changed
 int32_t tsMinSlidingTime = 10;
 
@@ -237,18 +225,20 @@ static int32_t taosLoadCfg(SConfig *pCfg, const char *inputCfgDir, const char *e
   char cfgFile[PATH_MAX + 100] = {0};
 
   taosExpandDir(inputCfgDir, cfgDir, PATH_MAX);
-  snprintf(cfgFile, sizeof(cfgFile), "%s" TD_DIRSEP "taos.cfg", cfgDir);
+  if (taosIsDir(cfgDir)) {
+    snprintf(cfgFile, sizeof(cfgFile), "%s" TD_DIRSEP "taos.cfg", cfgDir);
+  } else {
+    tstrncpy(cfgFile, cfgDir, sizeof(cfgDir));
+  }
 
   if (cfgLoad(pCfg, CFG_STYPE_APOLLO_URL, apolloUrl) != 0) {
     uError("failed to load from apollo url:%s since %s", apolloUrl, terrstr());
     return -1;
   }
 
-  if (cfgLoad(pCfg, CFG_STYPE_CFG_FILE, cfgDir) != 0) {
-    if (cfgLoad(pCfg, CFG_STYPE_CFG_FILE, cfgFile) != 0) {
-      uInfo("cfg file:%s not read since %s", cfgFile, terrstr());
-      return 0;
-    }
+  if (cfgLoad(pCfg, CFG_STYPE_CFG_FILE, cfgFile) != 0) {
+    uError("failed to load from cfg file:%s since %s", cfgFile, terrstr());
+    return -1;
   }
 
   if (cfgLoad(pCfg, CFG_STYPE_ENV_FILE, envFile) != 0) {
@@ -312,19 +302,10 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "serverPort", defaultServerPort, 1, 65056, 1) != 0) return -1;
   if (cfgAddDir(pCfg, "tempDir", tsTempDir, 1) != 0) return -1;
   if (cfgAddFloat(pCfg, "minimalTempDirGB", 1.0f, 0.001f, 10000000, 1) != 0) return -1;
-  if (cfgAddInt32(pCfg, "maxTmrCtrl", tsMaxTmrCtrl, 8, 2048, 1) != 0) return -1;
-  if (cfgAddInt32(pCfg, "rpcTimer", tsRpcTimer, 100, 3000, 1) != 0) return -1;
-  if (cfgAddInt32(pCfg, "rpcMaxTime", tsRpcMaxTime, 100, 7200, 1) != 0) return -1;
-  if (cfgAddBool(pCfg, "rpcForceTcp", tsRpcForceTcp, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "shellActivityTimer", tsShellActivityTimer, 1, 120, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "compressMsgSize", tsCompressMsgSize, -1, 100000000, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "compressColData", tsCompressColData, -1, 100000000, 1) != 0) return -1;
-  if (cfgAddInt32(pCfg, "maxWildCardsLength", tsMaxWildCardsLen, 0, TSDB_MAX_FIELD_LEN, 1) != 0) return -1;
-  if (cfgAddInt32(pCfg, "maxRegexStringLen", tsMaxRegexStringLen, 0, TSDB_MAX_FIELD_LEN, 1) != 0) return -1;
-  if (cfgAddInt32(pCfg, "maxNumOfOrderedRes", tsMaxNumOfOrderedResults, 128, TSDB_MAX_ALLOWED_SQL_LEN, 1) != 0)
-    return -1;
   if (cfgAddBool(pCfg, "keepColumnName", tsKeepOriginalColumnName, 1) != 0) return -1;
-  if (cfgAddInt32(pCfg, "maxBinaryDisplayWidth", tsMaxBinaryDisplayWidth, 1, 65536, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "queryPolicy", tsQueryPolicy, 1, 3, 1) != 0) return -1;
 
   tsNumOfTaskQueueThreads = tsNumOfCores / 4;
@@ -506,18 +487,10 @@ static int32_t taosSetClientCfg(SConfig *pCfg) {
     return -1;
   }
 
-  tsMaxTmrCtrl = cfgGetItem(pCfg, "maxTmrCtrl")->i32;
-  tsRpcTimer = cfgGetItem(pCfg, "rpcTimer")->i32;
-  tsRpcMaxTime = cfgGetItem(pCfg, "rpcMaxTime")->i32;
-  tsRpcForceTcp = cfgGetItem(pCfg, "rpcForceTcp")->i32;
   tsShellActivityTimer = cfgGetItem(pCfg, "shellActivityTimer")->i32;
   tsCompressMsgSize = cfgGetItem(pCfg, "compressMsgSize")->i32;
   tsCompressColData = cfgGetItem(pCfg, "compressColData")->i32;
-  tsMaxWildCardsLen = cfgGetItem(pCfg, "maxWildCardsLength")->i32;
-  tsMaxRegexStringLen = cfgGetItem(pCfg, "maxRegexStringLen")->i32;
-  tsMaxNumOfOrderedResults = cfgGetItem(pCfg, "maxNumOfOrderedRes")->i32;
   tsKeepOriginalColumnName = cfgGetItem(pCfg, "keepColumnName")->bval;
-  tsMaxBinaryDisplayWidth = cfgGetItem(pCfg, "maxBinaryDisplayWidth")->i32;
   tsNumOfTaskQueueThreads = cfgGetItem(pCfg, "numOfTaskQueueThreads")->i32;
   tsQueryPolicy = cfgGetItem(pCfg, "queryPolicy")->i32;
   return 0;
@@ -708,6 +681,6 @@ void taosCfgDynamicOptions(const char *option, const char *value) {
 
   if (strcasecmp(option, "resetlog") == 0) {
     taosResetLog();
-    cfgDumpCfg(tsCfg, 1, false);
+    cfgDumpCfg(tsCfg, 0, false);
   }
 }

source/common/src/tmsg.c

@@ -1568,13 +1568,8 @@ int32_t tSerializeSCreateFuncReq(void *buf, int32_t bufLen, SCreateFuncReq *pReq
   if (tEncodeI32(&encoder, pReq->codeLen) < 0) return -1;
   if (tEncodeI64(&encoder, pReq->signature) < 0) return -1;
 
-  int32_t codeSize = 0;
   if (pReq->pCode != NULL) {
-    codeSize = strlen(pReq->pCode) + 1;
-  }
-  if (tEncodeI32(&encoder, codeSize) < 0) return -1;
-  if (pReq->pCode != NULL) {
-    if (tEncodeCStr(&encoder, pReq->pCode) < 0) return -1;
+    if (tEncodeBinary(&encoder, pReq->pCode, pReq->codeLen) < 0) return -1;
   }
 
   int32_t commentSize = 0;
@@ -1608,10 +1603,8 @@ int32_t tDeserializeSCreateFuncReq(void *buf, int32_t bufLen, SCreateFuncReq *pR
   if (tDecodeI32(&decoder, &pReq->codeLen) < 0) return -1;
   if (tDecodeI64(&decoder, &pReq->signature) < 0) return -1;
 
-  int32_t codeSize = 0;
-  if (tDecodeI32(&decoder, &codeSize) < 0) return -1;
-  if (codeSize > 0) {
-    pReq->pCode = taosMemoryCalloc(1, codeSize);
+  if (pReq->codeLen > 0) {
+    pReq->pCode = taosMemoryCalloc(1, pReq->codeLen);
     if (pReq->pCode == NULL) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       return -1;
@@ -1734,7 +1727,7 @@ int32_t tSerializeSRetrieveFuncRsp(void *buf, int32_t bufLen, SRetrieveFuncRsp *
     if (tEncodeI32(&encoder, pInfo->codeSize) < 0) return -1;
     if (tEncodeI32(&encoder, pInfo->commentSize) < 0) return -1;
     if (pInfo->codeSize) {
-      if (tEncodeCStr(&encoder, pInfo->pCode) < 0) return -1;
+      if (tEncodeBinary(&encoder, pInfo->pCode, pInfo->codeSize) < 0) return -1;
     }
     if (pInfo->commentSize) {
       if (tEncodeCStr(&encoder, pInfo->pComment) < 0) return -1;

source/dnode/mgmt/exe/dmMain.c

@@ -106,7 +106,7 @@ static void dmPrintVersion() {
 
 static void dmDumpCfg() {
   SConfig *pCfg = taosGetCfg();
-  cfgDumpCfg(pCfg, 0, 1);
+  cfgDumpCfg(pCfg, 0, true);
 }
 
 static SDnodeOpt dmGetOpt() {
@@ -190,7 +190,7 @@ int main(int argc, char const *argv[]) {
   }
 
   if (dmInitLog() != 0) {
-    printf("failed to start since init log error\n");
+    dError("failed to start since init log error");
     return -1;
   }
 

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

@@ -17,7 +17,7 @@
 #include "dmImp.h"
 
 static void dmUpdateDnodeCfg(SDnode *pDnode, SDnodeCfg *pCfg) {
-  if (pDnode->data.dnodeId == 0) {
+  if (pDnode->data.dnodeId == 0 || pDnode->data.clusterId == 0) {
     dInfo("set dnodeId:%d clusterId:%" PRId64, pCfg->dnodeId, pCfg->clusterId);
     taosWLockLatch(&pDnode->data.latch);
     pDnode->data.dnodeId = pCfg->dnodeId;
@@ -57,6 +57,7 @@ void dmSendStatusReq(SDnode *pDnode) {
   req.dnodeVer = pDnode->data.dnodeVer;
   req.dnodeId = pDnode->data.dnodeId;
   req.clusterId = pDnode->data.clusterId;
+  if (req.clusterId == 0) req.dnodeId = 0;
   req.rebootTime = pDnode->data.rebootTime;
   req.updateTime = pDnode->data.updateTime;
   req.numOfCores = tsNumOfCores;
@@ -145,10 +146,10 @@ int32_t dmProcessCreateNodeReq(SDnode *pDnode, EDndNodeType ntype, SNodeMsg *pMs
     dError("node:%s, failed to create since %s", pWrapper->name, terrstr());
   } else {
     dDebug("node:%s, has been created", pWrapper->name);
+    (void)dmOpenNode(pWrapper);
     pWrapper->required = true;
     pWrapper->deployed = true;
     pWrapper->procType = pDnode->ptype;
-    (void)dmOpenNode(pWrapper);
   }
 
   taosThreadMutexUnlock(&pDnode->mutex);
@@ -170,13 +171,13 @@ int32_t dmProcessDropNodeReq(SDnode *pDnode, EDndNodeType ntype, SNodeMsg *pMsg)
     dError("node:%s, failed to drop since %s", pWrapper->name, terrstr());
   } else {
     dDebug("node:%s, has been dropped", pWrapper->name);
+    pWrapper->required = false;
+    pWrapper->deployed = false;
   }
 
   dmReleaseWrapper(pWrapper);
 
   if (code == 0) {
-    pWrapper->required = false;
-    pWrapper->deployed = false;
     dmCloseNode(pWrapper);
     taosRemoveDir(pWrapper->path);
   }
@@ -310,6 +311,9 @@ static void dmWatchUdfd(void *args) {
 }
 
 static int32_t dmStartUdfd(SDnode *pDnode) {
+  char dnodeId[8] = {0};
+  snprintf(dnodeId, sizeof(dnodeId), "%d", pDnode->data.dnodeId);
+  uv_os_setenv("DNODE_ID", dnodeId);
   SUdfdData *pData = &pDnode->udfdData;
   if (pData->startCalled) {
     dInfo("dnode-mgmt start udfd already called");
@@ -319,8 +323,17 @@ static int32_t dmStartUdfd(SDnode *pDnode) {
   uv_barrier_init(&pData->barrier, 2);
   uv_thread_create(&pData->thread, dmWatchUdfd, pDnode);
   uv_barrier_wait(&pData->barrier);
-  pData->needCleanUp = true;
-  return pData->spawnErr;
+  int32_t err = atomic_load_32(&pData->spawnErr);
+  if (err != 0) {
+    uv_barrier_destroy(&pData->barrier);
+    uv_async_send(&pData->stopAsync);
+    uv_thread_join(&pData->thread);
+    pData->needCleanUp = false;
+    dInfo("dnode-mgmt udfd cleaned up after spawn err");
+  } else {
+    pData->needCleanUp = true;
+  }
+  return err;
 }
 
 static int32_t dmStopUdfd(SDnode *pDnode) {
@@ -335,7 +348,7 @@ static int32_t dmStopUdfd(SDnode *pDnode) {
   uv_barrier_destroy(&pData->barrier);
   uv_async_send(&pData->stopAsync);
   uv_thread_join(&pData->thread);
-
+  dInfo("dnode-mgmt udfd cleaned up");
   return 0;
 }
 
@@ -370,9 +383,9 @@ static int32_t dmInitMgmt(SMgmtWrapper *pWrapper) {
   }
   dmReportStartup(pDnode, "dnode-transport", "initialized");
 
-//  if (dmStartUdfd(pDnode) != 0) {
-//    dError("failed to start udfd");
-//  }
+  if (dmStartUdfd(pDnode) != 0) {
+    dError("failed to start udfd");
+  }
 
   dInfo("dnode-mgmt is initialized");
   return 0;

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

@@ -59,6 +59,10 @@ static inline int32_t dmBuildMsg(SNodeMsg *pMsg, SRpcMsg *pRpc) {
   pMsg->clientIp = connInfo.clientIp;
   pMsg->clientPort = connInfo.clientPort;
   memcpy(&pMsg->rpcMsg, pRpc, sizeof(SRpcMsg));
+  if ((pRpc->msgType & 1u)) {
+    assert(pRpc->refId != 0);
+  }
+  // assert(pRpc->handle != NULL && pRpc->refId != 0 && pMsg->rpcMsg.refId != 0);
   return 0;
 }
 
@@ -67,12 +71,15 @@ static void dmProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSe
   SNodeMsg *pMsg = NULL;
   NodeMsgFp msgFp = NULL;
   uint16_t  msgType = pRpc->msgType;
+  bool      needRelease = false;
 
   if (pEpSet && pEpSet->numOfEps > 0 && msgType == TDMT_MND_STATUS_RSP) {
     dmSetMnodeEpSet(pWrapper->pDnode, pEpSet);
   }
 
   if (dmMarkWrapper(pWrapper) != 0) goto _OVER;
+
+  needRelease = true;
   if ((msgFp = dmGetMsgFp(pWrapper, pRpc)) == NULL) goto _OVER;
   if ((pMsg = taosAllocateQitem(sizeof(SNodeMsg))) == NULL) goto _OVER;
   if (dmBuildMsg(pMsg, pRpc) != 0) goto _OVER;
@@ -84,7 +91,7 @@ static void dmProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSe
     dTrace("msg:%p, is created and put into child queue, type:%s handle:%p user:%s", pMsg, TMSG_INFO(msgType),
            pRpc->handle, pMsg->user);
     code = taosProcPutToChildQ(pWrapper->procObj, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen, pRpc->handle,
-                               PROC_FUNC_REQ);
+                               pRpc->refId, PROC_FUNC_REQ);
   } else {
     dTrace("msg:%p, should not processed in child process, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user);
     ASSERT(1);
@@ -107,7 +114,7 @@ _OVER:
         }
       }
 
-      SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = code};
+      SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = code, .refId = pRpc->refId};
       tmsgSendRsp(&rsp);
     }
     dTrace("msg:%p, is freed", pMsg);
@@ -115,7 +122,9 @@ _OVER:
     rpcFreeCont(pRpc->pCont);
   }
 
-  dmReleaseWrapper(pWrapper);
+  if (needRelease) {
+    dmReleaseWrapper(pWrapper);
+  }
 }
 
 static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
@@ -131,10 +140,17 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
     return;
   }
 
+  if (msgType == TDMT_DND_NET_TEST) {
+    dTrace("net test req will be processed, handle:%p, app:%p", pMsg->handle, pMsg->ahandle);
+    dmProcessServerStatusReq(pDnode, pMsg);
+    return;
+  }
+
   if (pDnode->status != DND_STAT_RUNNING) {
     dError("msg:%s ignored since dnode not running, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle);
     if (isReq) {
-      SRpcMsg rspMsg = {.handle = pMsg->handle, .code = TSDB_CODE_APP_NOT_READY, .ahandle = pMsg->ahandle};
+      SRpcMsg rspMsg = {
+          .handle = pMsg->handle, .code = TSDB_CODE_APP_NOT_READY, .ahandle = pMsg->ahandle, .refId = pMsg->refId};
       rpcSendResponse(&rspMsg);
     }
     rpcFreeCont(pMsg->pCont);
@@ -143,7 +159,8 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
 
   if (isReq && pMsg->pCont == NULL) {
     dError("req:%s not processed since its empty, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle);
-    SRpcMsg rspMsg = {.handle = pMsg->handle, .code = TSDB_CODE_INVALID_MSG_LEN, .ahandle = pMsg->ahandle};
+    SRpcMsg rspMsg = {
+        .handle = pMsg->handle, .code = TSDB_CODE_INVALID_MSG_LEN, .ahandle = pMsg->ahandle, .refId = pMsg->refId};
     rpcSendResponse(&rspMsg);
     return;
   }
@@ -151,7 +168,8 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
   if (pWrapper == NULL) {
     dError("msg:%s not processed since no handle, handle:%p app:%p", TMSG_INFO(msgType), pMsg->handle, pMsg->ahandle);
     if (isReq) {
-      SRpcMsg rspMsg = {.handle = pMsg->handle, .code = TSDB_CODE_MSG_NOT_PROCESSED, .ahandle = pMsg->ahandle};
+      SRpcMsg rspMsg = {
+          .handle = pMsg->handle, .code = TSDB_CODE_MSG_NOT_PROCESSED, .ahandle = pMsg->ahandle, .refId = pMsg->refId};
       rpcSendResponse(&rspMsg);
     }
     rpcFreeCont(pMsg->pCont);
@@ -170,6 +188,9 @@ static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
   }
 
   dTrace("msg:%s will be processed by %s, app:%p", TMSG_INFO(msgType), pWrapper->name, pMsg->ahandle);
+  if (isReq) {
+    assert(pMsg->refId != 0);
+  }
   dmProcessRpcMsg(pWrapper, pMsg, pEpSet);
 }
 
@@ -317,7 +338,7 @@ static void dmConsumeChildQueue(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int16_t
   if (code != 0) {
     dError("msg:%p, failed to process since code:0x%04x:%s", pMsg, code & 0XFFFF, tstrerror(code));
     if (pRpc->msgType & 1U) {
-      SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = terrno};
+      SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = terrno, .refId = pRpc->refId};
       dmSendRsp(pWrapper, &rsp);
     }
 
@@ -346,7 +367,7 @@ static void dmConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t
       dmSendRpcReq(pWrapper->pDnode, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg);
       break;
     case PROC_FUNC_RSP:
-      taosProcRemoveHandle(pWrapper->procObj, pMsg->handle);
+      pMsg->refId = taosProcRemoveHandle(pWrapper->procObj, pMsg->handle);
       dmSendRpcRsp(pWrapper->pDnode, pMsg);
       break;
     default:

source/dnode/mgmt/implement/src/dmWorker.c

@@ -105,7 +105,7 @@ void dmStopMonitorThread(SDnode *pDnode) {
 }
 
 static void dmProcessMgmtQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
-  SDnode  *pDnode = pInfo->ahandle;
+  SDnode * pDnode = pInfo->ahandle;
   SRpcMsg *pRpc = &pMsg->rpcMsg;
   int32_t  code = -1;
   dTrace("msg:%p, will be processed in dnode-mgmt queue", pMsg);
@@ -150,7 +150,7 @@ static void dmProcessMgmtQueue(SQueueInfo *pInfo, SNodeMsg *pMsg) {
 
   if (pRpc->msgType & 1u) {
     if (code != 0) code = terrno;
-    SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = code};
+    SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = code, .refId = pRpc->refId};
     rpcSendResponse(&rsp);
   }
 

source/dnode/mgmt/interface/inc/dmInt.h

@@ -37,6 +37,7 @@ void   dmSetMsgHandle(SMgmtWrapper *pWrapper, tmsg_t msgType, NodeMsgFp nodeMsgF
 void   dmReportStartup(SDnode *pDnode, const char *pName, const char *pDesc);
 void   dmReportStartupByWrapper(SMgmtWrapper *pWrapper, const char *pName, const char *pDesc);
 void   dmProcessServerStatusReq(SDnode *pDnode, SRpcMsg *pMsg);
+void   dmProcessNettestReq(SDnode *pDnode, SRpcMsg *pMsg);
 void   dmGetMonitorSysInfo(SMonSysInfo *pInfo);
 
 // dmFile.c

source/dnode/mgmt/interface/src/dmInt.c

@@ -171,13 +171,21 @@ static void dmGetServerStatus(SDnode *pDnode, SServerStatusRsp *pStatus) {
   }
 }
 
+void dmProcessNettestReq(SDnode *pDnode, SRpcMsg *pRpc) {
+  dDebug("net test req is received");
+  SRpcMsg rsp = {.handle = pRpc->handle, .ahandle = pRpc->ahandle, .code = 0};
+  rsp.pCont = rpcMallocCont(pRpc->contLen);
+  rsp.contLen = pRpc->contLen;
+  rpcSendResponse(&rsp);
+}
+
 void dmProcessServerStatusReq(SDnode *pDnode, SRpcMsg *pReq) {
   dDebug("server status req is received");
 
   SServerStatusRsp statusRsp = {0};
   dmGetServerStatus(pDnode, &statusRsp);
 
-  SRpcMsg rspMsg = {.handle = pReq->handle, .ahandle = pReq->ahandle};
+  SRpcMsg rspMsg = {.handle = pReq->handle, .ahandle = pReq->ahandle, .refId = pReq->refId};
   int32_t rspLen = tSerializeSServerStatusRsp(NULL, 0, &statusRsp);
   if (rspLen < 0) {
     rspMsg.code = TSDB_CODE_OUT_OF_MEMORY;

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

@@ -53,7 +53,7 @@ int32_t bmProcessCreateReq(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
     return -1;
   }
 
-  if (createReq.dnodeId != pDnode->data.dnodeId) {
+  if (pDnode->data.dnodeId != 0 && createReq.dnodeId != pDnode->data.dnodeId) {
     terrno = TSDB_CODE_INVALID_OPTION;
     dError("failed to create bnode since %s, input:%d cur:%d", terrstr(), createReq.dnodeId, pDnode->data.dnodeId);
     return -1;

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

@@ -17,7 +17,8 @@
 #include "bmInt.h"
 
 static void bmSendErrorRsp(SNodeMsg *pMsg, int32_t code) {
-  SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code};
+  SRpcMsg rpcRsp = {
+      .handle = pMsg->rpcMsg.handle, .ahandle = pMsg->rpcMsg.ahandle, .code = code, .refId = pMsg->rpcMsg.refId};
   tmsgSendRsp(&rpcRsp);
 
   dTrace("msg:%p, is freed", pMsg);
@@ -38,6 +39,7 @@ static void bmSendErrorRsps(STaosQall *qall, int32_t numOfMsgs, int32_t code) {
 static inline void bmSendRsp(SNodeMsg *pMsg, int32_t code) {
   SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
                  .ahandle = pMsg->rpcMsg.ahandle,
+                 .refId = pMsg->rpcMsg.refId,
                  .code = code,
                  .pCont = pMsg->pRsp,
                  .contLen = pMsg->rspLen};
@@ -101,7 +103,7 @@ static void bmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 }
 
 int32_t bmProcessWriteMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SBnodeMgmt   *pMgmt = pWrapper->pMgmt;
+  SBnodeMgmt *  pMgmt = pWrapper->pMgmt;
   SMultiWorker *pWorker = &pMgmt->writeWorker;
 
   dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
@@ -110,7 +112,7 @@ int32_t bmProcessWriteMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 }
 
 int32_t bmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SBnodeMgmt    *pMgmt = pWrapper->pMgmt;
+  SBnodeMgmt *   pMgmt = pWrapper->pMgmt;
   SSingleWorker *pWorker = &pMgmt->monitorWorker;
 
   dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);

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

@@ -134,9 +134,9 @@ int32_t mmProcessAlterReq(SMnodeMgmt *pMgmt, SNodeMsg *pMsg) {
     return -1;
   }
 
-  if (alterReq.dnodeId != pDnode->data.dnodeId) {
+  if (pDnode->data.dnodeId != 0 && alterReq.dnodeId != pDnode->data.dnodeId) {
     terrno = TSDB_CODE_INVALID_OPTION;
-    dError("failed to alter mnode since %s, dnodeId:%d input:%d", terrstr(), pDnode->data.dnodeId, alterReq.dnodeId);
+    dError("failed to alter mnode since %s, input:%d cur:%d", terrstr(), alterReq.dnodeId, pDnode->data.dnodeId);
     return -1;
   } else {
     return mmAlter(pMgmt, &alterReq);

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

@@ -161,9 +161,9 @@ static int32_t mmOpen(SMgmtWrapper *pWrapper) {
   SMnodeOpt option = {0};
   if (!deployed) {
     dInfo("mnode start to deploy");
-    if (pWrapper->procType == DND_PROC_CHILD) {
+    // if (pWrapper->procType == DND_PROC_CHILD) {
       pWrapper->pDnode->data.dnodeId = 1;
-    }
+    // }
     mmBuildOptionForDeploy(pMgmt, &option);
   } else {
     dInfo("mnode start to open");

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

@@ -19,6 +19,7 @@
 static inline void mmSendRsp(SNodeMsg *pMsg, int32_t code) {
   SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
                  .ahandle = pMsg->rpcMsg.ahandle,
+                 .refId = pMsg->rpcMsg.refId,
                  .code = code,
                  .pCont = pMsg->pRsp,
                  .contLen = pMsg->rspLen};

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

@@ -19,6 +19,7 @@
 static inline void qmSendRsp(SNodeMsg *pMsg, int32_t code) {
   SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
                  .ahandle = pMsg->rpcMsg.ahandle,
+                 .refId = pMsg->rpcMsg.refId,
                  .code = code,
                  .pCont = pMsg->pRsp,
                  .contLen = pMsg->rspLen};

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

@@ -19,6 +19,7 @@
 static inline void smSendRsp(SNodeMsg *pMsg, int32_t code) {
   SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
                  .ahandle = pMsg->rpcMsg.ahandle,
+                 .refId = pMsg->rpcMsg.refId,
                  .code = code,
                  .pCont = pMsg->pRsp,
                  .contLen = pMsg->rspLen};
@@ -149,7 +150,7 @@ static FORCE_INLINE int32_t smGetSWTypeFromMsg(SRpcMsg *pMsg) {
 }
 
 int32_t smProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SSnodeMgmt   *pMgmt = pWrapper->pMgmt;
+  SSnodeMgmt *  pMgmt = pWrapper->pMgmt;
   SMultiWorker *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, 0);
   if (pWorker == NULL) {
     terrno = TSDB_CODE_INVALID_MSG;
@@ -162,7 +163,7 @@ int32_t smProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 }
 
 int32_t smProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SSnodeMgmt    *pMgmt = pWrapper->pMgmt;
+  SSnodeMgmt *   pMgmt = pWrapper->pMgmt;
   SSingleWorker *pWorker = &pMgmt->monitorWorker;
 
   dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
@@ -171,7 +172,7 @@ int32_t smProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 }
 
 int32_t smProcessUniqueMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SSnodeMgmt   *pMgmt = pWrapper->pMgmt;
+  SSnodeMgmt *  pMgmt = pWrapper->pMgmt;
   int32_t       index = smGetSWIdFromMsg(&pMsg->rpcMsg);
   SMultiWorker *pWorker = taosArrayGetP(pMgmt->uniqueWorkers, index);
   if (pWorker == NULL) {
@@ -185,7 +186,7 @@ int32_t smProcessUniqueMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 }
 
 int32_t smProcessSharedMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SSnodeMgmt    *pMgmt = pWrapper->pMgmt;
+  SSnodeMgmt *   pMgmt = pWrapper->pMgmt;
   SSingleWorker *pWorker = &pMgmt->sharedWorker;
 
   dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);

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

@@ -23,6 +23,7 @@
 static inline void vmSendRsp(SMgmtWrapper *pWrapper, SNodeMsg *pMsg, int32_t code) {
   SRpcMsg rsp = {.handle = pMsg->rpcMsg.handle,
                  .ahandle = pMsg->rpcMsg.ahandle,
+                 .refId = pMsg->rpcMsg.refId,
                  .code = code,
                  .pCont = pMsg->pRsp,
                  .contLen = pMsg->rspLen};
@@ -126,6 +127,7 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
     rsp.code = 0;
     rsp.handle = pRpc->handle;
     rsp.ahandle = pRpc->ahandle;
+    rsp.refId = pRpc->refId;
 
     int32_t code = vnodeProcessWriteReq(pVnode->pImpl, pRpc, version++, &rsp);
     tmsgSendRsp(&rsp);
@@ -134,13 +136,14 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
   // sync integration response
   for (int i = 0; i < taosArrayGetSize(pArray); i++) {
     SNodeMsg *pMsg;
-    SRpcMsg  *pRpc;
+    SRpcMsg * pRpc;
 
     pMsg = *(SNodeMsg **)taosArrayGet(pArray, i);
     pRpc = &pMsg->rpcMsg;
 
     rsp.ahandle = pRpc->ahandle;
     rsp.handle = pRpc->handle;
+    rsp.refId = pRpc->refId;
     rsp.pCont = NULL;
     rsp.contLen = 0;
 
@@ -172,11 +175,9 @@ static void vmProcessWriteQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 
 static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
-  SNodeMsg  *pMsg = NULL;
+  SNodeMsg * pMsg = NULL;
   SRpcMsg    rsp;
 
-  // static int64_t version = 0;
-
   for (int32_t i = 0; i < numOfMsgs; ++i) {
 #if 1
     // sync integration
@@ -208,6 +209,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
     if (pMsg->rpcMsg.handle != NULL && pMsg->rpcMsg.ahandle != NULL) {
       rsp.ahandle = pMsg->rpcMsg.ahandle;
       rsp.handle = pMsg->rpcMsg.handle;
+      rsp.refId = pMsg->rpcMsg.refId;
       tmsgSendRsp(&rsp);
     }
 #endif
@@ -216,7 +218,7 @@ static void vmProcessApplyQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 
 static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
-  SNodeMsg  *pMsg = NULL;
+  SNodeMsg * pMsg = NULL;
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
     taosGetQitem(qall, (void **)&pMsg);
@@ -229,7 +231,7 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf
 
 static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs) {
   SVnodeObj *pVnode = pInfo->ahandle;
-  SNodeMsg  *pMsg = NULL;
+  SNodeMsg * pMsg = NULL;
 
   for (int32_t i = 0; i < numOfMsgs; ++i) {
     taosGetQitem(qall, (void **)&pMsg);
@@ -246,7 +248,7 @@ static void vmProcessMergeQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numO
 }
 
 static int32_t vmPutNodeMsgToQueue(SVnodesMgmt *pMgmt, SNodeMsg *pMsg, EQueueType qtype) {
-  SRpcMsg  *pRpc = &pMsg->rpcMsg;
+  SRpcMsg * pRpc = &pMsg->rpcMsg;
   SMsgHead *pHead = pRpc->pCont;
   pHead->contLen = ntohl(pHead->contLen);
   pHead->vgId = ntohl(pHead->vgId);
@@ -315,7 +317,7 @@ int32_t vmProcessMergeMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 }
 
 int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SVnodesMgmt   *pMgmt = pWrapper->pMgmt;
+  SVnodesMgmt *  pMgmt = pWrapper->pMgmt;
   SSingleWorker *pWorker = &pMgmt->mgmtWorker;
   dTrace("msg:%p, will be written to vnode-mgmt queue, worker:%s", pMsg, pWorker->name);
   taosWriteQitem(pWorker->queue, pMsg);
@@ -323,7 +325,7 @@ int32_t vmProcessMgmtMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 }
 
 int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
-  SVnodesMgmt   *pMgmt = pWrapper->pMgmt;
+  SVnodesMgmt *  pMgmt = pWrapper->pMgmt;
   SSingleWorker *pWorker = &pMgmt->monitorWorker;
 
   dTrace("msg:%p, put into worker:%s", pMsg, pWorker->name);
@@ -333,7 +335,7 @@ int32_t vmProcessMonitorMsg(SMgmtWrapper *pWrapper, SNodeMsg *pMsg) {
 
 static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EQueueType qtype) {
   SVnodesMgmt *pMgmt = pWrapper->pMgmt;
-  SMsgHead    *pHead = pRpc->pCont;
+  SMsgHead *   pHead = pRpc->pCont;
 
   SVnodeObj *pVnode = vmAcquireVnode(pMgmt, pHead->vgId);
   if (pVnode == NULL) return -1;
@@ -346,6 +348,7 @@ static int32_t vmPutRpcMsgToQueue(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, EQueueT
   } else {
     dTrace("msg:%p, is created, type:%s", pMsg, TMSG_INFO(pRpc->msgType));
     pMsg->rpcMsg = *pRpc;
+    // if (pMsg->rpcMsg.handle != NULL) assert(pMsg->rpcMsg.refId != 0);
     switch (qtype) {
       case QUERY_QUEUE:
         dTrace("msg:%p, will be put into vnode-query queue", pMsg);

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

@@ -135,6 +135,7 @@ typedef struct {
   int32_t    failedTimes;
   void*      rpcHandle;
   void*      rpcAHandle;
+  int64_t    rpcRefId;
   void*      rpcRsp;
   int32_t    rpcRspLen;
   SArray*    redoLogs;

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

@@ -309,10 +309,10 @@ static int32_t mndProcessCreateFuncReq(SNodeMsg *pReq) {
     goto _OVER;
   }
 
-  if (createReq.pCode[0] == 0) {
-    terrno = TSDB_CODE_MND_INVALID_FUNC_CODE;
-    goto _OVER;
-  }
+ if (createReq.codeLen <= 1) {
+   terrno = TSDB_CODE_MND_INVALID_FUNC_CODE;
+   goto _OVER;
+ }
 
   if (createReq.bufSize <= 0 || createReq.bufSize > TSDB_FUNC_BUF_SIZE) {
     terrno = TSDB_CODE_MND_INVALID_FUNC_BUFSIZE;

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

@@ -193,9 +193,9 @@ TRANS_ENCODE_OVER:
 static SSdbRow *mndTransActionDecode(SSdbRaw *pRaw) {
   terrno = TSDB_CODE_OUT_OF_MEMORY;
 
-  SSdbRow     *pRow = NULL;
-  STrans      *pTrans = NULL;
-  char        *pData = NULL;
+  SSdbRow *    pRow = NULL;
+  STrans *     pTrans = NULL;
+  char *       pData = NULL;
   int32_t      dataLen = 0;
   int8_t       sver = 0;
   int32_t      redoLogNum = 0;
@@ -456,7 +456,7 @@ static int32_t mndTransActionUpdate(SSdb *pSdb, STrans *pOld, STrans *pNew) {
 }
 
 static STrans *mndAcquireTrans(SMnode *pMnode, int32_t transId) {
-  SSdb   *pSdb = pMnode->pSdb;
+  SSdb *  pSdb = pMnode->pSdb;
   STrans *pTrans = sdbAcquire(pSdb, SDB_TRANS, &transId);
   if (pTrans == NULL) {
     terrno = TSDB_CODE_MND_TRANS_NOT_EXIST;
@@ -484,6 +484,7 @@ STrans *mndTransCreate(SMnode *pMnode, ETrnPolicy policy, ETrnType type, const S
   pTrans->createdTime = taosGetTimestampMs();
   pTrans->rpcHandle = pReq->handle;
   pTrans->rpcAHandle = pReq->ahandle;
+  pTrans->rpcRefId = pReq->refId;
   pTrans->redoLogs = taosArrayInit(MND_TRANS_ARRAY_SIZE, sizeof(void *));
   pTrans->undoLogs = taosArrayInit(MND_TRANS_ARRAY_SIZE, sizeof(void *));
   pTrans->commitLogs = taosArrayInit(MND_TRANS_ARRAY_SIZE, sizeof(void *));
@@ -625,7 +626,7 @@ static int32_t mndCheckTransCanBeStartedInParallel(SMnode *pMnode, STrans *pNewT
   if (mndIsBasicTrans(pNewTrans)) return 0;
 
   STrans *pTrans = NULL;
-  void   *pIter = NULL;
+  void *  pIter = NULL;
   int32_t code = 0;
 
   while (1) {
@@ -703,6 +704,7 @@ int32_t mndTransPrepare(SMnode *pMnode, STrans *pTrans) {
 
   pNew->rpcHandle = pTrans->rpcHandle;
   pNew->rpcAHandle = pTrans->rpcAHandle;
+  pNew->rpcRefId = pTrans->rpcRefId;
   pNew->rpcRsp = pTrans->rpcRsp;
   pNew->rpcRspLen = pTrans->rpcRspLen;
   pTrans->rpcRsp = NULL;
@@ -767,6 +769,7 @@ static void mndTransSendRpcRsp(SMnode *pMnode, STrans *pTrans) {
     SRpcMsg rspMsg = {.handle = pTrans->rpcHandle,
                       .code = pTrans->code,
                       .ahandle = pTrans->rpcAHandle,
+                      .refId = pTrans->rpcRefId,
                       .pCont = rpcCont,
                       .contLen = pTrans->rpcRspLen};
     tmsgSendRsp(&rspMsg);
@@ -827,7 +830,7 @@ HANDLE_ACTION_RSP_OVER:
 }
 
 static int32_t mndTransExecuteLogs(SMnode *pMnode, SArray *pArray) {
-  SSdb   *pSdb = pMnode->pSdb;
+  SSdb *  pSdb = pMnode->pSdb;
   int32_t arraySize = taosArrayGetSize(pArray);
 
   if (arraySize == 0) return 0;
@@ -1202,11 +1205,11 @@ static int32_t mndKillTrans(SMnode *pMnode, STrans *pTrans) {
 }
 
 static int32_t mndProcessKillTransReq(SNodeMsg *pReq) {
-  SMnode       *pMnode = pReq->pNode;
+  SMnode *      pMnode = pReq->pNode;
   SKillTransReq killReq = {0};
   int32_t       code = -1;
-  SUserObj     *pUser = NULL;
-  STrans       *pTrans = NULL;
+  SUserObj *    pUser = NULL;
+  STrans *      pTrans = NULL;
 
   if (tDeserializeSKillTransReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &killReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
@@ -1246,7 +1249,7 @@ KILL_OVER:
 
 void mndTransPullup(SMnode *pMnode) {
   STrans *pTrans = NULL;
-  void   *pIter = NULL;
+  void *  pIter = NULL;
 
   while (1) {
     pIter = sdbFetch(pMnode->pSdb, SDB_TRANS, pIter, (void **)&pTrans);
@@ -1261,11 +1264,11 @@ void mndTransPullup(SMnode *pMnode) {
 
 static int32_t mndRetrieveTrans(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
   SMnode *pMnode = pReq->pNode;
-  SSdb   *pSdb = pMnode->pSdb;
+  SSdb *  pSdb = pMnode->pSdb;
   int32_t numOfRows = 0;
   STrans *pTrans = NULL;
   int32_t cols = 0;
-  char   *pWrite;
+  char *  pWrite;
 
   while (numOfRows < rows) {
     pShow->pIter = sdbFetch(pSdb, SDB_TRANS, pShow->pIter, (void **)&pTrans);

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

@@ -385,7 +385,11 @@ int32_t mndProcessMsg(SNodeMsg *pMsg) {
     terrno = code;
     mTrace("msg:%p, in progress, app:%p", pMsg, ahandle);
   } else if (code != 0) {
-    mError("msg:%p, failed to process since %s, app:%p", pMsg, terrstr(), ahandle);
+    if (terrno != TSDB_CODE_OPS_NOT_SUPPORT) {
+      mError("msg:%p, failed to process since %s, app:%p", pMsg, terrstr(), ahandle);
+    } else {
+      mTrace("msg:%p, failed to process since %s, app:%p", pMsg, terrstr(), ahandle);
+    }
   } else {
     mTrace("msg:%p, is processed, app:%p", pMsg, ahandle);
   }

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

@@ -22,17 +22,16 @@ class MndTestFunc : public ::testing::Test {
   void SetUp() override {}
   void TearDown() override {}
 
-  void SetCode(SCreateFuncReq* pReq, const char* pCode);
+  void SetCode(SCreateFuncReq* pReq, const char* pCode, int32_t size);
   void SetComment(SCreateFuncReq* pReq, const char* pComment);
 };
 
 Testbase MndTestFunc::test;
 
-void MndTestFunc::SetCode(SCreateFuncReq* pReq, const char* pCode) {
-  int32_t len = strlen(pCode);
-  pReq->pCode = (char*)taosMemoryCalloc(1, len + 1);
-  strcpy(pReq->pCode, pCode);
-  pReq->codeLen = len;
+void MndTestFunc::SetCode(SCreateFuncReq *pReq, const char *pCode, int32_t size) {
+  pReq->pCode = (char*)taosMemoryMalloc(size);
+  memcpy(pReq->pCode, pCode, size);
+  pReq->codeLen = size;
 }
 
 void MndTestFunc::SetComment(SCreateFuncReq* pReq, const char* pComment) {
@@ -79,7 +78,7 @@ TEST_F(MndTestFunc, 02_Create_Func) {
   {
     SCreateFuncReq createReq = {0};
     strcpy(createReq.name, "f1");
-    SetCode(&createReq, "");
+    SetCode(&createReq, "", 1);
     SetComment(&createReq, "comment1");
 
     int32_t contLen = tSerializeSCreateFuncReq(NULL, 0, &createReq);
@@ -95,7 +94,7 @@ TEST_F(MndTestFunc, 02_Create_Func) {
   {
     SCreateFuncReq createReq = {0};
     strcpy(createReq.name, "f1");
-    SetCode(&createReq, "code1");
+    SetCode(&createReq, "code1", 6);
     SetComment(&createReq, "comment1");
 
     int32_t contLen = tSerializeSCreateFuncReq(NULL, 0, &createReq);
@@ -111,7 +110,7 @@ TEST_F(MndTestFunc, 02_Create_Func) {
   {
     SCreateFuncReq createReq = {0};
     strcpy(createReq.name, "f1");
-    SetCode(&createReq, "code1");
+    SetCode(&createReq, "code1", 6);
     SetComment(&createReq, "comment1");
     createReq.bufSize = TSDB_FUNC_BUF_SIZE + 1;
 
@@ -128,7 +127,7 @@ TEST_F(MndTestFunc, 02_Create_Func) {
   for (int32_t i = 0; i < 3; ++i) {
     SCreateFuncReq createReq = {0};
     strcpy(createReq.name, "f1");
-    SetCode(&createReq, "code1");
+    SetCode(&createReq, "code1", 6);
     SetComment(&createReq, "comment1");
     createReq.bufSize = TSDB_FUNC_BUF_SIZE + 1;
     createReq.igExists = 0;
@@ -253,7 +252,7 @@ TEST_F(MndTestFunc, 03_Retrieve_Func) {
     createReq.outputLen = 24;
     createReq.bufSize = 6;
     createReq.signature = 18;
-    SetCode(&createReq, "code2");
+    SetCode(&createReq, "code2", 6);
     SetComment(&createReq, "comment2");
 
     int32_t contLen = tSerializeSCreateFuncReq(NULL, 0, &createReq);
@@ -439,3 +438,70 @@ TEST_F(MndTestFunc, 04_Drop_Func) {
   test.SendShowReq(TSDB_MGMT_TABLE_FUNC, "user_functions", "");
   EXPECT_EQ(test.GetShowRows(), 1);
 }
+
+TEST_F(MndTestFunc, 05_Actual_code) {
+  {
+    SCreateFuncReq createReq = {0};
+    strcpy(createReq.name, "udf1");
+    char code[300] = {0};
+    for (int32_t i = 0; i < sizeof(code); ++i) {
+      code[i] = (i) % 20;
+    }
+    SetCode(&createReq, code, 300);
+    SetComment(&createReq, "comment1");
+    createReq.bufSize = 8;
+    createReq.igExists = 0;
+    createReq.funcType = 1;
+    createReq.scriptType = 2;
+    createReq.outputType = TSDB_DATA_TYPE_SMALLINT;
+    createReq.outputLen = 12;
+    createReq.bufSize = 4;
+    createReq.signature = 5;
+
+    int32_t contLen = tSerializeSCreateFuncReq(NULL, 0, &createReq);
+    void*   pReq = rpcMallocCont(contLen);
+    tSerializeSCreateFuncReq(pReq, contLen, &createReq);
+    tFreeSCreateFuncReq(&createReq);
+
+    SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_FUNC, pReq, contLen);
+    ASSERT_NE(pRsp, nullptr);
+    ASSERT_EQ(pRsp->code, 0);
+  }
+
+  {
+    SRetrieveFuncReq retrieveReq = {0};
+    retrieveReq.numOfFuncs = 1;
+    retrieveReq.pFuncNames = taosArrayInit(1, TSDB_FUNC_NAME_LEN);
+    taosArrayPush(retrieveReq.pFuncNames, "udf1");
+
+    int32_t contLen = tSerializeSRetrieveFuncReq(NULL, 0, &retrieveReq);
+    void*   pReq = rpcMallocCont(contLen);
+    tSerializeSRetrieveFuncReq(pReq, contLen, &retrieveReq);
+    tFreeSRetrieveFuncReq(&retrieveReq);
+
+    SRpcMsg* pRsp = test.SendReq(TDMT_MND_RETRIEVE_FUNC, pReq, contLen);
+    ASSERT_NE(pRsp, nullptr);
+    ASSERT_EQ(pRsp->code, 0);
+
+    SRetrieveFuncRsp retrieveRsp = {0};
+    tDeserializeSRetrieveFuncRsp(pRsp->pCont, pRsp->contLen, &retrieveRsp);
+    EXPECT_EQ(retrieveRsp.numOfFuncs, 1);
+    EXPECT_EQ(retrieveRsp.numOfFuncs, (int32_t)taosArrayGetSize(retrieveRsp.pFuncInfos));
+
+    SFuncInfo* pFuncInfo = (SFuncInfo*)taosArrayGet(retrieveRsp.pFuncInfos, 0);
+
+    EXPECT_STREQ(pFuncInfo->name, "udf1");
+    EXPECT_EQ(pFuncInfo->funcType, 1);
+    EXPECT_EQ(pFuncInfo->scriptType, 2);
+    EXPECT_EQ(pFuncInfo->outputType, TSDB_DATA_TYPE_SMALLINT);
+    EXPECT_EQ(pFuncInfo->outputLen, 12);
+    EXPECT_EQ(pFuncInfo->bufSize, 4);
+    EXPECT_EQ(pFuncInfo->signature, 5);
+    EXPECT_STREQ("comment1", pFuncInfo->pComment);
+    for (int32_t i = 0; i < 300; ++i) {
+        EXPECT_EQ(pFuncInfo->pCode[i], (i) % 20);
+    }
+    tFreeSRetrieveFuncRsp(&retrieveRsp);
+  }
+
+}
\ No newline at end of file

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

@@ -1510,7 +1510,6 @@ static int32_t tsdbGetTSmaDataImpl(STsdb *pTsdb, char *pData, int64_t indexUid,
   }
 
   STSma *pTSma = pItem->pSma;
-
 #endif
 
   STSmaReadH tReadH = {0};

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

@@ -22,21 +22,21 @@ int vnodeQueryOpen(SVnode *pVnode) {
 void vnodeQueryClose(SVnode *pVnode) { qWorkerDestroy((void **)&pVnode->pQuery); }
 
 int vnodeGetTableMeta(SVnode *pVnode, SRpcMsg *pMsg) {
-  STbCfg         *pTbCfg = NULL;
-  STbCfg         *pStbCfg = NULL;
+  STbCfg *        pTbCfg = NULL;
+  STbCfg *        pStbCfg = NULL;
   tb_uid_t        uid;
   int32_t         nCols;
   int32_t         nTagCols;
   SSchemaWrapper *pSW = NULL;
-  STableMetaRsp  *pTbMetaMsg = NULL;
+  STableMetaRsp * pTbMetaMsg = NULL;
   STableMetaRsp   metaRsp = {0};
-  SSchema        *pTagSchema;
+  SSchema *       pTagSchema;
   SRpcMsg         rpcMsg;
   int             msgLen = 0;
   int32_t         code = 0;
   char            tableFName[TSDB_TABLE_FNAME_LEN];
   int32_t         rspLen = 0;
-  void           *pRsp = NULL;
+  void *          pRsp = NULL;
 
   STableInfoReq infoReq = {0};
   if (tDeserializeSTableInfoReq(pMsg->pCont, pMsg->contLen, &infoReq) != 0) {
@@ -142,6 +142,7 @@ _exit:
 
   rpcMsg.handle = pMsg->handle;
   rpcMsg.ahandle = pMsg->ahandle;
+  rpcMsg.refId = pMsg->refId;
   rpcMsg.pCont = pRsp;
   rpcMsg.contLen = rspLen;
   rpcMsg.code = code;

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

@@ -113,7 +113,7 @@ void vnodeSyncCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cb
         pFsm, cbMeta.index, cbMeta.isWeak, cbMeta.code, cbMeta.state, syncUtilState2String(cbMeta.state), beginIndex);
     syncRpcMsgLog2(logBuf, (SRpcMsg *)pMsg);
 
-    SVnode       *pVnode = (SVnode *)(pFsm->data);
+    SVnode *      pVnode = (SVnode *)(pFsm->data);
     SyncApplyMsg *pSyncApplyMsg = syncApplyMsgBuild2(pMsg, pVnode->config.vgId, &cbMeta);
     SRpcMsg       applyMsg;
     syncApplyMsg2RpcMsg(pSyncApplyMsg, &applyMsg);
@@ -133,6 +133,7 @@ void vnodeSyncCommitCb(struct SSyncFSM *pFsm, const SRpcMsg *pMsg, SFsmCbMeta cb
     if (ret == 1 && cbMeta.state == TAOS_SYNC_STATE_LEADER) {
       applyMsg.handle = saveRpcMsg.handle;
       applyMsg.ahandle = saveRpcMsg.ahandle;
+      applyMsg.refId = saveRpcMsg.refId;
     } else {
       applyMsg.handle = NULL;
       applyMsg.ahandle = NULL;

source/libs/command/src/explain.c

@@ -344,11 +344,6 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
       EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
       EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pTagScanNode->node.pOutputDataBlockDesc->totalRowSize);
       EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
-      EXPLAIN_ROW_APPEND(EXPLAIN_LOOPS_FORMAT, pTagScanNode->count);
-      if (pTagScanNode->reverse) {
-        EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
-        EXPLAIN_ROW_APPEND(EXPLAIN_REVERSE_FORMAT, pTagScanNode->reverse);
-      }
       EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT);
       EXPLAIN_ROW_END();
       QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level));
@@ -361,10 +356,6 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
         EXPLAIN_ROW_END();
         QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
         
-        EXPLAIN_ROW_NEW(level + 1, EXPLAIN_ORDER_FORMAT, EXPLAIN_ORDER_STRING(pTagScanNode->order));
-        EXPLAIN_ROW_END();
-        QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
-
         if (pResNode->pExecInfo) {
           QRY_ERR_RET(qExplainBufAppendVerboseExecInfo(pResNode->pExecInfo, tbuf, &tlen));
           if (tlen) {
@@ -388,11 +379,6 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
       EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
       EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pTblScanNode->scan.node.pOutputDataBlockDesc->totalRowSize);
       EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
-      EXPLAIN_ROW_APPEND(EXPLAIN_LOOPS_FORMAT, pTblScanNode->scan.count);
-      if (pTblScanNode->scan.reverse) {
-        EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
-        EXPLAIN_ROW_APPEND(EXPLAIN_REVERSE_FORMAT, pTblScanNode->scan.reverse);
-      }
       EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT);      
       EXPLAIN_ROW_END();
       QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level));
@@ -405,10 +391,6 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
         EXPLAIN_ROW_END();
         QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
 
-        EXPLAIN_ROW_NEW(level + 1, EXPLAIN_ORDER_FORMAT, EXPLAIN_ORDER_STRING(pTblScanNode->scan.order));
-        EXPLAIN_ROW_END();
-        QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
-        
         EXPLAIN_ROW_NEW(level + 1, EXPLAIN_TIMERANGE_FORMAT, pTblScanNode->scanRange.skey, pTblScanNode->scanRange.ekey);
         EXPLAIN_ROW_END();
         QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
@@ -434,11 +416,6 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
       EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
       EXPLAIN_ROW_APPEND(EXPLAIN_WIDTH_FORMAT, pSTblScanNode->scan.node.pOutputDataBlockDesc->totalRowSize);
       EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
-      EXPLAIN_ROW_APPEND(EXPLAIN_LOOPS_FORMAT, pSTblScanNode->scan.count);
-      if (pSTblScanNode->scan.reverse) {
-        EXPLAIN_ROW_APPEND(EXPLAIN_BLANK_FORMAT);
-        EXPLAIN_ROW_APPEND(EXPLAIN_REVERSE_FORMAT, pSTblScanNode->scan.reverse);
-      }
       EXPLAIN_ROW_APPEND(EXPLAIN_RIGHT_PARENTHESIS_FORMAT);      
       EXPLAIN_ROW_END();
       QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level));
@@ -451,10 +428,6 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
         EXPLAIN_ROW_END();
         QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));
 
-        EXPLAIN_ROW_NEW(level + 1, EXPLAIN_ORDER_FORMAT, EXPLAIN_ORDER_STRING(pSTblScanNode->scan.order));
-        EXPLAIN_ROW_END();
-        QRY_ERR_RET(qExplainResAppendRow(ctx, tbuf, tlen, level + 1));   
-
         if (pSTblScanNode->scan.node.pConditions) {
           EXPLAIN_ROW_NEW(level + 1, EXPLAIN_FILTER_FORMAT);
           QRY_ERR_RET(nodesNodeToSQL(pSTblScanNode->scan.node.pConditions, tbuf + VARSTR_HEADER_SIZE, TSDB_EXPLAIN_RESULT_ROW_SIZE, &tlen)); 

source/libs/function/inc/builtinsimpl.h

@@ -40,6 +40,11 @@ bool getMinmaxFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 int32_t minFunction(SqlFunctionCtx* pCtx);
 int32_t maxFunction(SqlFunctionCtx *pCtx);
 
+bool getAvgFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
+bool avgFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
+int32_t avgFunction(SqlFunctionCtx* pCtx);
+int32_t avgFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId);
+
 bool getStddevFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool stddevFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
 int32_t stddevFunction(SqlFunctionCtx* pCtx);

source/libs/function/inc/tudf.h

@@ -29,29 +29,32 @@ extern "C" {
 
 #define UDF_LISTEN_PIPE_NAME_LEN 32
 #define UDF_LISTEN_PIPE_NAME_PREFIX "udfd.sock."
+#define UDF_DNODE_ID_ENV_NAME "DNODE_ID"
 
 //======================================================================================
 //begin API to taosd and qworker
 
 enum {
   UDFC_CODE_STOPPING = -1,
-  UDFC_CODE_PIPE_READ_ERR = -3,
+  UDFC_CODE_PIPE_READ_ERR = -2,
+  UDFC_CODE_CONNECT_PIPE_ERR = -3,
+  UDFC_CODE_LOAD_UDF_FAILURE = -4,
+  UDFC_CODE_INVALID_STATE = -5
 };
 
-typedef void *UdfcHandle;
 typedef void *UdfcFuncHandle;
 
 /**
  * create udfd proxy, called once in process that call setupUdf/callUdfxxx/teardownUdf
  * @return error code
  */
-int32_t udfcOpen(int32_t dnodeId, UdfcHandle* proxyHandle);
+int32_t udfcOpen();
 
 /**
  * destroy udfd proxy
  * @return error code
  */
-int32_t udfcClose(UdfcHandle proxyhandle);
+int32_t udfcClose();
 
 
 /**
@@ -60,7 +63,7 @@ int32_t udfcClose(UdfcHandle proxyhandle);
  * @param handle, out
  * @return error code
  */
-int32_t setupUdf(UdfcHandle proxyHandle, char udfName[], SEpSet *epSet, UdfcFuncHandle *handle);
+int32_t setupUdf(char udfName[], UdfcFuncHandle *handle);
 
 typedef struct SUdfColumnMeta {
   int16_t type;

source/libs/function/inc/tudfInt.h

@@ -39,7 +39,6 @@ enum {
 
 typedef struct SUdfSetupRequest {
   char udfName[TSDB_FUNC_NAME_LEN];
-  SEpSet epSet;
 } SUdfSetupRequest;
 
 typedef struct SUdfSetupResponse {
@@ -112,6 +111,7 @@ void freeUdfDataDataBlock(SUdfDataBlock *block);
 int32_t convertDataBlockToUdfDataBlock(SSDataBlock *block, SUdfDataBlock *udfBlock);
 int32_t convertUdfColumnToDataBlock(SUdfColumn *udfCol, SSDataBlock *block);
 
+int32_t getUdfdPipeName(char* pipeName, int32_t size);
 #ifdef __cplusplus
 }
 #endif

source/libs/function/src/builtins.c

@@ -104,7 +104,7 @@ static int32_t translateCount(SFunctionNode* pFunc, char* pErrBuf, int32_t len)
   if (1 != LIST_LENGTH(pFunc->pParameterList)) {
     return invaildFuncParaNumErrMsg(pErrBuf, len, pFunc->functionName);
   }
-  pFunc->node.resType = (SDataType){.bytes = sizeof(int64_t), .type = TSDB_DATA_TYPE_BIGINT};
+  pFunc->node.resType = (SDataType){.bytes = tDataTypes[TSDB_DATA_TYPE_BIGINT].bytes, .type = TSDB_DATA_TYPE_BIGINT};
   return TSDB_CODE_SUCCESS;
 }
 
@@ -355,7 +355,7 @@ static int32_t translateToIso8601(SFunctionNode* pFunc, char* pErrBuf, int32_t l
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
-  pFunc->node.resType = (SDataType) { .bytes = 64, .type = TSDB_DATA_TYPE_BINARY};
+  pFunc->node.resType = (SDataType) { .bytes = 24, .type = TSDB_DATA_TYPE_BINARY};
   return TSDB_CODE_SUCCESS;
 }
 
@@ -479,6 +479,16 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .processFunc  = stddevFunction,
     .finalizeFunc = stddevFinalize
   },
+  {
+    .name = "avg",
+    .type = FUNCTION_TYPE_AVG,
+    .classification = FUNC_MGT_AGG_FUNC,
+    .translateFunc = translateInNumOutDou,
+    .getEnvFunc   = getAvgFuncEnv,
+    .initFunc     = avgFunctionSetup,
+    .processFunc  = avgFunction,
+    .finalizeFunc = avgFinalize
+  },
   {
     .name = "percentile",
     .type = FUNCTION_TYPE_PERCENTILE,

source/libs/function/src/builtinsimpl.c

@@ -20,6 +20,59 @@
 #include "tdatablock.h"
 #include "tpercentile.h"
 
+typedef struct SSumRes {
+  union {
+    int64_t  isum;
+    uint64_t usum;
+    double   dsum;
+  };
+} SSumRes;
+
+typedef struct SAvgRes {
+  double  result;
+  SSumRes sum;
+  int64_t count;
+} SAvgRes;
+
+typedef struct STopBotResItem {
+  SVariant v;
+  uint64_t uid;        // it is a table uid, used to extract tag data during building of the final result for the tag data
+  struct {
+   int32_t pageId;
+   int32_t offset;
+  } tuplePos;          // tuple data of this chosen row
+} STopBotResItem;
+
+typedef struct STopBotRes {
+  int32_t         pageId;
+//  int32_t   num;
+  STopBotResItem *pItems;
+} STopBotRes;
+
+typedef struct SStddevRes {
+  double  result;
+  int64_t count;
+  union  {double  quadraticDSum; int64_t quadraticISum;};
+  union  {double  dsum; int64_t isum;};
+} SStddevRes;
+
+typedef struct SPercentileInfo {
+  double      result;
+  tMemBucket *pMemBucket;
+  int32_t     stage;
+  double      minval;
+  double      maxval;
+  int64_t     numOfElems;
+} SPercentileInfo;
+
+typedef struct SDiffInfo {
+  bool  hasPrev;
+  bool  includeNull;
+  bool  ignoreNegative;
+  bool  firstOutput;
+  union { int64_t i64; double d64;} prev;
+} SDiffInfo;
+
 #define SET_VAL(_info, numOfElem, res)  \
   do {                                  \
     if ((numOfElem) <= 0) {             \
@@ -28,13 +81,50 @@
     (_info)->numOfRes = (res);          \
   } while (0)
 
-typedef struct SSumRes {
-  union {
-    int64_t  isum;
-    uint64_t usum;
-    double   dsum;
-  };
-} SSumRes;
+#define GET_TS_LIST(x)    ((TSKEY*)((x)->ptsList))
+#define GET_TS_DATA(x, y) (GET_TS_LIST(x)[(y)])
+
+#define DO_UPDATE_TAG_COLUMNS_WITHOUT_TS(ctx)                      \
+  do {                                                             \
+    for (int32_t _i = 0; _i < (ctx)->tagInfo.numOfTagCols; ++_i) { \
+      SqlFunctionCtx *__ctx = (ctx)->tagInfo.pTagCtxList[_i];      \
+      __ctx->fpSet.process(__ctx);                                 \
+    }                                                              \
+  } while (0);
+
+#define DO_UPDATE_SUBSID_RES(ctx, ts)                                 \
+  do {                                                                \
+    for (int32_t _i = 0; _i < (ctx)->subsidiaryRes.numOfCols; ++_i) { \
+      SqlFunctionCtx *__ctx = (ctx)->subsidiaryRes.pCtx[_i];          \
+      if (__ctx->functionId == FUNCTION_TS_DUMMY) {                   \
+        __ctx->tag.i = (ts);                                          \
+        __ctx->tag.nType = TSDB_DATA_TYPE_BIGINT;                     \
+      }                                                               \
+      __ctx->fpSet.process(__ctx);                                    \
+    }                                                                 \
+  } while (0)
+
+#define UPDATE_DATA(ctx, left, right, num, sign, _ts) \
+  do {                                                \
+    if (((left) < (right)) ^ (sign)) {                \
+      (left) = (right);                               \
+      DO_UPDATE_SUBSID_RES(ctx, _ts);                 \
+      (num) += 1;                                     \
+    }                                                 \
+  } while (0)
+
+#define LOOPCHECK_N(val, _col, ctx, _t, _nrow, _start, sign, num)        \
+  do {                                                                   \
+    _t *d = (_t *)((_col)->pData);                                       \
+    for (int32_t i = (_start); i < (_nrow) + (_start); ++i) {            \
+      if (((_col)->hasNull) && colDataIsNull_f((_col)->nullbitmap, i)) { \
+        continue;                                                        \
+      }                                                                  \
+      TSKEY ts = (ctx)->ptsList != NULL ? GET_TS_DATA(ctx, i) : 0;       \
+      UPDATE_DATA(ctx, val, d[i], num, sign, ts);                        \
+    }                                                                    \
+  } while (0)
+
 
 bool functionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo) {
   if (pResultInfo->initialized) {
@@ -135,7 +225,7 @@ int32_t sumFunction(SqlFunctionCtx *pCtx) {
   int32_t type = pInput->pData[0]->info.type;
 
   SSumRes* pSumRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
-  
+
   if (pInput->colDataAggIsSet) {
     numOfElem = pInput->numOfRows - pAgg->numOfNull;
     ASSERT(numOfElem >= 0);
@@ -190,6 +280,145 @@ bool getSumFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
   return true;
 }
 
+bool getAvgFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
+  pEnv->calcMemSize = sizeof(double);
+  return true;
+}
+
+bool avgFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo) {
+  if (!functionSetup(pCtx, pResultInfo)) {
+    return false;
+  }
+
+  SAvgRes* pRes = GET_ROWCELL_INTERBUF(pResultInfo);
+  memset(pRes, 0, sizeof(SAvgRes));
+  return true;
+}
+
+int32_t avgFunction(SqlFunctionCtx* pCtx) {
+  int32_t numOfElem = 0;
+
+  // Only the pre-computing information loaded and actual data does not loaded
+  SInputColumnInfoData* pInput = &pCtx->input;
+  int32_t               type = pInput->pData[0]->info.type;
+
+  SAvgRes* pAvgRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
+
+  // computing based on the true data block
+  SColumnInfoData* pCol = pInput->pData[0];
+
+  int32_t start = pInput->startRowIndex;
+  int32_t numOfRows = pInput->numOfRows;
+
+  switch (type) {
+    case TSDB_DATA_TYPE_TINYINT: {
+        int8_t* plist = (int8_t*)pCol->pData;
+        for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
+          if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
+            continue;
+          }
+
+          numOfElem += 1;
+          pAvgRes->count += 1;
+          pAvgRes->sum.isum += plist[i];
+        }
+
+        break;
+      }
+
+      case TSDB_DATA_TYPE_SMALLINT: {
+      int16_t* plist = (int16_t*)pCol->pData;
+      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
+        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
+          continue;
+        }
+
+        numOfElem += 1;
+        pAvgRes->count += 1;
+        pAvgRes->sum.isum += plist[i];
+      }
+      break;
+    }
+
+    case TSDB_DATA_TYPE_INT: {
+      int32_t* plist = (int32_t*)pCol->pData;
+      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
+        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
+          continue;
+        }
+
+        numOfElem += 1;
+        pAvgRes->count += 1;
+        pAvgRes->sum.isum += plist[i];
+      }
+
+      break;
+    }
+
+    case TSDB_DATA_TYPE_BIGINT: {
+      int64_t* plist = (int64_t*)pCol->pData;
+      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
+        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
+          continue;
+        }
+
+        numOfElem += 1;
+        pAvgRes->count += 1;
+        pAvgRes->sum.isum += plist[i];
+      }
+      break;
+    }
+
+    case TSDB_DATA_TYPE_FLOAT: {
+      float* plist = (float*)pCol->pData;
+      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
+        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
+          continue;
+        }
+
+        numOfElem += 1;
+        pAvgRes->count += 1;
+        pAvgRes->sum.dsum += plist[i];
+      }
+      break;
+    }
+
+    case TSDB_DATA_TYPE_DOUBLE: {
+      double* plist = (double*)pCol->pData;
+      for (int32_t i = start; i < numOfRows + pInput->startRowIndex; ++i) {
+        if (pCol->hasNull && colDataIsNull_f(pCol->nullbitmap, i)) {
+          continue;
+        }
+
+        numOfElem += 1;
+        pAvgRes->count += 1;
+        pAvgRes->sum.dsum += plist[i];
+      }
+      break;
+    }
+
+    default:
+      break;
+  }
+
+  // data in the check operation are all null, not output
+  SET_VAL(GET_RES_INFO(pCtx), numOfElem, 1);
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t avgFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId) {
+  SInputColumnInfoData* pInput = &pCtx->input;
+  int32_t type = pInput->pData[0]->info.type;
+  SAvgRes* pAvgRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
+  if (IS_INTEGER_TYPE(type)) {
+    pAvgRes->result = pAvgRes->sum.isum / ((double) pAvgRes->count);
+  } else {
+    pAvgRes->result = pAvgRes->sum.dsum / ((double) pAvgRes->count);
+  }
+
+  return functionFinalize(pCtx, pBlock, slotId);
+}
+
 EFuncDataRequired statisDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow){
   return FUNC_DATA_REQUIRED_STATIS_LOAD;
 }
@@ -292,49 +521,6 @@ bool getMinmaxFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
   return true;
 }
 
-#define GET_TS_LIST(x)    ((TSKEY*)((x)->ptsList))
-#define GET_TS_DATA(x, y) (GET_TS_LIST(x)[(y)])
-
-#define DO_UPDATE_TAG_COLUMNS_WITHOUT_TS(ctx)                      \
-  do {                                                             \
-    for (int32_t _i = 0; _i < (ctx)->tagInfo.numOfTagCols; ++_i) { \
-      SqlFunctionCtx *__ctx = (ctx)->tagInfo.pTagCtxList[_i];      \
-      __ctx->fpSet.process(__ctx);                                 \
-    }                                                              \
-  } while (0);
-
-#define DO_UPDATE_SUBSID_RES(ctx, ts)                                 \
-  do {                                                                \
-    for (int32_t _i = 0; _i < (ctx)->subsidiaryRes.numOfCols; ++_i) { \
-      SqlFunctionCtx *__ctx = (ctx)->subsidiaryRes.pCtx[_i];          \
-      if (__ctx->functionId == FUNCTION_TS_DUMMY) {                   \
-        __ctx->tag.i = (ts);                                          \
-        __ctx->tag.nType = TSDB_DATA_TYPE_BIGINT;                     \
-      }                                                               \
-      __ctx->fpSet.process(__ctx);                                    \
-    }                                                                 \
-  } while (0)
-
-#define UPDATE_DATA(ctx, left, right, num, sign, _ts) \
-  do {                                                \
-    if (((left) < (right)) ^ (sign)) {                \
-      (left) = (right);                               \
-      DO_UPDATE_SUBSID_RES(ctx, _ts);                 \
-      (num) += 1;                                     \
-    }                                                 \
-  } while (0)
-
-#define LOOPCHECK_N(val, _col, ctx, _t, _nrow, _start, sign, num)        \
-  do {                                                                   \
-    _t *d = (_t *)((_col)->pData);                                       \
-    for (int32_t i = (_start); i < (_nrow) + (_start); ++i) {            \
-      if (((_col)->hasNull) && colDataIsNull_f((_col)->nullbitmap, i)) { \
-        continue;                                                        \
-      }                                                                  \
-      TSKEY ts = (ctx)->ptsList != NULL ? GET_TS_DATA(ctx, i) : 0;       \
-      UPDATE_DATA(ctx, val, d[i], num, sign, ts);                        \
-    }                                                                    \
-  } while (0)
 
 int32_t doMinMaxHelper(SqlFunctionCtx *pCtx, int32_t isMinFunc) {
   int32_t numOfElems = 0;
@@ -479,13 +665,6 @@ int32_t maxFunction(SqlFunctionCtx *pCtx) {
   return TSDB_CODE_SUCCESS;
 }
 
-typedef struct SStddevRes {
-  double  result;
-  int64_t count;
-  union  {double  quadraticDSum; int64_t quadraticISum;};
-  union  {double  dsum; int64_t isum;};
-} SStddevRes;
-
 bool getStddevFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
   pEnv->calcMemSize = sizeof(SStddevRes);
   return true;
@@ -588,8 +767,8 @@ int32_t stddevFunction(SqlFunctionCtx* pCtx) {
 
         numOfElem += 1;
         pStddevRes->count += 1;
-        pStddevRes->isum += plist[i];
-        pStddevRes->quadraticISum += plist[i] * plist[i];
+        pStddevRes->dsum += plist[i];
+        pStddevRes->quadraticDSum += plist[i] * plist[i];
       }
       break;
     }
@@ -603,8 +782,8 @@ int32_t stddevFunction(SqlFunctionCtx* pCtx) {
 
         numOfElem += 1;
         pStddevRes->count += 1;
-        pStddevRes->isum += plist[i];
-        pStddevRes->quadraticISum += plist[i] * plist[i];
+        pStddevRes->dsum += plist[i];
+        pStddevRes->quadraticDSum += plist[i] * plist[i];
       }
       break;
     }
@@ -619,21 +798,21 @@ int32_t stddevFunction(SqlFunctionCtx* pCtx) {
 }
 
 int32_t stddevFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId) {
+  SInputColumnInfoData* pInput = &pCtx->input;
+  int32_t type = pInput->pData[0]->info.type;
   SStddevRes* pStddevRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
-  double avg = pStddevRes->isum / ((double) pStddevRes->count);
-  pStddevRes->result = sqrt(pStddevRes->quadraticISum/((double)pStddevRes->count) - avg*avg);
+  double avg;
+  if (IS_INTEGER_TYPE(type)) {
+    avg = pStddevRes->isum / ((double) pStddevRes->count);
+    pStddevRes->result = sqrt(pStddevRes->quadraticISum/((double)pStddevRes->count) - avg*avg);
+  } else {
+    avg = pStddevRes->dsum / ((double) pStddevRes->count);
+    pStddevRes->result = sqrt(pStddevRes->quadraticDSum/((double)pStddevRes->count) - avg*avg);
+  }
+
   return functionFinalize(pCtx, pBlock, slotId);
 }
 
-typedef struct SPercentileInfo {
-  double      result;
-  tMemBucket *pMemBucket;
-  int32_t     stage;
-  double      minval;
-  double      maxval;
-  int64_t     numOfElems;
-} SPercentileInfo;
-
 bool getPercentileFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
   pEnv->calcMemSize = sizeof(SPercentileInfo);
   return true;
@@ -928,14 +1107,6 @@ int32_t lastFunction(SqlFunctionCtx *pCtx) {
   return TSDB_CODE_SUCCESS;
 }
 
-typedef struct SDiffInfo {
-  bool  hasPrev;
-  bool  includeNull;
-  bool  ignoreNegative;
-  bool  firstOutput;
-  union { int64_t i64; double d64;} prev;
-} SDiffInfo;
-
 bool getDiffFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
   pEnv->calcMemSize = sizeof(SDiffInfo);
   return true;
@@ -1168,21 +1339,6 @@ int32_t diffFunction(SqlFunctionCtx *pCtx) {
   }
 }
 
-typedef struct STopBotResItem {
-  SVariant v;
-  uint64_t uid;        // it is a table uid, used to extract tag data during building of the final result for the tag data
-  struct {
-   int32_t pageId;
-   int32_t offset;
-  } tuplePos;          // tuple data of this chosen row
-} STopBotResItem;
-
-typedef struct STopBotRes {
-  int32_t         pageId;
-//  int32_t   num;
-  STopBotResItem *pItems;
-} STopBotRes;
-
 bool getTopBotFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
   SValueNode* pkNode = (SValueNode*) nodesListGetNode(pFunc->pParameterList, 1);
   pEnv->calcMemSize = sizeof(STopBotRes) + pkNode->datum.i * sizeof(STopBotResItem);
@@ -1335,4 +1491,4 @@ int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId
   return pEntryInfo->numOfRes;
 
 //  return functionFinalize(pCtx, pBlock, slotId);
-}
+}
\ No newline at end of file

source/libs/function/src/tudf.c

@@ -124,7 +124,7 @@ enum {
 
 int64_t gUdfTaskSeqNum = 0;
 typedef struct SUdfdProxy {
-  int32_t dnodeId;
+  char udfdPipeName[UDF_LISTEN_PIPE_NAME_LEN];
   uv_barrier_t gUdfInitBarrier;
 
   uv_loop_t gUdfdLoop;
@@ -137,11 +137,11 @@ typedef struct SUdfdProxy {
   int8_t gUdfcState;
   QUEUE gUdfTaskQueue;
   QUEUE gUvProcTaskQueue;
-  //  int8_t gUdfcState = UDFC_STATE_INITAL;
-  //  QUEUE gUdfTaskQueue = {0};
-  //  QUEUE gUvProcTaskQueue = {0};
+
+  int8_t initialized;
 } SUdfdProxy;
 
+SUdfdProxy gUdfdProxy = {0};
 
 typedef struct SUdfUvSession {
   SUdfdProxy *udfc;
@@ -209,19 +209,27 @@ enum {
   UDFC_STATE_STARTNG, // starting after udfcOpen
   UDFC_STATE_READY, // started and begin to receive quests
   UDFC_STATE_STOPPING, // stopping after udfcClose
-  UDFC_STATUS_FINAL, // stopped
 };
 
+int32_t getUdfdPipeName(char* pipeName, int32_t size) {
+  char    dnodeId[8] = {0};
+  size_t  dnodeIdSize;
+  int32_t err = uv_os_getenv(UDF_DNODE_ID_ENV_NAME, dnodeId, &dnodeIdSize);
+  if (err != 0) {
+    dnodeId[0] = '1';
+  }
+  snprintf(pipeName, size, "%s%s", UDF_LISTEN_PIPE_NAME_PREFIX, dnodeId);
+  return 0;
+}
+
 int32_t encodeUdfSetupRequest(void **buf, const SUdfSetupRequest *setup) {
   int32_t len = 0;
   len += taosEncodeBinary(buf, setup->udfName, TSDB_FUNC_NAME_LEN);
-  len += taosEncodeSEpSet(buf, &setup->epSet);
   return len;
 }
 
 void* decodeUdfSetupRequest(const void* buf, SUdfSetupRequest *request) {
   buf = taosDecodeBinaryTo(buf, request->udfName, TSDB_FUNC_NAME_LEN);
-  buf = taosDecodeSEpSet((void*)buf, &request->epSet);
   return (void*)buf;
 }
 
@@ -604,7 +612,7 @@ void onUdfcPipeClose(uv_handle_t *handle) {
 }
 
 int32_t udfcGetUvTaskResponseResult(SClientUdfTask *task, SClientUvTaskNode *uvTask) {
-  debugPrint("%s", "get uv task result");
+  fnDebug("udfc get uv task result. task: %p", task);
   if (uvTask->type == UV_TASK_REQ_RSP) {
     if (uvTask->rspBuf.base != NULL) {
       SUdfResponse rsp;
@@ -647,7 +655,6 @@ int32_t udfcGetUvTaskResponseResult(SClientUdfTask *task, SClientUvTaskNode *uvT
 }
 
 void udfcAllocateBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf) {
-  debugPrint("%s", "client allocate buffer to receive from pipe");
   SClientUvConn *conn = handle->data;
   SClientConnBuf *connBuf = &conn->readBuf;
 
@@ -662,7 +669,7 @@ void udfcAllocateBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf
       buf->base = connBuf->buf;
       buf->len = connBuf->cap;
     } else {
-      //TODO: log error
+      fnError("udfc allocate buffer failure. size: %d", msgHeadSize);
       buf->base = NULL;
       buf->len = 0;
     }
@@ -674,13 +681,13 @@ void udfcAllocateBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf
       buf->base = connBuf->buf + connBuf->len;
       buf->len = connBuf->cap - connBuf->len;
     } else {
-      //TODO: log error free connBuf->buf
+      fnError("udfc re-allocate buffer failure. size: %d", connBuf->cap);
       buf->base = NULL;
       buf->len = 0;
     }
   }
 
-  debugPrint("\tconn buf cap - len - total : %d - %d - %d", connBuf->cap, connBuf->len, connBuf->total);
+  fnTrace("conn buf cap - len - total : %d - %d - %d", connBuf->cap, connBuf->len, connBuf->total);
 
 }
 
@@ -689,6 +696,7 @@ bool isUdfcUvMsgComplete(SClientConnBuf *connBuf) {
     connBuf->total = *(int32_t *) (connBuf->buf);
   }
   if (connBuf->len == connBuf->cap && connBuf->total == connBuf->cap) {
+    fnTrace("udfc complete message is received, now handle it");
     return true;
   }
   return false;
@@ -696,10 +704,10 @@ bool isUdfcUvMsgComplete(SClientConnBuf *connBuf) {
 
 void udfcUvHandleRsp(SClientUvConn *conn) {
   SClientConnBuf *connBuf = &conn->readBuf;
-  int64_t seqNum = *(int64_t *) (connBuf->buf + sizeof(int32_t)); // msglen int32_t then seqnum
+  int64_t seqNum = *(int64_t *) (connBuf->buf + sizeof(int32_t)); // msglen then seqnum
 
   if (QUEUE_EMPTY(&conn->taskQueue)) {
-    //LOG error
+    fnError("udfc no task waiting for response on connection");
     return;
   }
   bool found = false;
@@ -713,7 +721,7 @@ void udfcUvHandleRsp(SClientUvConn *conn) {
         found = true;
         taskFound = task;
       } else {
-        //LOG error;
+        fnError("udfc more than one task waiting for the same response");
         continue;
       }
     }
@@ -727,7 +735,7 @@ void udfcUvHandleRsp(SClientUvConn *conn) {
     uv_sem_post(&taskFound->taskSem);
     QUEUE_REMOVE(&taskFound->procTaskQueue);
   } else {
-    //TODO: LOG error
+    fnError("no task is waiting for the response.");
   }
   connBuf->buf = NULL;
   connBuf->total = -1;
@@ -751,7 +759,7 @@ void udfcUvHandleError(SClientUvConn *conn) {
 }
 
 void onUdfcRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
-  debugPrint("%s, nread: %zd", "client read from pipe", nread);
+  fnTrace("udfc client %p, client read from pipe. nread: %zd", client, nread);
   if (nread == 0) return;
 
   SClientUvConn *conn = client->data;
@@ -764,9 +772,9 @@ void onUdfcRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
 
   }
   if (nread < 0) {
-    debugPrint("\tclient read error: %s", uv_strerror(nread));
+    fnError("udfc client pipe %p read error: %s", client, uv_strerror(nread));
     if (nread == UV_EOF) {
-      //TODO:
+      fnError("udfc client pipe %p closed", client);
     }
     udfcUvHandleError(conn);
   }
@@ -774,16 +782,15 @@ void onUdfcRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
 }
 
 void onUdfClientWrite(uv_write_t *write, int status) {
-  debugPrint("%s", "after writing to pipe");
   SClientUvTaskNode *uvTask = write->data;
+  uv_pipe_t *pipe = uvTask->pipe;
   if (status == 0) {
-    uv_pipe_t *pipe = uvTask->pipe;
     SClientUvConn *conn = pipe->data;
     QUEUE_INSERT_TAIL(&conn->taskQueue, &uvTask->connTaskQueue);
   } else {
-    //TODO Log error;
+    fnError("udfc client %p write error.", pipe);
   }
-  debugPrint("\tlength:%zu", uvTask->reqBuf.len);
+  fnTrace("udfc client %p write length:%zu", pipe, uvTask->reqBuf.len);
   taosMemoryFree(write);
   taosMemoryFree(uvTask->reqBuf.base);
 }
@@ -841,7 +848,7 @@ int32_t createUdfcUvTask(SClientUdfTask *task, int8_t uvTaskType, SClientUvTaskN
 }
 
 int32_t queueUvUdfTask(SClientUvTaskNode *uvTask) {
-  debugPrint("%s, %d", "queue uv task", uvTask->type);
+  fnTrace("queue uv task to event loop, task: %d, %p", uvTask->type, uvTask);
   SUdfdProxy *udfc = uvTask->udfc;
   uv_mutex_lock(&udfc->gUdfTaskQueueMutex);
   QUEUE_INSERT_TAIL(&udfc->gUdfTaskQueue, &uvTask->recvTaskQueue);
@@ -855,7 +862,7 @@ int32_t queueUvUdfTask(SClientUvTaskNode *uvTask) {
 }
 
 int32_t startUvUdfTask(SClientUvTaskNode *uvTask) {
-  debugPrint("%s, type %d", "start uv task ", uvTask->type);
+  fnTrace("event loop start uv task. task: %d, %p", uvTask->type, uvTask);
   switch (uvTask->type) {
     case UV_TASK_CONNECT: {
       uv_pipe_t *pipe = taosMemoryMalloc(sizeof(uv_pipe_t));
@@ -874,8 +881,7 @@ int32_t startUvUdfTask(SClientUvTaskNode *uvTask) {
 
       uv_connect_t *connReq = taosMemoryMalloc(sizeof(uv_connect_t));
       connReq->data = uvTask;
-
-      uv_pipe_connect(connReq, pipe, "udf.sock", onUdfClientConnect);
+      uv_pipe_connect(connReq, pipe, uvTask->udfc->udfdPipeName, onUdfClientConnect);
       break;
     }
     case UV_TASK_REQ_RSP: {
@@ -971,27 +977,37 @@ void constructUdfService(void *argsThread) {
   uv_loop_close(&udfc->gUdfdLoop);
 }
 
-int32_t udfcOpen(int32_t dnodeId, UdfcHandle *udfc) {
-  SUdfdProxy *proxy = taosMemoryCalloc(1, sizeof(SUdfdProxy));
-  proxy->dnodeId = dnodeId;
+int32_t udfcOpen() {
+  int8_t old = atomic_val_compare_exchange_8(&gUdfdProxy.initialized, 0, 1);
+  if (old == 1) {
+    return 0;
+  }
+  SUdfdProxy *proxy = &gUdfdProxy;
+  getUdfdPipeName(proxy->udfdPipeName, UDF_LISTEN_PIPE_NAME_LEN);
   proxy->gUdfcState = UDFC_STATE_STARTNG;
   uv_barrier_init(&proxy->gUdfInitBarrier, 2);
   uv_thread_create(&proxy->gUdfLoopThread, constructUdfService, proxy);
-  uv_barrier_wait(&proxy->gUdfInitBarrier);
+  atomic_store_8(&proxy->gUdfcState, UDFC_STATE_READY);
   proxy->gUdfcState = UDFC_STATE_READY;
-  *udfc = proxy;
+  uv_barrier_wait(&proxy->gUdfInitBarrier);
+  fnInfo("udfc initialized")
   return 0;
 }
 
-int32_t udfcClose(UdfcHandle udfcHandle) {
-  SUdfdProxy *udfc = udfcHandle;
+int32_t udfcClose() {
+  int8_t old = atomic_val_compare_exchange_8(&gUdfdProxy.initialized, 1, 0);
+  if (old == 0) {
+    return 0;
+  }
+
+  SUdfdProxy *udfc = &gUdfdProxy;
   udfc->gUdfcState = UDFC_STATE_STOPPING;
   uv_async_send(&udfc->gUdfLoopStopAsync);
   uv_thread_join(&udfc->gUdfLoopThread);
   uv_mutex_destroy(&udfc->gUdfTaskQueueMutex);
   uv_barrier_destroy(&udfc->gUdfInitBarrier);
-  udfc->gUdfcState = UDFC_STATUS_FINAL;
-  taosMemoryFree(udfc);
+  udfc->gUdfcState = UDFC_STATE_INITAL;
+  fnInfo("udfc cleaned up");
   return 0;
 }
 
@@ -1009,12 +1025,15 @@ int32_t udfcRunUvTask(SClientUdfTask *task, int8_t uvTaskType) {
   return task->errCode;
 }
 
-int32_t setupUdf(UdfcHandle udfc, char udfName[], SEpSet *epSet, UdfcFuncHandle *funcHandle) {
-  debugPrint("%s", "client setup udf");
+int32_t setupUdf(char udfName[], UdfcFuncHandle *funcHandle) {
+  fnInfo("udfc setup udf. udfName: %s", udfName);
+  if (gUdfdProxy.gUdfcState != UDFC_STATE_READY) {
+    return UDFC_CODE_INVALID_STATE;
+  }
   SClientUdfTask *task = taosMemoryMalloc(sizeof(SClientUdfTask));
   task->errCode = 0;
   task->session = taosMemoryMalloc(sizeof(SUdfUvSession));
-  task->session->udfc = udfc;
+  task->session->udfc = &gUdfdProxy;
   task->type = UDF_TASK_SETUP;
 
   SUdfSetupRequest *req = &task->_setup.req;
@@ -1022,15 +1041,20 @@ int32_t setupUdf(UdfcHandle udfc, char udfName[], SEpSet *epSet, UdfcFuncHandle
 
   int32_t errCode = udfcRunUvTask(task, UV_TASK_CONNECT);
   if (errCode != 0) {
-    //TODO: log error
-    return -1;
+    fnError("failed to connect to pipe. udfName: %s, pipe: %s", udfName, (&gUdfdProxy)->udfdPipeName);
+    return UDFC_CODE_CONNECT_PIPE_ERR;
   }
 
   udfcRunUvTask(task, UV_TASK_REQ_RSP);
 
   SUdfSetupResponse *rsp = &task->_setup.rsp;
   task->session->severHandle = rsp->udfHandle;
-  *funcHandle = task->session;
+  if (task->errCode != 0) {
+    fnError("failed to setup udf. err: %d", task->errCode)
+  } else {
+    fnInfo("sucessfully setup udf func handle. handle: %p", task->session);
+    *funcHandle = task->session;
+  }
   int32_t err = task->errCode;
   taosMemoryFree(task);
   return err;
@@ -1038,7 +1062,7 @@ int32_t setupUdf(UdfcHandle udfc, char udfName[], SEpSet *epSet, UdfcFuncHandle
 
 int32_t callUdf(UdfcFuncHandle handle, int8_t callType, SSDataBlock *input, SUdfInterBuf *state, SUdfInterBuf *state2,
                 SSDataBlock* output, SUdfInterBuf *newState) {
-  debugPrint("%s", "client call udf");
+  fnTrace("udfc call udf. callType: %d, funcHandle: %p", callType, handle);
 
   SClientUdfTask *task = taosMemoryMalloc(sizeof(SClientUdfTask));
   task->errCode = 0;
@@ -1076,35 +1100,37 @@ int32_t callUdf(UdfcFuncHandle handle, int8_t callType, SSDataBlock *input, SUdf
 
   udfcRunUvTask(task, UV_TASK_REQ_RSP);
 
-  SUdfCallResponse *rsp = &task->_call.rsp;
-  switch (callType) {
-    case TSDB_UDF_CALL_AGG_INIT: {
-      *newState = rsp->resultBuf;
-      break;
-    }
-    case TSDB_UDF_CALL_AGG_PROC: {
-      *newState = rsp->resultBuf;
-      break;
-    }
-    case TSDB_UDF_CALL_AGG_MERGE: {
-      *newState = rsp->resultBuf;
-      break;
-    }
-    case TSDB_UDF_CALL_AGG_FIN: {
-      *newState = rsp->resultBuf;
-      break;
-    }
-    case TSDB_UDF_CALL_SCALA_PROC: {
-      *output = rsp->resultData;
-      break;
+  if (task->errCode != 0) {
+    fnError("call udf failure. err: %d", task->errCode);
+  } else {
+    SUdfCallResponse *rsp = &task->_call.rsp;
+    switch (callType) {
+      case TSDB_UDF_CALL_AGG_INIT: {
+        *newState = rsp->resultBuf;
+        break;
+      }
+      case TSDB_UDF_CALL_AGG_PROC: {
+        *newState = rsp->resultBuf;
+        break;
+      }
+      case TSDB_UDF_CALL_AGG_MERGE: {
+        *newState = rsp->resultBuf;
+        break;
+      }
+      case TSDB_UDF_CALL_AGG_FIN: {
+        *newState = rsp->resultBuf;
+        break;
+      }
+      case TSDB_UDF_CALL_SCALA_PROC: {
+        *output = rsp->resultData;
+        break;
+      }
     }
   }
-
   taosMemoryFree(task);
   return task->errCode;
 }
 
-//TODO: translate these calls to callUdf
 int32_t callUdfAggInit(UdfcFuncHandle handle, SUdfInterBuf *interBuf) {
   int8_t callType = TSDB_UDF_CALL_AGG_INIT;
 
@@ -1148,7 +1174,7 @@ int32_t callUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t nu
 }
 
 int32_t teardownUdf(UdfcFuncHandle handle) {
-  debugPrint("%s", "client teardown udf");
+  fnInfo("tear down udf. udf func handle: %p", handle);
 
   SClientUdfTask *task = taosMemoryMalloc(sizeof(SClientUdfTask));
   task->errCode = 0;
@@ -1160,7 +1186,6 @@ int32_t teardownUdf(UdfcFuncHandle handle) {
 
   udfcRunUvTask(task, UV_TASK_REQ_RSP);
 
-
   SUdfTeardownResponse *rsp = &task->_teardown.rsp;
 
   int32_t err = task->errCode;

source/libs/function/src/udfd.c

@@ -20,6 +20,7 @@
 #include "tudf.h"
 #include "tudfInt.h"
 
+#include "tdatablock.h"
 #include "tdataformat.h"
 #include "tglobal.h"
 #include "tmsg.h"
@@ -31,8 +32,9 @@ typedef struct SUdfdContext {
   uv_signal_t intrSignal;
   char        listenPipeName[UDF_LISTEN_PIPE_NAME_LEN];
   uv_pipe_t   listeningPipe;
-  void       *clientRpc;
 
+  void       *clientRpc;
+  SCorEpSet  mgmtEp;
   uv_mutex_t udfsMutex;
   SHashObj  *udfsHash;
 
@@ -63,8 +65,13 @@ typedef struct SUdf {
   uv_mutex_t lock;
   uv_cond_t  condReady;
 
-  char   name[16];
-  int8_t type;
+  char   name[TSDB_FUNC_NAME_LEN];
+  int8_t funcType;
+  int8_t scriptType;
+  int8_t outputType;
+  int32_t outputLen;
+  int32_t bufSize;
+
   char   path[PATH_MAX];
 
   uv_lib_t              lib;
@@ -78,17 +85,17 @@ typedef struct SUdfcFuncHandle {
   SUdf *udf;
 } SUdfcFuncHandle;
 
-int32_t udfdFillUdfInfoFromMNode(void *clientRpc, SEpSet *pEpSet, char *udfName, SUdf *udf);
+int32_t udfdFillUdfInfoFromMNode(void *clientRpc, char *udfName, SUdf *udf);
 
-int32_t udfdLoadUdf(char *udfName, SEpSet *pEpSet, SUdf *udf) {
+int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
   strcpy(udf->name, udfName);
 
-  udfdFillUdfInfoFromMNode(global.clientRpc, pEpSet, udf->name, udf);
-
+  udfdFillUdfInfoFromMNode(global.clientRpc, udf->name, udf);
+  //strcpy(udf->path, "/home/slzhou/TDengine/debug/build/lib/libudf1.so");
   int err = uv_dlopen(udf->path, &udf->lib);
   if (err != 0) {
     fnError("can not load library %s. error: %s", udf->path, uv_strerror(err));
-    // TODO set error
+    return UDFC_CODE_LOAD_UDF_FAILURE;
   }
   // TODO: find all the functions
   char normalFuncName[TSDB_FUNC_NAME_LEN] = {0};
@@ -115,8 +122,8 @@ void udfdProcessRequest(uv_work_t *req) {
 
       SUdf *udf = NULL;
       uv_mutex_lock(&global.udfsMutex);
-      SUdf **udfInHash = taosHashGet(global.udfsHash, request.setup.udfName, TSDB_FUNC_NAME_LEN);
-      if (*udfInHash) {
+      SUdf **udfInHash = taosHashGet(global.udfsHash, request.setup.udfName, strlen(request.setup.udfName));
+      if (udfInHash) {
         ++(*udfInHash)->refCount;
         udf = *udfInHash;
         uv_mutex_unlock(&global.udfsMutex);
@@ -128,14 +135,14 @@ void udfdProcessRequest(uv_work_t *req) {
         uv_mutex_init(&udfNew->lock);
         uv_cond_init(&udfNew->condReady);
         udf = udfNew;
-        taosHashPut(global.udfsHash, request.setup.udfName, TSDB_FUNC_NAME_LEN, &udfNew, sizeof(&udfNew));
+        taosHashPut(global.udfsHash, request.setup.udfName, strlen(request.setup.udfName), &udfNew, sizeof(&udfNew));
         uv_mutex_unlock(&global.udfsMutex);
       }
 
       uv_mutex_lock(&udf->lock);
       if (udf->state == UDF_STATE_INIT) {
         udf->state = UDF_STATE_LOADING;
-        udfdLoadUdf(setup->udfName, &setup->epSet, udf);
+        udfdLoadUdf(setup->udfName, udf);
         udf->state = UDF_STATE_READY;
         uv_cond_broadcast(&udf->condReady);
         uv_mutex_unlock(&udf->lock);
@@ -214,7 +221,7 @@ void udfdProcessRequest(uv_work_t *req) {
       udf->refCount--;
       if (udf->refCount == 0) {
         unloadUdf = true;
-        taosHashRemove(global.udfsHash, udf->name, TSDB_FUNC_NAME_LEN);
+        taosHashRemove(global.udfsHash, udf->name, strlen(udf->name));
       }
       uv_mutex_unlock(&global.udfsMutex);
       if (unloadUdf) {
@@ -393,7 +400,48 @@ void udfdIntrSignalHandler(uv_signal_t *handle, int signum) {
 
 void udfdProcessRpcRsp(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet) { return; }
 
-int32_t udfdFillUdfInfoFromMNode(void *clientRpc, SEpSet *pEpSet, char *udfName, SUdf *udf) {
+int initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet) {
+  pEpSet->version = 0;
+
+  // init mnode ip set
+  SEpSet* mgmtEpSet = &(pEpSet->epSet);
+  mgmtEpSet->numOfEps = 0;
+  mgmtEpSet->inUse = 0;
+
+  if (firstEp && firstEp[0] != 0) {
+    if (strlen(firstEp) >= TSDB_EP_LEN) {
+      terrno = TSDB_CODE_TSC_INVALID_FQDN;
+      return -1;
+    }
+
+    int32_t code = taosGetFqdnPortFromEp(firstEp, &mgmtEpSet->eps[0]);
+    if (code != TSDB_CODE_SUCCESS) {
+      terrno = TSDB_CODE_TSC_INVALID_FQDN;
+      return terrno;
+    }
+
+    mgmtEpSet->numOfEps++;
+  }
+
+  if (secondEp && secondEp[0] != 0) {
+    if (strlen(secondEp) >= TSDB_EP_LEN) {
+      terrno = TSDB_CODE_TSC_INVALID_FQDN;
+      return -1;
+    }
+
+    taosGetFqdnPortFromEp(secondEp, &mgmtEpSet->eps[mgmtEpSet->numOfEps]);
+    mgmtEpSet->numOfEps++;
+  }
+
+  if (mgmtEpSet->numOfEps == 0) {
+    terrno = TSDB_CODE_TSC_INVALID_FQDN;
+    return -1;
+  }
+
+  return 0;
+}
+
+int32_t udfdFillUdfInfoFromMNode(void *clientRpc, char *udfName, SUdf *udf) {
   SRetrieveFuncReq retrieveReq = {0};
   retrieveReq.numOfFuncs = 1;
   retrieveReq.pFuncNames = taosArrayInit(1, TSDB_FUNC_NAME_LEN);
@@ -410,15 +458,21 @@ int32_t udfdFillUdfInfoFromMNode(void *clientRpc, SEpSet *pEpSet, char *udfName,
   rpcMsg.msgType = TDMT_MND_RETRIEVE_FUNC;
 
   SRpcMsg rpcRsp = {0};
-  rpcSendRecv(clientRpc, pEpSet, &rpcMsg, &rpcRsp);
+  rpcSendRecv(clientRpc, &global.mgmtEp.epSet, &rpcMsg, &rpcRsp);
   SRetrieveFuncRsp retrieveRsp = {0};
   tDeserializeSRetrieveFuncRsp(rpcRsp.pCont, rpcRsp.contLen, &retrieveRsp);
 
   SFuncInfo *pFuncInfo = (SFuncInfo *)taosArrayGet(retrieveRsp.pFuncInfos, 0);
 
+  udf->funcType = pFuncInfo->funcType;
+  udf->scriptType = pFuncInfo->scriptType;
+  udf->outputType = pFuncInfo->funcType;
+  udf->outputLen = pFuncInfo->outputLen;
+  udf->bufSize = pFuncInfo->bufSize;
+
   char path[PATH_MAX] = {0};
-  taosGetTmpfilePath("/tmp", "libudf", path);
-  TdFilePtr file = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC);
+  snprintf(path, sizeof(path), "%s/lib%s.so", "/tmp", udfName);
+  TdFilePtr file = taosOpenFile(path, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC | TD_FILE_AUTO_DEL);
   // TODO check for failure of flush to disk
   taosWriteFile(file, pFuncInfo->pCode, pFuncInfo->codeSize);
   taosCloseFile(&file);
@@ -531,15 +585,7 @@ static int32_t udfdUvInit() {
   uv_pipe_open(&global.ctrlPipe, 0);
   uv_read_start((uv_stream_t *)&global.ctrlPipe, udfdCtrlAllocBufCb, udfdCtrlReadCb);
 
-  char    dnodeId[8] = {0};
-  size_t  dnodeIdSize;
-  int32_t err = uv_os_getenv("DNODE_ID", dnodeId, &dnodeIdSize);
-  if (err != 0) {
-    dnodeId[0] = '1';
-  }
-  char listenPipeName[32] = {0};
-  snprintf(listenPipeName, sizeof(listenPipeName), "%s%s", UDF_LISTEN_PIPE_NAME_PREFIX, dnodeId);
-  strcpy(global.listenPipeName, listenPipeName);
+  getUdfdPipeName(global.listenPipeName, UDF_LISTEN_PIPE_NAME_LEN);
 
   removeListeningPipe();
 
@@ -550,7 +596,7 @@ static int32_t udfdUvInit() {
 
   int r;
   fnInfo("bind to pipe %s", global.listenPipeName);
-  if ((r = uv_pipe_bind(&global.listeningPipe, listenPipeName))) {
+  if ((r = uv_pipe_bind(&global.listeningPipe, global.listenPipeName))) {
     fnError("Bind error %s", uv_err_name(r));
     removeListeningPipe();
     return -1;
@@ -580,7 +626,7 @@ static int32_t udfdRun() {
 
   fnInfo("start the udfd");
   int code = uv_run(global.loop, UV_RUN_DEFAULT);
-  fnInfo("udfd stopped. result: %s", uv_err_name(code));
+  fnInfo("udfd stopped. result: %s, code: %d", uv_err_name(code), code);
   int codeClose = uv_loop_close(global.loop);
   fnDebug("uv loop close. result: %s", uv_err_name(codeClose));
   udfdCloseClientRpc();
@@ -615,5 +661,6 @@ int main(int argc, char *argv[]) {
     return -1;
   }
 
+  initEpSetFromCfg(tsFirst, tsSecond, &global.mgmtEp);
   return udfdRun();
 }

source/libs/function/test/runUdf.c

 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
-
 #include "uv.h"
+
+#include "fnLog.h"
 #include "os.h"
-#include "tudf.h"
 #include "tdatablock.h"
+#include "tglobal.h"
+#include "tudf.h"
 
-int main(int argc, char *argv[]) {
-  UdfcHandle udfc;
-  udfcOpen(1, &udfc);
-    uv_sleep(1000);
-    char path[256] = {0};
-    size_t cwdSize = 256;
-    int err = uv_cwd(path, &cwdSize);
-    if (err != 0) {
-        fprintf(stderr, "err cwd: %s\n", uv_strerror(err));
-	    return err;
-    }
-    fprintf(stdout, "current working directory:%s\n", path);
-    strcat(path, "/libudf1.so");
-
-    UdfcFuncHandle handle;
-    SEpSet epSet;
-    setupUdf(udfc, "udf1", &epSet, &handle);
-
-    SSDataBlock block = {0};
-    SSDataBlock* pBlock = &block;
-    pBlock->pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
-    pBlock->info.numOfCols = 1;
-    pBlock->info.rows = 4;
-    char data[16] = {0};
-    char bitmap[4] = {0};
-    for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
-      SColumnInfoData colInfo = {0};
-      colInfo.info.type = TSDB_DATA_TYPE_INT;
-      colInfo.info.bytes = sizeof(int32_t);
-      colInfo.info.colId = 1;
-      colInfo.pData = data;
-      colInfo.nullbitmap = bitmap;
-      for (int32_t j = 0; j < pBlock->info.rows; ++j) {
-        colDataAppendInt32(&colInfo, j, &j);
+static int32_t parseArgs(int32_t argc, char *argv[]) {
+  for (int32_t i = 1; i < argc; ++i) {
+    if (strcmp(argv[i], "-c") == 0) {
+      if (i < argc - 1) {
+        if (strlen(argv[++i]) >= PATH_MAX) {
+          printf("config file path overflow");
+          return -1;
+        }
+        tstrncpy(configDir, argv[i], PATH_MAX);
+      } else {
+        printf("'-c' requires a parameter, default is %s\n", configDir);
+        return -1;
       }
-      taosArrayPush(pBlock->pDataBlock, &colInfo);
     }
+  }
 
-    SScalarParam input = {0};
-    input.numOfRows = pBlock->info.rows;
-    input.columnData = taosArrayGet(pBlock->pDataBlock, 0);
-    SScalarParam output = {0};
-    callUdfScalarFunc(handle, &input, 1 , &output);
+  return 0;
+}
 
-    SColumnInfoData *col = output.columnData;
-    for (int32_t i = 0; i < output.numOfRows; ++i) {
-      fprintf(stderr, "%d\t%d\n" , i, *(int32_t*)(col->pData + i *sizeof(int32_t)));
+static int32_t initLog() {
+  char logName[12] = {0};
+  snprintf(logName, sizeof(logName), "%slog", "udfc");
+  return taosCreateLog(logName, 1, configDir, NULL, NULL, NULL, 0);
+}
+
+int main(int argc, char *argv[]) {
+  parseArgs(argc, argv);
+  initLog();
+  if (taosInitCfg(configDir, NULL, NULL, NULL, 0) != 0) {
+    fnError("failed to start since read config error");
+    return -1;
+  }
+
+  udfcOpen();
+  uv_sleep(1000);
+
+  UdfcFuncHandle handle;
+
+  setupUdf("udf1", &handle);
+
+  SSDataBlock  block = {0};
+  SSDataBlock *pBlock = &block;
+  pBlock->pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
+  pBlock->info.numOfCols = 1;
+  pBlock->info.rows = 4;
+  char data[16] = {0};
+  char bitmap[4] = {0};
+  for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
+    SColumnInfoData colInfo = {0};
+    colInfo.info.type = TSDB_DATA_TYPE_INT;
+    colInfo.info.bytes = sizeof(int32_t);
+    colInfo.info.colId = 1;
+    colInfo.pData = data;
+    colInfo.nullbitmap = bitmap;
+    for (int32_t j = 0; j < pBlock->info.rows; ++j) {
+      colDataAppendInt32(&colInfo, j, &j);
     }
-    teardownUdf(handle);
+    taosArrayPush(pBlock->pDataBlock, &colInfo);
+  }
+
+  SScalarParam input = {0};
+  input.numOfRows = pBlock->info.rows;
+  input.columnData = taosArrayGet(pBlock->pDataBlock, 0);
+  SScalarParam output = {0};
+  callUdfScalarFunc(handle, &input, 1, &output);
 
-    udfcClose(udfc);
+  SColumnInfoData *col = output.columnData;
+  for (int32_t i = 0; i < output.numOfRows; ++i) {
+    fprintf(stderr, "%d\t%d\n", i, *(int32_t *)(col->pData + i * sizeof(int32_t)));
+  }
+  teardownUdf(handle);
+  udfcClose();
 }

source/libs/parser/inc/sql.y

@@ -611,6 +611,7 @@ function_expression(A) ::= function_name(B) NK_LP expression_list(C) NK_RP(D).
 function_expression(A) ::= star_func(B) NK_LP star_func_para_list(C) NK_RP(D).    { A = createRawExprNodeExt(pCxt, &B, &D, createFunctionNode(pCxt, &B, C)); }
 function_expression(A) ::= CAST(B) NK_LP expression(C) AS type_name(D) NK_RP(E).  { A = createRawExprNodeExt(pCxt, &B, &E, createCastFunctionNode(pCxt, releaseRawExprNode(pCxt, C), D)); }
 function_expression(A) ::= noarg_func(B) NK_LP NK_RP(C).                          { A = createRawExprNodeExt(pCxt, &B, &C, createFunctionNodeNoArg(pCxt, &B)); }
+//function_expression(A) ::= NOW(B).                                                { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
 
 %type noarg_func                                                                  { SToken }
 %destructor noarg_func                                                            { }

source/libs/parser/src/parCalcConst.c

@@ -83,7 +83,7 @@ static EDealRes calcConstOperator(SOperatorNode** pNode, void* pContext) {
 
 static EDealRes calcConstFunction(SFunctionNode** pNode, void* pContext) {
   SFunctionNode* pFunc = *pNode;
-  if (!fmIsScalarFunc(pFunc->funcId)) {
+  if (!fmIsScalarFunc(pFunc->funcId) || fmIsUserDefinedFunc(pFunc->funcId)) {
     return DEAL_RES_CONTINUE;
   }
   SNode* pParam = NULL;

source/libs/parser/src/parUtil.c

@@ -128,6 +128,8 @@ static char* getSyntaxErrFormat(int32_t errCode) {
       return "soffset/offset can not be less than 0";
     case TSDB_CODE_PAR_SLIMIT_LEAK_PARTITION_BY:
       return "slimit/soffset only available for PARTITION BY query";
+    case TSDB_CODE_PAR_INVALID_TOPIC_QUERY:
+      return "Invalid topic query";
     case TSDB_CODE_OUT_OF_MEMORY:
       return "Out of memory";
     default:

source/libs/planner/src/planLogicCreater.c

@@ -199,7 +199,8 @@ static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
 
   TSWAP(pScan->pMeta, pRealTable->pMeta, STableMeta*);
   TSWAP(pScan->pVgroupList, pRealTable->pVgroupList, SVgroupsInfo*);
-  pScan->scanFlag = MAIN_SCAN;
+  pScan->scanSeq[0] = 1;
+  pScan->scanSeq[1] = 0;
   pScan->scanRange = TSWINDOW_INITIALIZER;  
   pScan->tableName.type = TSDB_TABLE_NAME_T;
   pScan->tableName.acctId = pCxt->pPlanCxt->acctId;

source/util/src/ttimer.c

@@ -110,7 +110,7 @@ typedef struct time_wheel_t {
   tmr_obj_t**     slots;
 } time_wheel_t;
 
-int32_t tsMaxTmrCtrl = 512;
+static int32_t tsMaxTmrCtrl = 512;
 
 static TdThreadOnce  tmrModuleInit = PTHREAD_ONCE_INIT;
 static TdThreadMutex tmrCtrlMutex;