diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c
index 6dc78d6400..a73808f2ed 100644
--- a/source/common/src/tglobal.c
+++ b/source/common/src/tglobal.c
@@ -226,8 +226,17 @@ static int32_t taosSetTfsCfg(SConfig *pCfg) {
   }
 
   if (tsDataDir[0] == 0) {
-    uError("datadir not set");
-    return -1;
+    if (pItem->str != NULL) {
+      taosAddDataDir(0, pItem->str, 0, 1);
+      tstrncpy(tsDataDir, pItem->str, PATH_MAX);
+      if (taosMulMkDir(tsDataDir) != 0) {
+        uError("failed to create dataDir:%s since %s", tsDataDir, terrstr());
+        return -1;
+      }
+    } else {
+      uError("datadir not set");
+      return -1;
+    }
   }
 
   return 0;
diff --git a/source/libs/function/src/tudf.c b/source/libs/function/src/tudf.c
index 6ad552576c..dae8c99aba 100644
--- a/source/libs/function/src/tudf.c
+++ b/source/libs/function/src/tudf.c
@@ -875,7 +875,7 @@ bool isUdfcUvMsgComplete(SClientConnBuf *connBuf);
 void udfcUvHandleRsp(SClientUvConn *conn);
 void udfcUvHandleError(SClientUvConn *conn);
 void onUdfcPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf);
-void onUdfcPipetWrite(uv_write_t *write, int status);
+void onUdfcPipeWrite(uv_write_t *write, int status);
 void onUdfcPipeConnect(uv_connect_t *connect, int status);
 int32_t udfcCreateUvTask(SClientUdfTask *task, int8_t uvTaskType, SClientUvTaskNode **pUvTask);
 int32_t udfcQueueUvTask(SClientUvTaskNode *uvTask);
@@ -1226,7 +1226,7 @@ int32_t udfcGetUdfTaskResultFromUvTask(SClientUdfTask *task, SClientUvTaskNode *
 }
 
 void udfcAllocateBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf) {
-  SClientUvConn *conn = handle->data;
+  SClientUvConn  *conn = handle->data;
   SClientConnBuf *connBuf = &conn->readBuf;
 
   int32_t msgHeadSize = sizeof(int32_t) + sizeof(int64_t);
@@ -1244,6 +1244,9 @@ void udfcAllocateBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf
       buf->base = NULL;
       buf->len = 0;
     }
+  } else if (connBuf->total == -1 && connBuf->len < msgHeadSize) {
+    buf->base = connBuf->buf + connBuf->len;
+    buf->len = msgHeadSize - connBuf->len;
   } else {
     connBuf->cap = connBuf->total > connBuf->cap ? connBuf->total : connBuf->cap;
     void *resultBuf = taosMemoryRealloc(connBuf->buf, connBuf->cap);
@@ -1258,8 +1261,7 @@ void udfcAllocateBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf
     }
   }
 
-  fnTrace("conn buf cap - len - total : %d - %d - %d", connBuf->cap, connBuf->len, connBuf->total);
-
+  fnDebug("udfc uv alloc buffer: cap - len - total : %d - %d - %d", connBuf->cap, connBuf->len, connBuf->total);
 }
 
 bool isUdfcUvMsgComplete(SClientConnBuf *connBuf) {
@@ -1267,7 +1269,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");
+    fnDebug("udfc complete message is received, now handle it");
     return true;
   }
   return false;
@@ -1278,7 +1280,7 @@ void udfcUvHandleRsp(SClientUvConn *conn) {
   int64_t seqNum = *(int64_t *) (connBuf->buf + sizeof(int32_t)); // msglen then seqnum
 
   if (QUEUE_EMPTY(&conn->taskQueue)) {
-    fnError("udfc no task waiting for response on connection");
+    fnError("udfc no task waiting on connection. response seqnum:%"PRId64, seqNum);
     return;
   }
   bool found = false;
@@ -1287,6 +1289,7 @@ void udfcUvHandleRsp(SClientUvConn *conn) {
   SClientUvTaskNode *task = QUEUE_DATA(h, SClientUvTaskNode, connTaskQueue);
 
   while (h != &conn->taskQueue) {
+    fnDebug("udfc handle response iterate through queue. uvTask:%d-%p", task->seqNum, task);
     if (task->seqNum == seqNum) {
       if (found == false) {
         found = true;
@@ -1315,6 +1318,7 @@ void udfcUvHandleRsp(SClientUvConn *conn) {
 }
 
 void udfcUvHandleError(SClientUvConn *conn) {
+  fnDebug("handle error on conn: %p, pipe: %p", conn, conn->pipe);
   while (!QUEUE_EMPTY(&conn->taskQueue)) {
     QUEUE* h = QUEUE_HEAD(&conn->taskQueue);
     SClientUvTaskNode *task = QUEUE_DATA(h, SClientUvTaskNode, connTaskQueue);
@@ -1328,7 +1332,7 @@ void udfcUvHandleError(SClientUvConn *conn) {
 }
 
 void onUdfcPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
-  fnTrace("udfc client %p, client read from pipe. nread: %zd", client, nread);
+  fnDebug("udfc client %p, client read from pipe. nread: %zd", client, nread);
   if (nread == 0) return;
 
   SClientUvConn *conn = client->data;
@@ -1338,31 +1342,25 @@ void onUdfcPipeRead(uv_stream_t *client, ssize_t nread, const uv_buf_t *buf) {
     if (isUdfcUvMsgComplete(connBuf)) {
       udfcUvHandleRsp(conn);
     }
-
   }
   if (nread < 0) {
-    fnError("udfc client pipe %p read error: %zd, %s.", client, nread, uv_strerror(nread));
+    fnError("udfc client pipe %p read error: %zd(%s).", client, nread, uv_strerror(nread));
     if (nread == UV_EOF) {
       fnError("\tudfc client pipe %p closed", client);
     }
     udfcUvHandleError(conn);
   }
-
 }
 
-void onUdfcPipetWrite(uv_write_t *write, int status) {
-  SClientUvTaskNode *uvTask = write->data;
-  uv_pipe_t *pipe = uvTask->pipe;
-  fnTrace("udfc client %p write length:%zu", pipe, uvTask->reqBuf.len);
-  SClientUvConn *conn = pipe->data;
-  if (status == 0) {
-    QUEUE_INSERT_TAIL(&conn->taskQueue, &uvTask->connTaskQueue);
-  } else {
-    fnError("udfc client %p write error.", pipe);
+void onUdfcPipeWrite(uv_write_t *write, int status) {
+  SClientUvConn *conn = write->data;
+  if (status < 0) {
+    fnError("udfc client connection %p write failed. status: %d(%s)", conn, status, uv_strerror(status));
     udfcUvHandleError(conn);
+  } else {
+    fnDebug("udfc client connection %p write succeed", conn);
   }
   taosMemoryFree(write);
-  taosMemoryFree(uvTask->reqBuf.base);
 }
 
 void onUdfcPipeConnect(uv_connect_t *connect, int status) {
@@ -1419,7 +1417,7 @@ int32_t udfcCreateUvTask(SClientUdfTask *task, int8_t uvTaskType, SClientUvTaskN
 }
 
 int32_t udfcQueueUvTask(SClientUvTaskNode *uvTask) {
-  fnTrace("queue uv task to event loop, task: %d, %p", uvTask->type, uvTask);
+  fnDebug("queue uv task to event loop, uvTask: %d-%p", uvTask->type, uvTask);
   SUdfcProxy *udfc = uvTask->udfc;
   uv_mutex_lock(&udfc->taskQueueMutex);
   QUEUE_INSERT_TAIL(&udfc->taskQueue, &uvTask->recvTaskQueue);
@@ -1427,14 +1425,14 @@ int32_t udfcQueueUvTask(SClientUvTaskNode *uvTask) {
   uv_async_send(&udfc->loopTaskAync);
 
   uv_sem_wait(&uvTask->taskSem);
-  fnInfo("udfc uv task finished. task: %d, %p", uvTask->type, uvTask);
+  fnInfo("udfc uvTask finished. uvTask:%"PRId64"-%d-%p", uvTask->seqNum, uvTask->type, uvTask);
   uv_sem_destroy(&uvTask->taskSem);
 
   return 0;
 }
 
 int32_t udfcStartUvTask(SClientUvTaskNode *uvTask) {
-  fnTrace("event loop start uv task. task: %d, %p", uvTask->type, uvTask);
+  fnDebug("event loop start uv task. uvTask: %"PRId64"-%d-%p", uvTask->seqNum, uvTask->type, uvTask);
   int32_t code = 0;
 
   switch (uvTask->type) {
@@ -1465,10 +1463,12 @@ int32_t udfcStartUvTask(SClientUvTaskNode *uvTask) {
         code = TSDB_CODE_UDF_PIPE_NO_PIPE;
       } else {
         uv_write_t *write = taosMemoryMalloc(sizeof(uv_write_t));
-        write->data = uvTask;
-        int err = uv_write(write, (uv_stream_t *)pipe, &uvTask->reqBuf, 1, onUdfcPipetWrite);
+        write->data = pipe->data;
+        QUEUE* connTaskQueue = &((SClientUvConn*)pipe->data)->taskQueue;
+        QUEUE_INSERT_TAIL(connTaskQueue, &uvTask->connTaskQueue);
+        int err = uv_write(write, (uv_stream_t *)pipe, &uvTask->reqBuf, 1, onUdfcPipeWrite);
         if (err != 0) {
-          fnError("udfc event loop start req/rsp task uv_write failed. code: %s", uv_strerror(err));
+          fnError("udfc event loop start req_rsp task uv_write failed. uvtask: %p, code: %s", uvTask, uv_strerror(err));
         }
         code = err;
       }
@@ -1618,6 +1618,7 @@ int32_t udfcRunUdfUvTask(SClientUdfTask *task, int8_t uvTaskType) {
   SClientUvTaskNode *uvTask = NULL;
 
   udfcCreateUvTask(task, uvTaskType, &uvTask);
+  fnDebug("udfc client task: %p created uvTask: %p. pipe: %p", task, uvTask, task->session->udfUvPipe);
   udfcQueueUvTask(uvTask);
   udfcGetUdfTaskResultFromUvTask(task, uvTask);
   if (uvTaskType == UV_TASK_CONNECT) {
@@ -1625,6 +1626,8 @@ int32_t udfcRunUdfUvTask(SClientUdfTask *task, int8_t uvTaskType) {
     SClientUvConn *conn = uvTask->pipe->data;
     conn->session = task->session;
   }
+  taosMemoryFree(uvTask->reqBuf.base);
+  uvTask->reqBuf.base = NULL;
   taosMemoryFree(uvTask);
   uvTask = NULL;
   return task->errCode;
@@ -1670,7 +1673,7 @@ int32_t doSetupUdf(char udfName[], UdfcFuncHandle *funcHandle) {
 
 int32_t callUdf(UdfcFuncHandle handle, int8_t callType, SSDataBlock *input, SUdfInterBuf *state, SUdfInterBuf *state2,
                 SSDataBlock* output, SUdfInterBuf *newState) {
-  fnTrace("udfc call udf. callType: %d, funcHandle: %p", callType, handle);
+  fnDebug("udfc call udf. callType: %d, funcHandle: %p", callType, handle);
   SUdfcUvSession *session = (SUdfcUvSession *) handle;
   if (session->udfUvPipe == NULL) {
     fnError("No pipe to udfd");
diff --git a/source/libs/function/src/udfd.c b/source/libs/function/src/udfd.c
index 2402607251..24749729c0 100644
--- a/source/libs/function/src/udfd.c
+++ b/source/libs/function/src/udfd.c
@@ -671,6 +671,9 @@ void udfdAllocBuffer(uv_handle_t *handle, size_t suggestedSize, uv_buf_t *buf) {
       fnError("udfd can not allocate enough memory") buf->base = NULL;
       buf->len = 0;
     }
+  } else if (ctx->inputTotal == -1 && ctx->inputLen < msgHeadSize) {
+      buf->base = ctx->inputBuf + ctx->inputLen;
+      buf->len = msgHeadSize - ctx->inputLen;
   } else {
     ctx->inputCap = ctx->inputTotal > ctx->inputCap ? ctx->inputTotal : ctx->inputCap;
     void *inputBuf = taosMemoryRealloc(ctx->inputBuf, ctx->inputCap);
diff --git a/source/util/src/tconfig.c b/source/util/src/tconfig.c
index db1207f057..11ae31919a 100644
--- a/source/util/src/tconfig.c
+++ b/source/util/src/tconfig.c
@@ -626,6 +626,7 @@ void cfgDumpCfg(SConfig *pCfg, bool tsc, bool dump) {
 int32_t cfgLoadFromEnvVar(SConfig *pConfig) {
   char   *line = NULL, *name, *value, *value2, *value3;
   int32_t olen, vlen, vlen2, vlen3;
+  int32_t code = 0;
   ssize_t _bytes = 0;
   TdCmdPtr pCmd = taosOpenCmd("set");
   if (pCmd == NULL) {
@@ -658,9 +659,12 @@ int32_t cfgLoadFromEnvVar(SConfig *pConfig) {
       if (vlen3 != 0) value3[vlen3] = 0;
     }
 
-    cfgSetItem(pConfig, name, value, CFG_STYPE_ENV_VAR);
     if (value2 != NULL && value3 != NULL && value2[0] != 0 && value3[0] != 0 && strcasecmp(name, "dataDir") == 0) {
-      cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_ENV_VAR);
+      code = cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_ENV_VAR);
+      if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
+    } else {
+      code = cfgSetItem(pConfig, name, value, CFG_STYPE_ENV_VAR);
+      if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
     }
   }
 
@@ -674,6 +678,7 @@ int32_t cfgLoadFromEnvVar(SConfig *pConfig) {
 int32_t cfgLoadFromEnvCmd(SConfig *pConfig, const char **envCmd) {
   char    buf[1024], *name, *value, *value2, *value3;
   int32_t olen, vlen, vlen2, vlen3;
+  int32_t code = 0;
   int32_t index = 0;
   if (envCmd == NULL) return 0;
   while (envCmd[index]!=NULL) {
@@ -700,9 +705,12 @@ int32_t cfgLoadFromEnvCmd(SConfig *pConfig, const char **envCmd) {
       if (vlen3 != 0) value3[vlen3] = 0;
     }
 
-    cfgSetItem(pConfig, name, value, CFG_STYPE_ENV_CMD);
     if (value2 != NULL && value3 != NULL && value2[0] != 0 && value3[0] != 0 && strcasecmp(name, "dataDir") == 0) {
-      cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_ENV_CMD);
+      code = cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_ENV_CMD);
+      if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
+    } else {
+      code = cfgSetItem(pConfig, name, value, CFG_STYPE_ENV_CMD);
+      if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
     }
   }
 
@@ -713,6 +721,7 @@ int32_t cfgLoadFromEnvCmd(SConfig *pConfig, const char **envCmd) {
 int32_t cfgLoadFromEnvFile(SConfig *pConfig, const char *envFile) {
   char   *line = NULL, *name, *value, *value2, *value3;
   int32_t olen, vlen, vlen2, vlen3;
+  int32_t code = 0;
   ssize_t _bytes = 0;
 
   const char *filepath = ".env";
@@ -761,9 +770,12 @@ int32_t cfgLoadFromEnvFile(SConfig *pConfig, const char *envFile) {
       if (vlen3 != 0) value3[vlen3] = 0;
     }
 
-    cfgSetItem(pConfig, name, value, CFG_STYPE_ENV_FILE);
     if (value2 != NULL && value3 != NULL && value2[0] != 0 && value3[0] != 0 && strcasecmp(name, "dataDir") == 0) {
-      cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_ENV_FILE);
+      code = cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_ENV_FILE);
+      if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
+    } else {
+      code = cfgSetItem(pConfig, name, value, CFG_STYPE_ENV_FILE);
+      if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
     }
   }
 
@@ -819,11 +831,12 @@ int32_t cfgLoadFromCfgFile(SConfig *pConfig, const char *filepath) {
       if (vlen3 != 0) value3[vlen3] = 0;
     }
 
-    code = cfgSetItem(pConfig, name, value, CFG_STYPE_CFG_FILE);
-    if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
     if (value2 != NULL && value3 != NULL && value2[0] != 0 && value3[0] != 0 && strcasecmp(name, "dataDir") == 0) {
       code = cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_CFG_FILE);
       if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
+    } else {
+      code = cfgSetItem(pConfig, name, value, CFG_STYPE_CFG_FILE);
+      if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
     }
   }
 
@@ -839,9 +852,75 @@ int32_t cfgLoadFromCfgFile(SConfig *pConfig, const char *filepath) {
   }
 }
 
+// int32_t cfgLoadFromCfgText(SConfig *pConfig, const char *configText) {
+//   char   *line = NULL, *name, *value, *value2, *value3;
+//   int32_t olen, vlen, vlen2, vlen3;
+//   ssize_t _bytes = 0;
+//   int32_t code = 0;
+
+//   TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_READ | TD_FILE_STREAM);
+//   if (pFile == NULL) {
+//     // success when the file does not exist
+//     if (errno == ENOENT) {
+//       terrno = TAOS_SYSTEM_ERROR(errno);
+//       uInfo("failed to load from cfg file %s since %s, use default parameters", filepath, terrstr());
+//       return 0;
+//     } else {
+//       uError("failed to load from cfg file %s since %s", filepath, terrstr());
+//       return -1;
+//     }
+//   }
+
+//   while (!taosEOFFile(pFile)) {
+//     name = value = value2 = value3 = NULL;
+//     olen = vlen = vlen2 = vlen3 = 0;
+
+//     _bytes = taosGetLineFile(pFile, &line);
+//     if (_bytes <= 0) {
+//       break;
+//     }
+
+//     if(line[_bytes - 1] == '\n') line[_bytes - 1] = 0;
+
+//     paGetToken(line, &name, &olen);
+//     if (olen == 0) continue;
+//     name[olen] = 0;
+
+//     paGetToken(name + olen + 1, &value, &vlen);
+//     if (vlen == 0) continue;
+//     value[vlen] = 0;
+
+//     paGetToken(value + vlen + 1, &value2, &vlen2);
+//     if (vlen2 != 0) {
+//       value2[vlen2] = 0;
+//       paGetToken(value2 + vlen2 + 1, &value3, &vlen3);
+//       if (vlen3 != 0) value3[vlen3] = 0;
+//     }
+
+//     code = cfgSetItem(pConfig, name, value, CFG_STYPE_CFG_FILE);
+//     if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
+//     if (value2 != NULL && value3 != NULL && value2[0] != 0 && value3[0] != 0 && strcasecmp(name, "dataDir") == 0) {
+//       code = cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_CFG_FILE);
+//       if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
+//     }
+//   }
+
+//   taosCloseFile(&pFile);
+//   if (line != NULL) taosMemoryFreeClear(line);
+
+//   if (code == 0 || (code != 0 && terrno == TSDB_CODE_CFG_NOT_FOUND)) {
+//     uInfo("load from cfg file %s success", filepath);
+//     return 0;
+//   } else {
+//     uError("failed to load from cfg file %s since %s", filepath, terrstr());
+//     return -1;
+//   }
+// }
+
 int32_t cfgLoadFromApollUrl(SConfig *pConfig, const char *url) {
   char   *cfgLineBuf = NULL, *name, *value, *value2, *value3;
   int32_t olen, vlen, vlen2, vlen3;
+  int32_t code = 0;
   if (url == NULL || strlen(url) == 0) {
     uInfo("fail to load apoll url");
     return 0;
@@ -916,9 +995,12 @@ int32_t cfgLoadFromApollUrl(SConfig *pConfig, const char *url) {
           paGetToken(value2 + vlen2 + 1, &value3, &vlen3);
           if (vlen3 != 0) value3[vlen3] = 0;
         }
-        cfgSetItem(pConfig, name, value, CFG_STYPE_APOLLO_URL);
         if (value2 != NULL && value3 != NULL && value2[0] != 0 && value3[0] != 0 && strcasecmp(name, "dataDir") == 0) {
-          cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_APOLLO_URL);
+          code = cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_APOLLO_URL);
+          if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
+        } else {
+          code = cfgSetItem(pConfig, name, value, CFG_STYPE_APOLLO_URL);
+          if (code != 0 && terrno != TSDB_CODE_CFG_NOT_FOUND) break;
         }
       }
     }