Merge branch '3.0' of https://github.com/taosdata/TDengine into feature/3.0_mhli

cmake/cmake.options

@@ -18,6 +18,13 @@ IF(${TD_WINDOWS})
             ON
           )
 
+    MESSAGE("build iconv Win32")
+    option(
+            BUILD_WITH_ICONV
+            "If build iconv on Windows"
+            ON
+          )
+
 ENDIF ()
 
 IF(${TD_LINUX} MATCHES TRUE)

cmake/iconv_CMakeLists.txt.in

+
+# iconv
+ExternalProject_Add(iconv
+        GIT_REPOSITORY https://github.com/win-iconv/win-iconv.git
+        GIT_TAG v0.0.8
+        SOURCE_DIR "${CMAKE_CONTRIB_DIR}/iconv"
+        BINARY_DIR ""
+        CONFIGURE_COMMAND ""
+        BUILD_COMMAND ""
+        INSTALL_COMMAND ""
+        TEST_COMMAND ""
+        )
\ No newline at end of file

contrib/CMakeLists.txt

@@ -83,6 +83,11 @@ if(${BUILD_WITH_NURAFT})
     cat("${CMAKE_SUPPORT_DIR}/nuraft_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
 endif(${BUILD_WITH_NURAFT})
 
+# iconv
+if(${BUILD_WITH_ICONV})
+    cat("${CMAKE_SUPPORT_DIR}/iconv_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+endif(${BUILD_WITH_ICONV})
+
 # download dependencies
 configure_file(${CONTRIB_TMP_FILE} "${CMAKE_CONTRIB_DIR}/deps-download/CMakeLists.txt")
 execute_process(COMMAND "${CMAKE_COMMAND}" -G "${CMAKE_GENERATOR}" .
@@ -208,9 +213,10 @@ endif(${BUILD_WITH_TRAFT})
 
 # LIBUV
 if(${BUILD_WITH_UV})
-    if (NOT ${CMAKE_SYSTEM_NAME} MATCHES "Windows")
-        MESSAGE("Windows need set no-sign-compare")
-        add_compile_options(-Wno-sign-compare)
+    if (${TD_WINDOWS})
+        file(READ "libuv/include/uv.h" CONTENTS)
+        string(REGEX REPLACE "/([\r]*)\nstruct uv_tcp_s {" "/\\1\ntypedef BOOL (PASCAL *LPFN_CONNECTEX) (SOCKET s, const struct sockaddr* name, int namelen, PVOID lpSendBuffer, DWORD dwSendDataLength,LPDWORD lpdwBytesSent, LPOVERLAPPED lpOverlapped);\\1\nstruct uv_tcp_s {" CONTENTS_NEW "${CONTENTS}")
+        file(WRITE "libuv/include/uv.h" "${CONTENTS_NEW}")
     endif ()
     add_subdirectory(libuv)
 endif(${BUILD_WITH_UV})
@@ -243,7 +249,15 @@ if(${BUILD_WITH_SQLITE})
 endif(${BUILD_WITH_SQLITE})
 
 # pthread
-
+if(${BUILD_PTHREAD})
+    add_definitions(-DPTW32_STATIC_LIB)
+    add_subdirectory(pthread)
+endif(${BUILD_PTHREAD})
+
+# iconv
+if(${BUILD_WITH_ICONV})
+    add_subdirectory(iconv)
+endif(${BUILD_WITH_ICONV})
 
 # ================================================================================================
 # Build test

include/client/taos.h

@@ -257,9 +257,15 @@ DLL_EXPORT void           tmq_conf_set_offset_commit_cb(tmq_conf_t *conf, tmq_co
 void    tmqShowMsg(tmq_message_t *tmq_message);
 int32_t tmqGetSkipLogNum(tmq_message_t *tmq_message);
 
-typedef void (*TAOS_SUBSCRIBE_CALLBACK)(TAOS_SUB* tsub, TAOS_RES *res, void* param, int code);
+/* -------------------------TMQ MSG HANDLE INTERFACE---------------------- */
 
-DLL_EXPORT int taos_stmt_affected_rows(TAOS_STMT* stmt);
+DLL_EXPORT TAOS_ROW tmq_get_row(tmq_message_t *message);
+DLL_EXPORT char    *tmq_get_topic_name(tmq_message_t *message);
+
+/* ---------------------- OTHER ---------------------------- */
+typedef void (*TAOS_SUBSCRIBE_CALLBACK)(TAOS_SUB *tsub, TAOS_RES *res, void *param, int code);
+
+DLL_EXPORT int taos_stmt_affected_rows(TAOS_STMT *stmt);
 
 #ifdef __cplusplus
 }

include/common/tcommon.h

@@ -62,18 +62,17 @@ typedef struct SDataBlockInfo {
   union {int64_t uid; int64_t blockId;};
 } SDataBlockInfo;
 
-typedef struct SConstantItem {
-  SColumnInfo info;
-  int32_t     startRow;  // run-length-encoding to save the space for multiple rows
-  int32_t     endRow;
-  SVariant    value;
-} SConstantItem;
+//typedef struct SConstantItem {
+//  SColumnInfo info;
+//  int32_t     startRow;  // run-length-encoding to save the space for multiple rows
+//  int32_t     endRow;
+//  SVariant    value;
+//} SConstantItem;
 
 // info.numOfCols = taosArrayGetSize(pDataBlock) + taosArrayGetSize(pConstantList);
 typedef struct SSDataBlock {
   SColumnDataAgg *pBlockAgg;
   SArray         *pDataBlock;    // SArray<SColumnInfoData>
-  SArray         *pConstantList; // SArray<SConstantItem>, it is a constant/tags value of the corresponding result value.
   SDataBlockInfo  info;
 } SSDataBlock;
 
@@ -95,66 +94,15 @@ typedef struct SColumnInfoData {
   };
 } SColumnInfoData;
 
-static FORCE_INLINE int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock) {
-  int64_t tbUid = pBlock->info.uid;
-  int16_t numOfCols = pBlock->info.numOfCols;
-  int16_t hasVarCol = pBlock->info.hasVarCol;
-  int32_t rows = pBlock->info.rows;
-  int32_t sz = taosArrayGetSize(pBlock->pDataBlock);
-
-  int32_t tlen = 0;
-  tlen += taosEncodeFixedI64(buf, tbUid);
-  tlen += taosEncodeFixedI16(buf, numOfCols);
-  tlen += taosEncodeFixedI16(buf, hasVarCol);
-  tlen += taosEncodeFixedI32(buf, rows);
-  tlen += taosEncodeFixedI32(buf, sz);
-  for (int32_t i = 0; i < sz; i++) {
-    SColumnInfoData* pColData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i);
-    tlen += taosEncodeFixedI16(buf, pColData->info.colId);
-    tlen += taosEncodeFixedI16(buf, pColData->info.type);
-    tlen += taosEncodeFixedI32(buf, pColData->info.bytes);
-    int32_t colSz = rows * pColData->info.bytes;
-    tlen += taosEncodeBinary(buf, pColData->pData, colSz);
-  }
-  return tlen;
-}
-
-static FORCE_INLINE void* tDecodeDataBlock(const void* buf, SSDataBlock* pBlock) {
-  int32_t sz;
-
-  buf = taosDecodeFixedI64(buf, &pBlock->info.uid);
-  buf = taosDecodeFixedI16(buf, &pBlock->info.numOfCols);
-  buf = taosDecodeFixedI16(buf, &pBlock->info.hasVarCol);
-  buf = taosDecodeFixedI32(buf, &pBlock->info.rows);
-  buf = taosDecodeFixedI32(buf, &sz);
-  pBlock->pDataBlock = taosArrayInit(sz, sizeof(SColumnInfoData));
-  for (int32_t i = 0; i < sz; i++) {
-    SColumnInfoData data = {0};
-    buf = taosDecodeFixedI16(buf, &data.info.colId);
-    buf = taosDecodeFixedI16(buf, &data.info.type);
-    buf = taosDecodeFixedI32(buf, &data.info.bytes);
-    int32_t colSz = pBlock->info.rows * data.info.bytes;
-    buf = taosDecodeBinary(buf, (void**)&data.pData, colSz);
-    taosArrayPush(pBlock->pDataBlock, &data);
-  }
-  return (void*)buf;
-}
+void*   blockDataDestroy(SSDataBlock* pBlock);
+int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock);
+void*   tDecodeDataBlock(const void* buf, SSDataBlock* pBlock);
 
 static FORCE_INLINE void tDeleteSSDataBlock(SSDataBlock* pBlock) {
   if (pBlock == NULL) {
     return;
   }
-
-  // int32_t numOfOutput = pBlock->info.numOfCols;
-  int32_t sz = taosArrayGetSize(pBlock->pDataBlock);
-  for (int32_t i = 0; i < sz; ++i) {
-    SColumnInfoData* pColInfoData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i);
-    tfree(pColInfoData->pData);
-  }
-
-  taosArrayDestroy(pBlock->pDataBlock);
-  tfree(pBlock->pBlockAgg);
-  // tfree(pBlock);
+  blockDataDestroy(pBlock);
 }
 
 static FORCE_INLINE int32_t tEncodeSMqPollRsp(void** buf, const SMqPollRsp* pRsp) {

include/common/tdatablock.h

@@ -52,6 +52,21 @@ SEpSet getEpSet_s(SCorEpSet* pEpSet);
     BMCharPos(bm_, r_) |= (1u << (7u - BitPos(r_))); \
   } while (0)
 
+static FORCE_INLINE bool colDataIsNull_s(const SColumnInfoData* pColumnInfoData, uint32_t row) {
+  if (!pColumnInfoData->hasNull) {
+    return false;
+  }
+  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
+    return pColumnInfoData->varmeta.offset[row] == -1;
+  } else {
+    if (pColumnInfoData->nullbitmap == NULL) {
+      return false;
+    }
+
+    return colDataIsNull_f(pColumnInfoData->nullbitmap, row);
+  }
+}
+
 static FORCE_INLINE bool colDataIsNull(const SColumnInfoData* pColumnInfoData, uint32_t totalRows, uint32_t row,
                                        SColumnDataAgg* pColAgg) {
   if (!pColumnInfoData->hasNull) {
@@ -79,9 +94,9 @@ static FORCE_INLINE bool colDataIsNull(const SColumnInfoData* pColumnInfoData, u
   }
 }
 
-#define BitmapLen(_n)     (((_n) + ((1<<NBIT)-1)) >> NBIT)
-
+#define BitmapLen(_n) (((_n) + ((1 << NBIT) - 1)) >> NBIT)
 
+// SColumnInfoData, rowNumber
 #define colDataGetData(p1_, r_)                                                        \
   ((IS_VAR_DATA_TYPE((p1_)->info.type)) ? ((p1_)->pData + (p1_)->varmeta.offset[(r_)]) \
                                         : ((p1_)->pData + ((r_) * (p1_)->info.bytes)))

include/common/tmsg.h

@@ -197,6 +197,11 @@ typedef struct {
   };
 } SMsgHead;
 
+typedef struct {
+  int32_t workerType;
+  int32_t streamTaskId;
+} SStreamExecMsgHead;
+
 // Submit message for one table
 typedef struct SSubmitBlk {
   int64_t uid;        // table unique id
@@ -419,7 +424,7 @@ typedef struct {
   };
 } SColumnFilterList;
 /*
- * for client side struct, we only need the column id, type, bytes are not necessary
+ * for client side struct, only column id, type, bytes are necessary
  * But for data in vnode side, we need all the following information.
  */
 typedef struct {
@@ -2047,27 +2052,19 @@ static FORCE_INLINE void* tDecodeTSma(void* buf, STSma* pSma) {
   buf = taosDecodeFixedI64(buf, &pSma->sliding);
 
   if (pSma->exprLen > 0) {
-    pSma->expr = (char*)calloc(pSma->exprLen, 1);
-    if (pSma->expr != NULL) {
-      buf = taosDecodeStringTo(buf, pSma->expr);
-    } else {
+    if ((buf = taosDecodeString(buf, &pSma->expr)) == NULL) {
       tdDestroyTSma(pSma);
       return NULL;
     }
-
   } else {
     pSma->expr = NULL;
   }
 
   if (pSma->tagsFilterLen > 0) {
-    pSma->tagsFilter = (char*)calloc(pSma->tagsFilterLen, 1);
-    if (pSma->tagsFilter != NULL) {
-      buf = taosDecodeStringTo(buf, pSma->tagsFilter);
-    } else {
+    if ((buf = taosDecodeString(buf, &pSma->tagsFilter)) == NULL) {
       tdDestroyTSma(pSma);
       return NULL;
     }
-
   } else {
     pSma->tagsFilter = NULL;
   }
@@ -2158,6 +2155,26 @@ typedef struct {
   SArray* topics;  // SArray<SMqSubTopicEp>
 } SMqCMGetSubEpRsp;
 
+typedef struct {
+  SMqRspHead head;
+  union {
+    SMqPollRsp       consumeRsp;
+    SMqCMGetSubEpRsp getEpRsp;
+  };
+  void* extra;
+} SMqMsgWrapper;
+
+typedef struct {
+  int32_t curBlock;
+  int32_t curRow;
+  void**  uData;
+} SMqRowIter;
+
+struct tmq_message_t_v1 {
+  SMqPollRsp rsp;
+  SMqRowIter iter;
+};
+
 struct tmq_message_t {
   SMqRspHead head;
   union {
@@ -2165,6 +2182,9 @@ struct tmq_message_t {
     SMqCMGetSubEpRsp getEpRsp;
   };
   void*   extra;
+  int32_t curBlock;
+  int32_t curRow;
+  void**  uData;
 };
 
 static FORCE_INLINE void tDeleteSMqSubTopicEp(SMqSubTopicEp* pSubTopicEp) { taosArrayDestroy(pSubTopicEp->vgs); }
@@ -2287,7 +2307,7 @@ typedef struct {
 } SStreamTaskDeployRsp;
 
 typedef struct {
-  SMsgHead head;
+  SStreamExecMsgHead head;
   // TODO: other info needed by task
 } SStreamTaskExecReq;
 

include/common/tmsgdef.h

@@ -189,6 +189,7 @@ enum {
 
   TD_DEF_MSG_TYPE(TDMT_VND_SUBSCRIBE, "vnode-subscribe", SMVSubscribeReq, SMVSubscribeRsp)
   TD_DEF_MSG_TYPE(TDMT_VND_CONSUME, "vnode-consume", SMqCVConsumeReq, SMqCVConsumeRsp)
+  TD_DEF_MSG_TYPE(TDMT_VND_TASK_DEPLOY, "vnode-task-deploy", SStreamTaskDeployReq, SStreamTaskDeployRsp)
 
   TD_DEF_MSG_TYPE(TDMT_VND_CREATE_SMA, "vnode-create-sma", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_CANCEL_SMA, "vnode-cancel-sma", NULL, NULL)

include/common/tvariant.h

@@ -31,7 +31,7 @@ typedef struct SVariant {
     uint64_t u;
     double   d;
     char    *pz;
-    wchar_t *wpz;
+    TdUcs4  *ucs4;
     SArray  *arr;  // only for 'in' query to hold value list, not value for a field
   };
 } SVariant;

include/libs/nodes/plannodes.h

@@ -69,7 +69,7 @@ typedef struct SProjectLogicNode {
 } SProjectLogicNode;
 
 typedef struct SVnodeModifLogicNode {
-  SLogicNode node;;
+  SLogicNode node;
   int32_t msgType;
   SArray* pDataBlocks;
   SVgDataBlocks* pVgDataBlocks;
@@ -124,7 +124,7 @@ typedef struct SSubLogicPlan {
 } SSubLogicPlan;
 
 typedef struct SQueryLogicPlan {
-  ENodeType type;;
+  ENodeType type;
   int32_t totalLevel;
   SNodeList* pTopSubplans;
 } SQueryLogicPlan;
@@ -252,7 +252,7 @@ typedef struct SSubplan {
 } SSubplan;
 
 typedef struct SQueryPlan {
-  ENodeType type;;
+  ENodeType type;
   uint64_t queryId;
   int32_t numOfSubplans;
   SNodeList* pSubplans; // Element is SNodeListNode. The execution level of subplan, starting from 0.

include/libs/qcom/query.h

@@ -49,6 +49,10 @@ typedef struct STableComInfo {
   int32_t rowSize;       // row size of the schema
 } STableComInfo;
 
+typedef struct SIndexMeta {
+
+} SIndexMeta;
+
 /*
  * ASSERT(sizeof(SCTableMeta) == 24)
  * ASSERT(tableType == TSDB_CHILD_TABLE)

include/libs/tfs/tfs.h

@@ -198,6 +198,16 @@ void tfsBasename(const STfsFile *pFile, char *dest);
  */
 void tfsDirname(const STfsFile *pFile, char *dest);
 
+/**
+ * @brief Get the absolute file name of rname.
+ *
+ * @param pTfs
+ * @param diskId
+ * @param rname relative file name
+ * @param aname absolute file name
+ */
+void tfsAbsoluteName(STfs *pTfs, SDiskID diskId, const char *rname, char *aname);
+
 /**
  * @brief Remove file in tfs.
  *

include/libs/transport/trpc.h

@@ -38,11 +38,14 @@ typedef struct SRpcConnInfo {
 
 typedef struct SRpcMsg {
   tmsg_t  msgType;
+  tmsg_t  expectMsgType;
   void *  pCont;
   int     contLen;
   int32_t code;
   void *  handle;   // rpc handle returned to app
   void *  ahandle;  // app handle set by client
+  int     noResp;   // has response or not(default 0 indicate resp);
+
 } SRpcMsg;
 
 typedef struct SRpcInit {

include/os/os.h

@@ -25,10 +25,11 @@ extern "C" {
 #include <pthread.h>
 #include <semaphore.h>
 
+#include <regex.h>
+
 #if !defined(WINDOWS)
 #include <unistd.h>
 #include <dirent.h>
-#include <regex.h>
 #include <sched.h>
 #include <wordexp.h>
 #include <libgen.h>
@@ -36,6 +37,12 @@ extern "C" {
 #include <sys/utsname.h>
 #include <sys/param.h>
 #include <sys/mman.h>
+#include <sys/ioctl.h>
+#include <sys/stat.h>
+#include <sys/time.h>
+#include <sys/types.h>
+#include <termios.h>
+#include <sys/statvfs.h>
 
 #if defined(DARWIN)
 #else
@@ -61,12 +68,7 @@ extern "C" {
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
-#include <sys/ioctl.h>
-#include <sys/stat.h>
-#include <sys/time.h>
-#include <sys/types.h>
 #include <wchar.h>
-#include <termios.h>
 #include <wctype.h>
 
 

include/os/osDef.h

@@ -56,6 +56,7 @@ extern "C" {
   // specific
   typedef int (*__compar_fn_t)(const void *, const void *);
   #define ssize_t int
+  #define _SSIZE_T_
   #define bzero(ptr, size) memset((ptr), 0, (size))
   #define strcasecmp  _stricmp
   #define strncasecmp _strnicmp

include/os/osFile.h

@@ -22,15 +22,6 @@ extern "C" {
 
 #include "osSocket.h"
 
-#if defined(WINDOWS)
-typedef int32_t FileFd;
-typedef SOCKET  SocketFd;
-#else
-typedef int32_t FileFd;
-typedef int32_t SocketFd;
-#endif
-
-int64_t taosRead(FileFd fd, void *buf, int64_t count);
 // If the error is in a third-party library, place this header file under the third-party library header file.
 #ifndef ALLOW_FORBID_FUNC
     #define open OPEN_FUNC_TAOS_FORBID
@@ -42,6 +33,7 @@ int64_t taosRead(FileFd fd, void *buf, int64_t count);
     #define close CLOSE_FUNC_TAOS_FORBID
     #define fclose FCLOSE_FUNC_TAOS_FORBID
     #define fsync FSYNC_FUNC_TAOS_FORBID
+    #define getline GETLINE_FUNC_TAOS_FORBID
     // #define fflush FFLUSH_FUNC_TAOS_FORBID
 #endif
 
@@ -49,15 +41,6 @@ int64_t taosRead(FileFd fd, void *buf, int64_t count);
 #define PATH_MAX 256
 #endif
 
-typedef int32_t FileFd;
-
-typedef struct TdFile {
-  pthread_rwlock_t rwlock;
-  int      refId;
-  FileFd   fd;
-  FILE    *fp;
-} * TdFilePtr, TdFile;
-
 typedef struct TdFile *TdFilePtr;
  
 #define TD_FILE_CTEATE    0x0001
@@ -95,10 +78,6 @@ int64_t taosPReadFile(TdFilePtr pFile, void *buf, int64_t count, int64_t offset)
 int64_t taosWriteFile(TdFilePtr pFile, const void *buf, int64_t count);
 void    taosFprintfFile(TdFilePtr pFile, const char *format, ...);
 
-#if defined(WINDOWS)
-#define __restrict__
-#endif // WINDOWS
-
 int64_t taosGetLineFile(TdFilePtr pFile, char ** __restrict__ ptrBuf);
 
 int32_t taosEOFFile(TdFilePtr pFile);
@@ -111,15 +90,7 @@ int32_t taosRemoveFile(const char *path);
  
 void    taosGetTmpfilePath(const char *inputTmpDir, const char *fileNamePrefix, char *dstPath);
 
-#if defined(_TD_DARWIN_64)
-typedef int32_t SocketFd;
-
-int64_t taosSendFile(SocketFd fdDst, FileFd pFileSrc, int64_t *offset, int64_t size);
-int64_t taosFSendFile(FILE *pFileOut, FILE *pFileIn, int64_t *offset, int64_t size);
-#else
-int64_t taosSendFile(SocketFd fdDst, TdFilePtr pFileSrc, int64_t *offset, int64_t size);
 int64_t taosFSendFile(TdFilePtr pFileOut, TdFilePtr pFileIn, int64_t *offset, int64_t size);
-#endif
 
 void *taosMmapReadOnlyFile(TdFilePtr pFile, int64_t length);
 bool taosValidFile(TdFilePtr pFile);

include/os/osSocket.h

@@ -25,6 +25,8 @@
     #define epoll_create EPOLL_CREATE_FUNC_TAOS_FORBID
     #define epoll_ctl EPOLL_CTL_FUNC_TAOS_FORBID
     #define epoll_wait EPOLL_WAIT_FUNC_TAOS_FORBID
+    #define inet_addr INET_ADDR_FUNC_TAOS_FORBID
+    #define inet_ntoa INET_NTOA_FUNC_TAOS_FORBID
 #endif
 
 #if defined(WINDOWS)

include/os/osString.h

@@ -20,16 +20,28 @@
 extern "C" {
 #endif
 
+typedef wchar_t TdWchar;
+typedef int32_t TdUcs4;
+
+// If the error is in a third-party library, place this header file under the third-party library header file.
+#ifndef ALLOW_FORBID_FUNC
+    #define iconv_open ICONV_OPEN_FUNC_TAOS_FORBID
+    #define iconv_close ICONV_CLOSE_FUNC_TAOS_FORBID
+    #define iconv ICONV_FUNC_TAOS_FORBID
+    #define wcwidth WCWIDTH_FUNC_TAOS_FORBID
+    #define wcswidth WCSWIDTH_FUNC_TAOS_FORBID
+    #define mbtowc MBTOWC_FUNC_TAOS_FORBID
+    #define mbstowcs MBSTOWCS_FUNC_TAOS_FORBID
+    #define wctomb WCTOMB_FUNC_TAOS_FORBID
+    #define wcstombs WCSTOMBS_FUNC_TAOS_FORBID
+    #define wcsncpy WCSNCPY_FUNC_TAOS_FORBID
+    #define wchar_t WCHAR_T_FUNC_TAOS_FORBID
+#endif
+
 #if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
   #define tstrdup(str) _strdup(str)
-  #define tstrndup(str, size) _strndup(str, size)  
-  int32_t tgetline(char **lineptr, size_t *n, FILE *stream);
-  int32_t twcslen(const wchar_t *wcs);
 #else
   #define tstrdup(str) strdup(str)
-  #define tstrndup(str, size) strndup(str, size)
-  #define tgetline(lineptr, n, stream) getline(lineptr, n, stream)
-  #define twcslen wcslen
 #endif
 
 #define tstrncpy(dst, src, size) \
@@ -38,14 +50,22 @@ extern "C" {
     (dst)[(size)-1] = 0;            \
   } while (0)
 
+int32_t taosUcs4len(TdUcs4 *ucs4);
 int64_t taosStr2int64(const char *str);
 
-// USE_LIBICONV
-int32_t taosUcs4ToMbs(void *ucs4, int32_t ucs4_max_len, char *mbs);
-bool    taosMbsToUcs4(const char *mbs, size_t mbs_len, char *ucs4, int32_t ucs4_max_len, int32_t *len);
-int32_t tasoUcs4Compare(void *f1_ucs4, void *f2_ucs4, int32_t bytes, int8_t ncharSize);
+int32_t taosUcs4ToMbs(TdUcs4 *ucs4, int32_t ucs4_max_len, char *mbs);
+bool    taosMbsToUcs4(const char *mbs, size_t mbs_len, TdUcs4 *ucs4, int32_t ucs4_max_len, int32_t *len);
+int32_t tasoUcs4Compare(TdUcs4 *f1_ucs4, TdUcs4 *f2_ucs4, int32_t bytes);
+TdUcs4* tasoUcs4Copy(TdUcs4 *target_ucs4, TdUcs4 *source_ucs4, int32_t len_ucs4);
 bool    taosValidateEncodec(const char *encodec);
 
+int32_t taosWcharWidth(TdWchar wchar);
+int32_t taosWcharsWidth(TdWchar *pWchar, int32_t size);
+int32_t taosMbToWchar(TdWchar *pWchar, const char *pStr, int32_t size);
+int32_t taosMbsToWchars(TdWchar *pWchars, const char *pStrs, int32_t size);
+int32_t taosWcharToMb(char *pStr, TdWchar wchar);
+int32_t taosWcharsToMbs(char *pStrs, TdWchar *pWchars, int32_t size);
+
 #ifdef __cplusplus
 }
 #endif

include/os/osSysinfo.h

@@ -16,7 +16,6 @@
 #ifndef _TD_OS_SYSINFO_H_
 #define _TD_OS_SYSINFO_H_
 
-#include <sys/statvfs.h>
 #include "os.h"
 
 #ifdef __cplusplus

include/util/tcompare.h

@@ -46,7 +46,7 @@ typedef struct SPatternCompareInfo {
 
 int32_t patternMatch(const char *pattern, const char *str, size_t size, const SPatternCompareInfo *pInfo);
 
-int32_t WCSPatternMatch(const wchar_t *pattern, const wchar_t *str, size_t size, const SPatternCompareInfo *pInfo);
+int32_t WCSPatternMatch(const TdUcs4 *pattern, const TdUcs4 *str, size_t size, const SPatternCompareInfo *pInfo);
 
 int32_t taosArrayCompareString(const void *a, const void *b);
 

include/util/tdef.h

@@ -41,7 +41,7 @@ extern const int32_t TYPE_BYTES[15];
 #define DOUBLE_BYTES    sizeof(double)
 #define POINTER_BYTES   sizeof(void *)  // 8 by default  assert(sizeof(ptrdiff_t) == sizseof(void*)
 #define TSDB_KEYSIZE    sizeof(TSKEY)
-#define TSDB_NCHAR_SIZE sizeof(int32_t)
+#define TSDB_NCHAR_SIZE sizeof(TdUcs4)
 
 // NULL definition
 #define TSDB_DATA_BOOL_NULL      0x02
@@ -448,6 +448,11 @@ typedef struct {
 #define SND_UNIQUE_THREAD_NUM 2
 #define SND_SHARED_THREAD_NUM 2
 
+enum {
+  SND_WORKER_TYPE__SHARED = 1,
+  SND_WORKER_TYPE__UNIQUE,
+};
+
 #ifdef __cplusplus
 }
 #endif

include/util/tencode.h

@@ -75,7 +75,7 @@ typedef struct {
 #define TD_CODER_CURRENT(CODER)                        ((CODER)->data + (CODER)->pos)
 #define TD_CODER_MOVE_POS(CODER, MOVE)                 ((CODER)->pos += (MOVE))
 #define TD_CODER_CHECK_CAPACITY_FAILED(CODER, EXPSIZE) (((CODER)->size - (CODER)->pos) < (EXPSIZE))
-#define TCODER_MALLOC(SIZE, CODER)                     TFL_MALLOC(SIZE, &((CODER)->fl))
+#define TCODER_MALLOC(PTR, TYPE, SIZE, CODER)          TFL_MALLOC(PTR, TYPE, SIZE, &((CODER)->fl))
 
 void tCoderInit(SCoder* pCoder, td_endian_t endian, uint8_t* data, int32_t size, td_coder_t type);
 void tCoderClear(SCoder* pCoder);

include/util/tfreelist.h

@@ -29,15 +29,17 @@ struct SFreeListNode {
 
 typedef TD_SLIST(SFreeListNode) SFreeList;
 
-#define TFL_MALLOC(SIZE, LIST)                                 \
-  ({                                                           \
+#define TFL_MALLOC(PTR, TYPE, SIZE, LIST)                      \
+  do {                                                         \
     void *ptr = malloc((SIZE) + sizeof(struct SFreeListNode)); \
     if (ptr) {                                                 \
       TD_SLIST_PUSH((LIST), (struct SFreeListNode *)ptr);      \
       ptr = ((struct SFreeListNode *)ptr)->payload;            \
+      (PTR) = (TYPE)(ptr);                                     \
+    }else{                                                     \
+      (PTR) = NULL;                                            \
     }                                                          \
-    ptr;                                                       \
-  })
+  }while(0);
 
 #define tFreeListInit(pFL) TD_SLIST_INIT(pFL)
 

include/util/types.h

@@ -82,7 +82,7 @@ typedef uint16_t VarDataLenT;  // maxVarDataLen: 32767
 #define VARSTR_HEADER_SIZE sizeof(VarDataLenT)
 
 #define varDataLen(v) ((VarDataLenT *)(v))[0]
-#define varDataVal(v) ((void *)((char *)v + VARSTR_HEADER_SIZE))
+#define varDataVal(v) ((char *)(v) + VARSTR_HEADER_SIZE)
 
 typedef int32_t VarDataOffsetT;
 

source/client/src/tmq.c

@@ -700,6 +700,10 @@ int32_t tmqPollCb(void* param, const SDataBuf* pMsg, int32_t code) {
   }
   memcpy(pRsp, pMsg->pData, sizeof(SMqRspHead));
   tDecodeSMqPollRsp(POINTER_SHIFT(pMsg->pData, sizeof(SMqRspHead)), &pRsp->consumeRsp);
+  pRsp->curBlock = 0;
+  pRsp->curRow = 0;
+  // TODO: alloc mem
+  /*pRsp->*/
   /*printf("rsp commit off:%ld rsp off:%ld has data:%d\n", pRsp->committedOffset, pRsp->rspOffset, pRsp->numOfTopics);*/
   if (pRsp->consumeRsp.numOfTopics == 0) {
     /*printf("no data\n");*/
@@ -758,9 +762,9 @@ int32_t tmqAskEpCb(void* param, const SDataBuf* pMsg, int32_t code) {
     goto END;
   }
 
-  // tmq's epoch is monotomically increase,
+  // tmq's epoch is monotonically increase,
   // so it's safe to discard any old epoch msg.
-  // epoch will only increase when received newer epoch ep msg
+  // Epoch will only increase when received newer epoch ep msg
   SMqRspHead* head = pMsg->pData;
   int32_t     epoch = atomic_load_32(&tmq->epoch);
   if (head->epoch <= epoch) {
@@ -1282,6 +1286,34 @@ const char* tmq_err2str(tmq_resp_err_t err) {
   return "fail";
 }
 
+TAOS_ROW tmq_get_row(tmq_message_t* message) {
+  SMqPollRsp* rsp = &message->consumeRsp;
+  while (1) {
+    if (message->curBlock < taosArrayGetSize(rsp->pBlockData)) {
+      SSDataBlock* pBlock = taosArrayGet(rsp->pBlockData, message->curBlock);
+      if (message->curRow < pBlock->info.rows) {
+        for (int i = 0; i < pBlock->info.numOfCols; i++) {
+          SColumnInfoData* pData = taosArrayGet(pBlock->pDataBlock, i);
+          if (colDataIsNull_s(pData, message->curRow))
+            message->uData[i] = NULL;
+          else {
+            message->uData[i] = colDataGetData(pData, message->curRow);
+          }
+        }
+        message->curRow++;
+        return message->uData;
+      } else {
+        message->curBlock++;
+        message->curRow = 0;
+        continue;
+      }
+    }
+    return NULL;
+  }
+}
+
+char* tmq_get_topic_name(tmq_message_t* message) { return "not implemented yet"; }
+
 #if 0
 tmq_t* tmqCreateConsumerImpl(TAOS* conn, tmq_conf_t* conf) {
   tmq_t* pTmq = malloc(sizeof(tmq_t));

source/common/src/tdatablock.c

@@ -240,10 +240,7 @@ int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, uint32_t numOfRow1, co
 }
 
 size_t blockDataGetNumOfCols(const SSDataBlock* pBlock) {
-  ASSERT(pBlock);
-
-  size_t constantCols = (pBlock->pConstantList != NULL)? taosArrayGetSize(pBlock->pConstantList):0;
-  ASSERT( pBlock->info.numOfCols == taosArrayGetSize(pBlock->pDataBlock) + constantCols);
+  ASSERT(pBlock && pBlock->info.numOfCols == taosArrayGetSize(pBlock->pDataBlock));
   return pBlock->info.numOfCols;
 }
 
@@ -1166,3 +1163,67 @@ SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock) {
 size_t blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize) {
   return pageSize / (blockDataGetSerialRowSize(pBlock) + blockDataGetSerialMetaSize(pBlock));
 }
+
+int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock) {
+  int64_t tbUid = pBlock->info.uid;
+  int16_t numOfCols = pBlock->info.numOfCols;
+  int16_t hasVarCol = pBlock->info.hasVarCol;
+  int32_t rows = pBlock->info.rows;
+  int32_t sz = taosArrayGetSize(pBlock->pDataBlock);
+
+  int32_t tlen = 0;
+  tlen += taosEncodeFixedI64(buf, tbUid);
+  tlen += taosEncodeFixedI16(buf, numOfCols);
+  tlen += taosEncodeFixedI16(buf, hasVarCol);
+  tlen += taosEncodeFixedI32(buf, rows);
+  tlen += taosEncodeFixedI32(buf, sz);
+  for (int32_t i = 0; i < sz; i++) {
+    SColumnInfoData* pColData = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, i);
+    tlen += taosEncodeFixedI16(buf, pColData->info.colId);
+    tlen += taosEncodeFixedI16(buf, pColData->info.type);
+    tlen += taosEncodeFixedI32(buf, pColData->info.bytes);
+
+    if (IS_VAR_DATA_TYPE(pColData->info.type)) {
+      tlen += taosEncodeBinary(buf, pColData->varmeta.offset, sizeof(int32_t) * rows);
+    } else {
+      tlen += taosEncodeBinary(buf, pColData->nullbitmap, BitmapLen(rows));
+    }
+
+    int32_t len = colDataGetLength(pColData, rows);
+    taosEncodeFixedI32(buf, len);
+
+    tlen += taosEncodeBinary(buf, pColData->pData, len);
+  }
+  return tlen;
+}
+
+void* tDecodeDataBlock(const void* buf, SSDataBlock* pBlock) {
+  int32_t sz;
+
+  buf = taosDecodeFixedI64(buf, &pBlock->info.uid);
+  buf = taosDecodeFixedI16(buf, &pBlock->info.numOfCols);
+  buf = taosDecodeFixedI16(buf, &pBlock->info.hasVarCol);
+  buf = taosDecodeFixedI32(buf, &pBlock->info.rows);
+  buf = taosDecodeFixedI32(buf, &sz);
+  pBlock->pDataBlock = taosArrayInit(sz, sizeof(SColumnInfoData));
+  for (int32_t i = 0; i < sz; i++) {
+    SColumnInfoData data = {0};
+    buf = taosDecodeFixedI16(buf, &data.info.colId);
+    buf = taosDecodeFixedI16(buf, &data.info.type);
+    buf = taosDecodeFixedI32(buf, &data.info.bytes);
+
+    if (IS_VAR_DATA_TYPE(data.info.type)) {
+      buf = taosDecodeBinary(buf, (void**)&data.varmeta.offset, pBlock->info.rows * sizeof(int32_t));
+      data.varmeta.length = pBlock->info.rows * sizeof(int32_t);
+      data.varmeta.allocLen = data.varmeta.length;
+    } else {
+      buf = taosDecodeBinary(buf, (void**)&data.nullbitmap, BitmapLen(pBlock->info.rows));
+    }
+
+    int32_t len = 0;
+    buf = taosDecodeFixedI32(buf, &len);
+    buf = taosDecodeBinary(buf, (void**)&data.pData, len);
+    taosArrayPush(pBlock->pDataBlock, &data);
+  }
+  return (void*)buf;
+}
\ No newline at end of file

source/common/src/tmsg.c

@@ -2467,7 +2467,7 @@ int32_t tEncodeSMqCMCommitOffsetReq(SCoder *encoder, const SMqCMCommitOffsetReq
 int32_t tDecodeSMqCMCommitOffsetReq(SCoder *decoder, SMqCMCommitOffsetReq *pReq) {
   if (tStartDecode(decoder) < 0) return -1;
   if (tDecodeI32(decoder, &pReq->num) < 0) return -1;
-  pReq->offsets = TCODER_MALLOC(pReq->num * sizeof(SMqOffset), decoder);
+  TCODER_MALLOC(pReq->offsets, SMqOffset*, pReq->num * sizeof(SMqOffset), decoder);
   if (pReq->offsets == NULL) return -1;
   for (int32_t i = 0; i < pReq->num; i++) {
     tDecodeSMqOffset(decoder, &pReq->offsets[i]);

source/common/src/tname.c

@@ -217,7 +217,7 @@ int32_t tNameSetDbName(SName* dst, int32_t acct, const char* dbName, size_t name
 }
 
 int32_t tNameSetAcctId(SName* dst, int32_t acctId) {
-  assert(dst != NULL && acct != NULL);
+  assert(dst != NULL);
   dst->acctId = acctId;
   return 0;
 }

source/common/src/tvariant.c

@@ -199,8 +199,8 @@ void taosVariantCreateFromBinary(SVariant *pVar, const char *pz, size_t len, uin
     case TSDB_DATA_TYPE_NCHAR: {  // here we get the nchar length from raw binary bits length
       size_t lenInwchar = len / TSDB_NCHAR_SIZE;
 
-      pVar->wpz = calloc(1, (lenInwchar + 1) * TSDB_NCHAR_SIZE);
-      memcpy(pVar->wpz, pz, lenInwchar * TSDB_NCHAR_SIZE);
+      pVar->ucs4 = calloc(1, (lenInwchar + 1) * TSDB_NCHAR_SIZE);
+      memcpy(pVar->ucs4, pz, lenInwchar * TSDB_NCHAR_SIZE);
       pVar->nLen = (int32_t)len;
 
       break;
@@ -343,7 +343,7 @@ int32_t taosVariantToString(SVariant *pVar, char *dst) {
 
     case TSDB_DATA_TYPE_NCHAR: {
       dst[0] = '\'';
-      taosUcs4ToMbs(pVar->wpz, (twcslen(pVar->wpz) + 1) * TSDB_NCHAR_SIZE, dst + 1);
+      taosUcs4ToMbs(pVar->ucs4, (taosUcs4len(pVar->ucs4) + 1) * TSDB_NCHAR_SIZE, dst + 1);
       int32_t len = (int32_t)strlen(dst);
       dst[len] = '\'';
       dst[len + 1] = 0;
@@ -384,7 +384,7 @@ static FORCE_INLINE int32_t convertToBoolImpl(char *pStr, int32_t len) {
   }
 }
 
-static FORCE_INLINE int32_t wcsconvertToBoolImpl(wchar_t *pstr, int32_t len) {
+static FORCE_INLINE int32_t wcsconvertToBoolImpl(TdUcs4 *pstr, int32_t len) {
   if ((wcsncasecmp(pstr, L"true", len) == 0) && (len == 4)) {
     return TSDB_TRUE;
   } else if (wcsncasecmp(pstr, L"false", len) == 0 && (len == 5)) {
@@ -412,11 +412,11 @@ static int32_t toBinary(SVariant *pVariant, char **pDest, int32_t *pDestSize) {
         pBuf = realloc(pBuf, newSize + 1);
       }
 
-      taosUcs4ToMbs(pVariant->wpz, (int32_t)newSize, pBuf);
-      free(pVariant->wpz);
+      taosUcs4ToMbs(pVariant->ucs4, (int32_t)newSize, pBuf);
+      free(pVariant->ucs4);
       pBuf[newSize] = 0;
     } else {
-      taosUcs4ToMbs(pVariant->wpz, (int32_t)newSize, *pDest);
+      taosUcs4ToMbs(pVariant->ucs4, (int32_t)newSize, *pDest);
     }
 
   } else {
@@ -460,8 +460,8 @@ static int32_t toNchar(SVariant *pVariant, char **pDest, int32_t *pDestSize) {
   }
 
   if (*pDest == pVariant->pz) {
-    wchar_t *pWStr = calloc(1, (nLen + 1) * TSDB_NCHAR_SIZE);
-    bool     ret = taosMbsToUcs4(pDst, nLen, (char *)pWStr, (nLen + 1) * TSDB_NCHAR_SIZE, NULL);
+    TdUcs4 *pWStr = calloc(1, (nLen + 1) * TSDB_NCHAR_SIZE);
+    bool     ret = taosMbsToUcs4(pDst, nLen, pWStr, (nLen + 1) * TSDB_NCHAR_SIZE, NULL);
     if (!ret) {
       tfree(pWStr);
       return -1;
@@ -469,21 +469,21 @@ static int32_t toNchar(SVariant *pVariant, char **pDest, int32_t *pDestSize) {
 
     // free the binary buffer in the first place
     if (pVariant->nType == TSDB_DATA_TYPE_BINARY) {
-      free(pVariant->wpz);
+      free(pVariant->ucs4);
     }
 
-    pVariant->wpz = pWStr;
-    *pDestSize = twcslen(pVariant->wpz);
+    pVariant->ucs4 = pWStr;
+    *pDestSize = taosUcs4len(pVariant->ucs4);
 
     // shrink the allocate memory, no need to check here.
-    char *tmp = realloc(pVariant->wpz, (*pDestSize + 1) * TSDB_NCHAR_SIZE);
+    char *tmp = realloc(pVariant->ucs4, (*pDestSize + 1) * TSDB_NCHAR_SIZE);
     assert(tmp != NULL);
 
-    pVariant->wpz = (wchar_t *)tmp;
+    pVariant->ucs4 = (TdUcs4 *)tmp;
   } else {
     int32_t output = 0;
 
-    bool ret = taosMbsToUcs4(pDst, nLen, *pDest, (nLen + 1) * TSDB_NCHAR_SIZE, &output);
+    bool ret = taosMbsToUcs4(pDst, nLen, (TdUcs4*)*pDest, (nLen + 1) * TSDB_NCHAR_SIZE, &output);
     if (!ret) {
       return -1;
     }
@@ -554,7 +554,7 @@ static FORCE_INLINE int32_t convertToInteger(SVariant *pVariant, int64_t *result
     *result = res;
   } else if (pVariant->nType == TSDB_DATA_TYPE_NCHAR) {
     errno = 0;
-    wchar_t *endPtr = NULL;
+    TdUcs4 *endPtr = NULL;
     
     SToken token = {0};
     token.n = tGetToken(pVariant->pz, &token.type);
@@ -564,7 +564,7 @@ static FORCE_INLINE int32_t convertToInteger(SVariant *pVariant, int64_t *result
     }
     
     if (token.type == TK_FLOAT) {
-      double v = wcstod(pVariant->wpz, &endPtr);
+      double v = wcstod(pVariant->ucs4, &endPtr);
       if (releaseVariantPtr) {
         free(pVariant->pz);
         pVariant->nLen = 0;
@@ -583,7 +583,7 @@ static FORCE_INLINE int32_t convertToInteger(SVariant *pVariant, int64_t *result
       setNull((char *)result, type, tDataTypes[type].bytes);
       return 0;
     } else {
-      int64_t val = wcstoll(pVariant->wpz, &endPtr, 10);
+      int64_t val = wcstoll(pVariant->ucs4, &endPtr, 10);
       if (releaseVariantPtr) {
         free(pVariant->pz);
         pVariant->nLen = 0;
@@ -649,7 +649,7 @@ static int32_t convertToBool(SVariant *pVariant, int64_t *pDest) {
     *pDest = ret;
   } else if (pVariant->nType == TSDB_DATA_TYPE_NCHAR) {
     int32_t ret = 0;
-    if ((ret = wcsconvertToBoolImpl(pVariant->wpz, pVariant->nLen)) < 0) {
+    if ((ret = wcsconvertToBoolImpl(pVariant->ucs4, pVariant->nLen)) < 0) {
       return ret;
     }
     *pDest = ret;
@@ -899,7 +899,7 @@ int32_t tVariantDumpEx(SVariant *pVariant, char *payload, int16_t type, bool inc
               return -1;
             }
           } else {
-            wcsncpy((wchar_t *)payload, pVariant->wpz, pVariant->nLen);
+            tasoUcs4Copy((TdUcs4*)payload, pVariant->ucs4, pVariant->nLen);
           }
         }
       } else {
@@ -913,7 +913,7 @@ int32_t tVariantDumpEx(SVariant *pVariant, char *payload, int16_t type, bool inc
               return -1;
             }
           } else {
-            memcpy(p, pVariant->wpz, pVariant->nLen);
+            memcpy(p, pVariant->ucs4, pVariant->nLen);
             newlen = pVariant->nLen;
           }
 
@@ -979,7 +979,7 @@ int32_t taosVariantTypeSetType(SVariant *pVariant, char type) {
         pVariant->d = v;
       } else if (pVariant->nType == TSDB_DATA_TYPE_NCHAR) {
         errno = 0;
-        double v = wcstod(pVariant->wpz, NULL);
+        double v = wcstod(pVariant->ucs4, NULL);
         if ((errno == ERANGE && v == -1) || (isinf(v) || isnan(v))) {
           free(pVariant->pz);
           return -1;

source/dnode/mgmt/daemon/src/dmnMain.c

@@ -27,7 +27,7 @@ static struct {
 } dmn = {0};
 
 static void dmnSigintHandle(int signum, void *info, void *ctx) {
-  uInfo("singal:%d is received", signum);
+  uInfo("signal:%d is received", signum);
   dmn.stop = true;
 }
 

source/dnode/mgmt/impl/inc/dndSnode.h

@@ -24,10 +24,15 @@ extern "C" {
 int32_t dndInitSnode(SDnode *pDnode);
 void    dndCleanupSnode(SDnode *pDnode);
 
-void    dndProcessSnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
+// void    dndProcessSnodeWriteMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
 int32_t dndProcessCreateSnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg);
 int32_t dndProcessDropSnodeReq(SDnode *pDnode, SRpcMsg *pRpcMsg);
 
+void dndProcessSnodeMgmtMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
+void dndProcessSnodeUniqueMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
+void dndProcessSnodeSharedMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
+void dndProcessSnodeExecMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet);
+
 #ifdef __cplusplus
 }
 #endif

source/dnode/mgmt/impl/src/dndSnode.c

@@ -382,6 +382,12 @@ static void dndProcessSnodeSharedQueue(SDnode *pDnode, SRpcMsg *pMsg) {
   taosFreeQitem(pMsg);
 }
 
+static FORCE_INLINE int32_t dndGetSWTypeFromMsg(SRpcMsg *pMsg) {
+  SStreamExecMsgHead *pHead = pMsg->pCont;
+  pHead->workerType = htonl(pHead->workerType);
+  return pHead->workerType;
+}
+
 static FORCE_INLINE int32_t dndGetSWIdFromMsg(SRpcMsg *pMsg) {
   SMsgHead *pHead = pMsg->pCont;
   pHead->streamTaskId = htonl(pHead->streamTaskId);
@@ -450,6 +456,18 @@ void dndProcessSnodeMgmtMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
   dndWriteSnodeMsgToMgmtWorker(pDnode, pMsg);
 }
 
+void dndProcessSnodeExecMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
+  SSnode *pSnode = dndAcquireSnode(pDnode);
+  if (pSnode != NULL) {
+    int32_t workerType = dndGetSWTypeFromMsg(pMsg);
+    if (workerType == SND_WORKER_TYPE__SHARED) {
+      dndWriteSnodeMsgToWorker(pDnode, &pDnode->smgmt.sharedWorker, pMsg);
+    } else {
+      dndWriteSnodeMsgToWorkerByMsg(pDnode, pMsg);
+    }
+  }
+}
+
 void dndProcessSnodeUniqueMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
   dndWriteSnodeMsgToWorkerByMsg(pDnode, pMsg);
 }

source/dnode/mgmt/impl/src/dndTransport.c

@@ -23,6 +23,7 @@
 #include "dndTransport.h"
 #include "dndMgmt.h"
 #include "dndMnode.h"
+#include "dndSnode.h"
 #include "dndVnodes.h"
 
 #define INTERNAL_USER   "_dnd"
@@ -153,10 +154,14 @@ static void dndInitMsgFp(STransMgmt *pMgmt) {
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_SET_CUR)] = dndProcessVnodeFetchMsg;
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_CONSUME)] = dndProcessVnodeFetchMsg;
   pMgmt->msgFp[TMSG_INDEX(TDMT_VND_QUERY_HEARTBEAT)] = dndProcessVnodeFetchMsg;
+
+  // Requests handled by SNODE
+  pMgmt->msgFp[TMSG_INDEX(TDMT_SND_TASK_DEPLOY)] = dndProcessSnodeMgmtMsg;
+  pMgmt->msgFp[TMSG_INDEX(TDMT_SND_TASK_EXEC)] = dndProcessSnodeExecMsg;
 }
 
 static void dndProcessResponse(void *parent, SRpcMsg *pRsp, SEpSet *pEpSet) {
-  SDnode *    pDnode = parent;
+  SDnode     *pDnode = parent;
   STransMgmt *pMgmt = &pDnode->tmgmt;
 
   tmsg_t msgType = pRsp->msgType;
@@ -219,7 +224,7 @@ static void dndCleanupClient(SDnode *pDnode) {
 }
 
 static void dndProcessRequest(void *param, SRpcMsg *pReq, SEpSet *pEpSet) {
-  SDnode *    pDnode = param;
+  SDnode     *pDnode = param;
   STransMgmt *pMgmt = &pDnode->tmgmt;
 
   tmsg_t msgType = pReq->msgType;
@@ -313,7 +318,7 @@ static int32_t dndRetrieveUserAuthInfo(void *parent, char *user, char *spi, char
   SAuthReq authReq = {0};
   tstrncpy(authReq.user, user, TSDB_USER_LEN);
   int32_t contLen = tSerializeSAuthReq(NULL, 0, &authReq);
-  void *  pReq = rpcMallocCont(contLen);
+  void   *pReq = rpcMallocCont(contLen);
   tSerializeSAuthReq(pReq, contLen, &authReq);
 
   SRpcMsg rpcMsg = {.pCont = pReq, .contLen = contLen, .msgType = TDMT_MND_AUTH, .ahandle = (void *)9528};

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

@@ -27,6 +27,22 @@ int32_t tEncodeSStreamObj(SCoder *pEncoder, const SStreamObj *pObj) {
   if (tEncodeCStr(pEncoder, pObj->sql) < 0) return -1;
   if (tEncodeCStr(pEncoder, pObj->logicalPlan) < 0) return -1;
   if (tEncodeCStr(pEncoder, pObj->physicalPlan) < 0) return -1;
+  // TODO encode tasks
+  if (pObj->tasks) {
+    int32_t sz = taosArrayGetSize(pObj->tasks);
+    tEncodeI32(pEncoder, sz);
+    for (int32_t i = 0; i < sz; i++) {
+      SArray *pArray = taosArrayGet(pObj->tasks, i);
+      int32_t innerSz = taosArrayGetSize(pArray);
+      tEncodeI32(pEncoder, innerSz);
+      for (int32_t j = 0; j < innerSz; j++) {
+        SStreamTask *pTask = taosArrayGet(pArray, j);
+        tEncodeSStreamTask(pEncoder, pTask);
+      }
+    }
+  } else {
+    tEncodeI32(pEncoder, 0);
+  }
   return pEncoder->pos;
 }
 
@@ -42,5 +58,23 @@ int32_t tDecodeSStreamObj(SCoder *pDecoder, SStreamObj *pObj) {
   if (tDecodeCStrAlloc(pDecoder, &pObj->sql) < 0) return -1;
   if (tDecodeCStrAlloc(pDecoder, &pObj->logicalPlan) < 0) return -1;
   if (tDecodeCStrAlloc(pDecoder, &pObj->physicalPlan) < 0) return -1;
+  int32_t sz;
+  if (tDecodeI32(pDecoder, &sz) < 0) return -1;
+  if (sz != 0) {
+    pObj->tasks = taosArrayInit(sz, sizeof(SArray));
+    for (int32_t i = 0; i < sz; i++) {
+      int32_t innerSz;
+      if (tDecodeI32(pDecoder, &innerSz) < 0) return -1;
+      SArray *pArray = taosArrayInit(innerSz, sizeof(SStreamTask));
+      for (int32_t j = 0; j < innerSz; j++) {
+        SStreamTask task;
+        if (tDecodeSStreamTask(pDecoder, &task) < 0) return -1;
+        taosArrayPush(pArray, &task);
+      }
+      taosArrayPush(pObj->tasks, pArray);
+    }
+  } else {
+    pObj->tasks = NULL;
+  }
   return 0;
 }

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

@@ -77,6 +77,32 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
           return -1;
         }
         taosArrayPush(taskOneLevel, pTask);
+
+        SCoder encoder;
+        tCoderInit(&encoder, TD_LITTLE_ENDIAN, NULL, 0, TD_ENCODER);
+        tEncodeSStreamTask(&encoder, pTask);
+        int32_t tlen = sizeof(SMsgHead) + encoder.pos;
+        tCoderClear(&encoder);
+        void* buf = rpcMallocCont(tlen);
+        if (buf == NULL) {
+          terrno = TSDB_CODE_OUT_OF_MEMORY;
+          return -1;
+        }
+        ((SMsgHead*)buf)->streamTaskId = pTask->taskId;
+        void* abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
+        tCoderInit(&encoder, TD_LITTLE_ENDIAN, abuf, tlen, TD_ENCODER);
+        tEncodeSStreamTask(&encoder, pTask);
+        tCoderClear(&encoder);
+
+        STransAction action = {0};
+        action.epSet = plan->execNode.epSet;
+        action.pCont = buf;
+        action.contLen = tlen;
+        action.msgType = TDMT_VND_TASK_DEPLOY;
+        if (mndTransAppendRedoAction(pTrans, &action) != 0) {
+          rpcFreeCont(buf);
+          return -1;
+        }
       }
     } else if (plan->subplanType == SUBPLAN_TYPE_SCAN) {
       // duplicatable
@@ -101,6 +127,32 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
         }
 
         taosArrayPush(taskOneLevel, pTask);
+
+        SCoder encoder;
+        tCoderInit(&encoder, TD_LITTLE_ENDIAN, NULL, 0, TD_ENCODER);
+        tEncodeSStreamTask(&encoder, pTask);
+        int32_t tlen = sizeof(SMsgHead) + encoder.pos;
+        tCoderClear(&encoder);
+        void* buf = rpcMallocCont(tlen);
+        if (buf == NULL) {
+          terrno = TSDB_CODE_OUT_OF_MEMORY;
+          return -1;
+        }
+        ((SMsgHead*)buf)->streamTaskId = pTask->taskId;
+        void* abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
+        tCoderInit(&encoder, TD_LITTLE_ENDIAN, abuf, tlen, TD_ENCODER);
+        tEncodeSStreamTask(&encoder, pTask);
+        tCoderClear(&encoder);
+
+        STransAction action = {0};
+        action.epSet = plan->execNode.epSet;
+        action.pCont = buf;
+        action.contLen = tlen;
+        action.msgType = TDMT_SND_TASK_DEPLOY;
+        if (mndTransAppendRedoAction(pTrans, &action) != 0) {
+          rpcFreeCont(buf);
+          return -1;
+        }
       }
     } else {
       // not duplicatable
@@ -117,6 +169,32 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
         return -1;
       }
       taosArrayPush(taskOneLevel, pTask);
+
+      SCoder encoder;
+      tCoderInit(&encoder, TD_LITTLE_ENDIAN, NULL, 0, TD_ENCODER);
+      tEncodeSStreamTask(&encoder, pTask);
+      int32_t tlen = sizeof(SMsgHead) + encoder.pos;
+      tCoderClear(&encoder);
+      void* buf = rpcMallocCont(tlen);
+      if (buf == NULL) {
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
+        return -1;
+      }
+      ((SMsgHead*)buf)->streamTaskId = pTask->taskId;
+      void* abuf = POINTER_SHIFT(buf, sizeof(SMsgHead));
+      tCoderInit(&encoder, TD_LITTLE_ENDIAN, abuf, tlen, TD_ENCODER);
+      tEncodeSStreamTask(&encoder, pTask);
+      tCoderClear(&encoder);
+
+      STransAction action = {0};
+      action.epSet = plan->execNode.epSet;
+      action.pCont = buf;
+      action.contLen = tlen;
+      action.msgType = TDMT_SND_TASK_DEPLOY;
+      if (mndTransAppendRedoAction(pTrans, &action) != 0) {
+        rpcFreeCont(buf);
+        return -1;
+      }
     }
     taosArrayPush(pStream->tasks, taskOneLevel);
   }

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

@@ -230,19 +230,19 @@ static int32_t mndCreateStream(SMnode *pMnode, SMnodeMsg *pReq, SCMCreateStreamR
   }
   mDebug("trans:%d, used to create stream:%s", pTrans->id, pCreate->name);
 
-  SSdbRaw *pRedoRaw = mndStreamActionEncode(&streamObj);
-  if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) {
-    mError("trans:%d, failed to append redo log since %s", pTrans->id, terrstr());
+  if (mndScheduleStream(pMnode, pTrans, &streamObj) < 0) {
+    mError("stream:%ld, schedule stream since %s", streamObj.uid, terrstr());
     mndTransDrop(pTrans);
     return -1;
   }
-  sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
 
-  if (mndScheduleStream(pMnode, pTrans, &streamObj) < 0) {
-    mError("stream:%ld, schedule stream since %s", streamObj.uid, terrstr());
+  SSdbRaw *pRedoRaw = mndStreamActionEncode(&streamObj);
+  if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) {
+    mError("trans:%d, failed to append redo log since %s", pTrans->id, terrstr());
     mndTransDrop(pTrans);
     return -1;
   }
+  sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY);
 
   if (mndTransPrepare(pMnode, pTrans) != 0) {
     mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());

source/dnode/vnode/inc/tq.h

@@ -55,6 +55,7 @@ int tqCommit(STQ*);
 int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessSetConnReq(STQ* pTq, char* msg);
 int32_t tqProcessRebReq(STQ* pTq, char* msg);
+int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen);
 
 #ifdef __cplusplus
 }

source/dnode/vnode/inc/tsdb.h

@@ -107,10 +107,8 @@ int32_t tsdbUpdateSmaWindow(STsdb *pTsdb, int8_t smaType, char *msg);
 int32_t tsdbInsertRSmaData(STsdb *pTsdb, char *msg);
 
 // TODO: This is the basic params, and should wrap the params to a queryHandle.
-int32_t tsdbGetTSmaData(STsdb *pTsdb, STSmaDataWrapper *pData, int64_t indexUid, int64_t interval,
-                            int8_t intervalUnit, tb_uid_t tableUid, col_id_t colId, TSKEY querySkey,
-                            int32_t nMaxResult);
-
+int32_t tsdbGetTSmaData(STsdb *pTsdb, STSmaDataWrapper *pData, int64_t indexUid, int64_t interval, int8_t intervalUnit,
+                        tb_uid_t tableUid, col_id_t colId, TSKEY querySKey, int32_t nMaxResult);
 
 // STsdbCfg
 int  tsdbOptionsInit(STsdbCfg *);

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

@@ -161,6 +161,7 @@ struct STQ {
   STqMemRef     tqMemRef;
   STqMetaStore* tqMeta;
   STqPushMgr*   tqPushMgr;
+  SHashObj*     pStreamTasks;
   SWal*         pWal;
   SMeta*        pVnodeMeta;
 };

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

@@ -26,6 +26,7 @@ typedef struct SDBFile SDBFile;
 typedef DB_ENV*        TDBEnv;
 
 struct SDBFile {
+  int32_t fid;
   DB*     pDB;
   char*   path;
 };

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

@@ -636,7 +636,7 @@ STSma *metaGetSmaInfoByIndex(SMeta *pMeta, int64_t indexUid) {
   }
 
   // Decode
-  pCfg = (STSma *)malloc(sizeof(STSma));
+  pCfg = (STSma *)calloc(1, sizeof(STSma));
   if (pCfg == NULL) {
     return NULL;
   }
@@ -884,7 +884,7 @@ const char *metaSmaCursorNext(SMSmaCursor *pCur) {
 STSmaWrapper *metaGetSmaInfoByTable(SMeta *pMeta, tb_uid_t uid) {
   STSmaWrapper *pSW = NULL;
 
-  pSW = calloc(sizeof(*pSW), 1);
+  pSW = calloc(1, sizeof(*pSW));
   if (pSW == NULL) {
     return NULL;
   }

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

@@ -55,6 +55,8 @@ STQ* tqOpen(const char* path, SWal* pWal, SMeta* pVnodeMeta, STqCfg* tqConfig, S
     return NULL;
   }
 
+  pTq->pStreamTasks = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
+
   return pTq;
 }
 
@@ -416,3 +418,18 @@ int32_t tqProcessSetConnReq(STQ* pTq, char* msg) {
   terrno = TSDB_CODE_SUCCESS;
   return 0;
 }
+
+int32_t tqProcessTaskDeploy(STQ* pTq, char* msg, int32_t msgLen) {
+  SStreamTask* pTask = malloc(sizeof(SStreamTask));
+  if (pTask == NULL) {
+    return -1;
+  }
+  SCoder decoder;
+  tCoderInit(&decoder, TD_LITTLE_ENDIAN, (uint8_t*)msg, msgLen, TD_DECODER);
+  tDecodeSStreamTask(&decoder, pTask);
+  tCoderClear(&decoder);
+
+  taosHashPut(pTq->pStreamTasks, &pTask->taskId, sizeof(int32_t), pTask, sizeof(SStreamTask));
+
+  return 0;
+}

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

@@ -3386,7 +3386,7 @@ void filterPrepare(void* expr, void* param) {
     if (size < (uint32_t)pSchema->bytes) {
       size = pSchema->bytes;
     }
-    // to make sure tonchar does not cause invalid write, since the '\0' needs at least sizeof(wchar_t) space.
+    // to make sure tonchar does not cause invalid write, since the '\0' needs at least sizeof(TdUcs4) space.
     pInfo->q = calloc(1, size + TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE);
     tVariantDump(pCond, pInfo->q, pSchema->type, true);
   }

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

@@ -132,6 +132,11 @@ int  vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
       if (tqProcessRebReq(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead))) < 0) {
       }
     } break;
+    case TDMT_VND_TASK_DEPLOY: {
+      if (tqProcessTaskDeploy(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
+                              pMsg->contLen - sizeof(SMsgHead)) < 0) {
+      }
+    } break;
     case TDMT_VND_CREATE_SMA: {  // timeRangeSMA
       SSmaCfg vCreateSmaReq = {0};
       if (tDeserializeSVCreateTSmaReq(POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), &vCreateSmaReq) == NULL) {
@@ -139,6 +144,9 @@ int  vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
         return -1;
       }
 
+      // record current timezone of server side
+      tstrncpy(vCreateSmaReq.tSma.timezone, tsTimezone, TD_TIMEZONE_LEN);
+
       if (metaCreateTSma(pVnode->pMeta, &vCreateSmaReq) < 0) {
         // TODO: handle error
         tdDestroyTSma(&vCreateSmaReq.tSma);

source/dnode/vnode/test/tsdbSmaTest.cpp

@@ -49,7 +49,7 @@ TEST(testCase, tSma_Meta_Encode_Decode_Test) {
   STSmaWrapper tSmaWrapper = {.number = 1, .tSma = &tSma};
   uint32_t     bufLen = tEncodeTSmaWrapper(NULL, &tSmaWrapper);
 
-  void *buf = calloc(bufLen, 1);
+  void *buf = calloc(1, bufLen);
   ASSERT_NE(buf, nullptr);
 
   STSmaWrapper *pSW = (STSmaWrapper *)buf;
@@ -84,6 +84,7 @@ TEST(testCase, tSma_Meta_Encode_Decode_Test) {
   }
 
   // resource release
+  tfree(pSW);
   tdDestroyTSma(&tSma);
   tdDestroyTSmaWrapper(&dstTSmaWrapper);
 }
@@ -113,7 +114,7 @@ TEST(testCase, tSma_metaDB_Put_Get_Del_Test) {
   tSma.tableUid = tbUid;
 
   tSma.exprLen = strlen(expr);
-  tSma.expr = (char *)calloc(tSma.exprLen + 1, 1);
+  tSma.expr = (char *)calloc(1, tSma.exprLen + 1);
   ASSERT_NE(tSma.expr, nullptr);
   tstrncpy(tSma.expr, expr, tSma.exprLen + 1);
 
@@ -251,12 +252,12 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
   tSma.tableUid = tbUid;
 
   tSma.exprLen = strlen(expr);
-  tSma.expr = (char *)calloc(tSma.exprLen + 1, 1);
+  tSma.expr = (char *)calloc(1, tSma.exprLen + 1);
   ASSERT_NE(tSma.expr, nullptr);
   tstrncpy(tSma.expr, expr, tSma.exprLen + 1);
 
   tSma.tagsFilterLen = strlen(tagsFilter);
-  tSma.tagsFilter = (char *)calloc(tSma.tagsFilterLen + 1, 1);
+  tSma.tagsFilter = (char *)calloc(1, tSma.tagsFilterLen + 1);
   ASSERT_NE(tSma.tagsFilter, nullptr);
   tstrncpy(tSma.tagsFilter, tagsFilter, tSma.tagsFilterLen + 1);
 
@@ -273,20 +274,20 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
 
   // step 2: insert data
   STSmaDataWrapper *pSmaData = NULL;
-  STsdb             tsdb = {0};
-  STsdbCfg *        pCfg = &tsdb.config;
-
-  tsdb.pMeta = pMeta;
-  tsdb.vgId = 2;
-  tsdb.config.daysPerFile = 10;  // default days is 10
-  tsdb.config.keep1 = 30;
-  tsdb.config.keep2 = 90;
-  tsdb.config.keep = 365;
-  tsdb.config.precision = TSDB_TIME_PRECISION_MILLI;
-  tsdb.config.update = TD_ROW_OVERWRITE_UPDATE;
-  tsdb.config.compression = TWO_STAGE_COMP;
-
-  switch (tsdb.config.precision) {
+  STsdb *           pTsdb = (STsdb *)calloc(1, sizeof(STsdb));
+  STsdbCfg *        pCfg = &pTsdb->config;
+
+  pTsdb->pMeta = pMeta;
+  pTsdb->vgId = 2;
+  pTsdb->config.daysPerFile = 10;  // default days is 10
+  pTsdb->config.keep1 = 30;
+  pTsdb->config.keep2 = 90;
+  pTsdb->config.keep = 365;
+  pTsdb->config.precision = TSDB_TIME_PRECISION_MILLI;
+  pTsdb->config.update = TD_ROW_OVERWRITE_UPDATE;
+  pTsdb->config.compression = TWO_STAGE_COMP;
+
+  switch (pTsdb->config.precision) {
     case TSDB_TIME_PRECISION_MILLI:
       skey1 *= 1e3;
       break;
@@ -301,9 +302,15 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
       break;
   }
 
-  char *msg = (char *)calloc(100, 1);
-  assert(msg != NULL);
-  ASSERT_EQ(tsdbUpdateSmaWindow(&tsdb, TSDB_SMA_TYPE_TIME_RANGE, msg), 0);
+  SDiskCfg pDisks = {.level = 0, .primary = 1};
+  strncpy(pDisks.dir, "/var/lib/taos", TSDB_FILENAME_LEN);
+  int32_t numOfDisks = 1;
+  pTsdb->pTfs = tfsOpen(&pDisks, numOfDisks);
+  ASSERT_NE(pTsdb->pTfs, nullptr);
+
+  char *msg = (char *)calloc(1, 100);
+  ASSERT_NE(msg, nullptr);
+  ASSERT_EQ(tsdbUpdateSmaWindow(pTsdb, TSDB_SMA_TYPE_TIME_RANGE, msg), 0);
 
   // init
   int32_t allocCnt = 0;
@@ -361,13 +368,13 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
   ASSERT_GE(bufSize, pSmaData->dataLen);
 
   // execute
-  ASSERT_EQ(tsdbInsertTSmaData(&tsdb, (char *)pSmaData), TSDB_CODE_SUCCESS);
+  ASSERT_EQ(tsdbInsertTSmaData(pTsdb, (char *)pSmaData), TSDB_CODE_SUCCESS);
 
   // step 3: query
   uint32_t checkDataCnt = 0;
   for (int32_t t = 0; t < numOfTables; ++t) {
     for (col_id_t c = 0; c < numOfCols; ++c) {
-      ASSERT_EQ(tsdbGetTSmaData(&tsdb, NULL, indexUid1, interval1, intervalUnit1, tbUid + t,
+      ASSERT_EQ(tsdbGetTSmaData(pTsdb, NULL, indexUid1, interval1, intervalUnit1, tbUid + t,
                                 c + PRIMARYKEY_TIMESTAMP_COL_ID, skey1, 1),
                 TSDB_CODE_SUCCESS);
       ++checkDataCnt;
@@ -377,9 +384,12 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
   printf("%s:%d The sma data check count for insert and query is %" PRIu32 "\n", __FILE__, __LINE__, checkDataCnt);
 
   // release data
+  tfree(msg);
   taosTZfree(buf);
   // release meta
   tdDestroyTSma(&tSma);
+  tfsClose(pTsdb->pTfs);
+  tsdbClose(pTsdb);
   metaClose(pMeta);
 }
 #endif

source/libs/catalog/src/catalog.c

@@ -2349,6 +2349,9 @@ _return:
   CTG_API_LEAVE(code);
 }
 
+int32_t catalogUpdateVgEpSet(SCatalog* pCtg, const char* dbFName, int32_t vgId, SEpSet *epSet) {
+
+}
 
 int32_t catalogRemoveDB(SCatalog* pCtg, const char* dbFName, uint64_t dbId) {
   CTG_API_ENTER();
@@ -2394,6 +2397,9 @@ _return:
   CTG_API_LEAVE(code);
 }
 
+int32_t catalogGetIndexMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, const char *pIndexName, SIndexMeta** pIndexMeta) {
+
+}
 
 int32_t catalogGetTableMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta) {
   CTG_API_ENTER();
@@ -2663,11 +2669,14 @@ _return:
 int32_t catalogGetQnodeList(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SArray* pQnodeList) {
   CTG_API_ENTER();
   
+  int32_t code = 0;
   if (NULL == pCtg || NULL == pRpc  || NULL == pMgmtEps || NULL == pQnodeList) {
     CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
   }
 
-  //TODO
+  CTG_ERR_JRET(ctgGetQnodeListFromMnode(pCtg, pRpc, pMgmtEps, &pQnodeList));
+
+_return:
 
   CTG_API_LEAVE(TSDB_CODE_SUCCESS);
 }

source/libs/parser/src/parInsert.c

@@ -622,7 +622,7 @@ static FORCE_INLINE int32_t MemRowAppend(const void* value, int32_t len, void* p
     // if the converted output len is over than pColumnModel->bytes, return error: 'Argument list too long'
     int32_t     output = 0;
     const char* rowEnd = tdRowEnd(rb->pBuf);
-    if (!taosMbsToUcs4(value, len, (char*)varDataVal(rowEnd), pa->schema->bytes - VARSTR_HEADER_SIZE, &output)) {
+    if (!taosMbsToUcs4(value, len, (TdUcs4*)varDataVal(rowEnd), pa->schema->bytes - VARSTR_HEADER_SIZE, &output)) {
       return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
     }
     varDataSetLen(rowEnd, output);
@@ -725,7 +725,7 @@ static int32_t KvRowAppend(const void *value, int32_t len, void *param) {
   } else if (TSDB_DATA_TYPE_NCHAR == type) {
     // if the converted output len is over than pColumnModel->bytes, return error: 'Argument list too long'
     int32_t output = 0;
-    if (!taosMbsToUcs4(value, len, varDataVal(pa->buf), pa->schema->bytes - VARSTR_HEADER_SIZE, &output)) {
+    if (!taosMbsToUcs4(value, len, (TdUcs4*)varDataVal(pa->buf), pa->schema->bytes - VARSTR_HEADER_SIZE, &output)) {
       return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
     }
 

source/libs/parser/test/mockCatalogService.cpp

@@ -13,12 +13,11 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "mockCatalogService.h"
-
 #include <iomanip>
 #include <iostream>
 #include <map>
 #include "tdatablock.h"
+#include "mockCatalogService.h"
 
 #include "tname.h"
 #include "ttypes.h"

source/libs/planner/src/planLogicCreater.c

@@ -310,7 +310,8 @@ static SLogicNode* createWindowLogicNodeByInterval(SLogicPlanContext* pCxt, SInt
   pWindow->node.id = pCxt->planNodeId++;
 
   pWindow->winType = WINDOW_TYPE_INTERVAL;
-  SValueNode* pIntervalNode = (SValueNode*)((SRawExprNode*)(pInterval->pInterval))->pNode;
+  //SValueNode* pIntervalNode = (SValueNode*)((SRawExprNode*)(pInterval->pInterval))->pNode;
+  SValueNode* pIntervalNode = (SValueNode*)(pInterval->pInterval);
 
   pWindow->interval = pIntervalNode->datum.i;
   pWindow->offset = (NULL != pInterval->pOffset ? ((SValueNode*)pInterval->pOffset)->datum.i : 0);

source/libs/planner/src/planSpliter.c

@@ -89,7 +89,7 @@ static SSubLogicPlan* stsCreateScanSubplan(SSplitContext* pCxt, SScanLogicNode*
   pSubplan->id.groupId = pCxt->groupId;
   pSubplan->subplanType = SUBPLAN_TYPE_SCAN;
   pSubplan->pNode = (SLogicNode*)nodesCloneNode(pScan);
-  TSWAP(pSubplan->pVgroupList, ((SScanLogicNode*)pSubplan->pNode)->pVgroupList, SVgroupsInfo);
+  TSWAP(pSubplan->pVgroupList, ((SScanLogicNode*)pSubplan->pNode)->pVgroupList, SVgroupsInfo*);
   SPLIT_FLAG_SET_MASK(pSubplan->splitFlag, SPLIT_FLAG_STS);
   return pSubplan;
 }

source/libs/scalar/src/filter.c

@@ -1813,7 +1813,7 @@ int32_t fltInitValFieldData(SFilterInfo *info) {
     if(type == TSDB_DATA_TYPE_NCHAR &&
         (unit->compare.optr == OP_TYPE_MATCH || unit->compare.optr == OP_TYPE_NMATCH)){
       char newValData[TSDB_REGEX_STRING_DEFAULT_LEN * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE] = {0};
-      int32_t len = taosUcs4ToMbs(varDataVal(fi->data), varDataLen(fi->data), varDataVal(newValData));
+      int32_t len = taosUcs4ToMbs((TdUcs4*)varDataVal(fi->data), varDataLen(fi->data), varDataVal(newValData));
       if (len < 0){
         qError("filterInitValFieldData taosUcs4ToMbs error 1");
         return TSDB_CODE_QRY_APP_ERROR;
@@ -2992,7 +2992,7 @@ bool filterExecuteImplMisc(void *pinfo, int32_t numOfRows, int8_t** p, SColumnDa
 
     if(info->cunits[uidx].dataType == TSDB_DATA_TYPE_NCHAR && (info->cunits[uidx].optr == OP_TYPE_MATCH || info->cunits[uidx].optr == OP_TYPE_NMATCH)){
       char *newColData = calloc(info->cunits[uidx].dataSize * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE, 1);
-      int32_t len = taosUcs4ToMbs(varDataVal(colData), varDataLen(colData), varDataVal(newColData));
+      int32_t len = taosUcs4ToMbs((TdUcs4*)varDataVal(colData), varDataLen(colData), varDataVal(newColData));
       if (len < 0){
         qError("castConvert1 taosUcs4ToMbs error");
       }else{
@@ -3052,7 +3052,7 @@ bool filterExecuteImpl(void *pinfo, int32_t numOfRows, int8_t** p, SColumnDataAg
             } else {
               if(cunit->dataType == TSDB_DATA_TYPE_NCHAR && (cunit->optr == OP_TYPE_MATCH || cunit->optr == OP_TYPE_NMATCH)){
                 char *newColData = calloc(cunit->dataSize * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE, 1);
-                int32_t len = taosUcs4ToMbs(varDataVal(colData), varDataLen(colData), varDataVal(newColData));
+                int32_t len = taosUcs4ToMbs((TdUcs4*)varDataVal(colData), varDataLen(colData), varDataVal(newColData));
                 if (len < 0){
                   qError("castConvert1 taosUcs4ToMbs error");
                 }else{
@@ -3433,7 +3433,7 @@ int32_t filterConverNcharColumns(SFilterInfo* info, int32_t rows, bool *gotNchar
           varDataCopy(dst, src);
           continue;
         }
-        bool ret = taosMbsToUcs4(varDataVal(src), varDataLen(src), varDataVal(dst), bufSize, &len);
+        bool ret = taosMbsToUcs4(varDataVal(src), varDataLen(src), (TdUcs4*)varDataVal(dst), bufSize, &len);
         if(!ret) {
           qError("filterConverNcharColumns taosMbsToUcs4 error");
           return TSDB_CODE_FAILED;

source/libs/scalar/src/sclvector.c

@@ -316,7 +316,7 @@ int32_t vectorConvertFromVarData(SScalarParam* pIn, SScalarParam* pOut, int32_t
         tmp = realloc(tmp, bufSize);
       }
       
-      int len = taosUcs4ToMbs(varDataVal(pIn->data), varDataLen(pIn->data), tmp);
+      int len = taosUcs4ToMbs((TdUcs4*)varDataVal(pIn->data), varDataLen(pIn->data), tmp);
       if (len < 0){
         sclError("castConvert taosUcs4ToMbs error 1");
         tfree(tmp);

source/libs/sync/src/syncUtil.c

@@ -14,15 +14,12 @@
  */
 
 #include "syncUtil.h"
-#include <arpa/inet.h>
-#include <netinet/in.h>
-#include <sys/socket.h>
 #include "syncEnv.h"
 
 // ---- encode / decode
 uint64_t syncUtilAddr2U64(const char* host, uint16_t port) {
   uint64_t u64;
-  uint32_t hostU32 = (uint32_t)inet_addr(host);
+  uint32_t hostU32 = (uint32_t)taosInetAddr(host);
   // assert(hostU32 != (uint32_t)-1);
   u64 = (((uint64_t)hostU32) << 32) | (((uint32_t)port) << 16);
   return u64;
@@ -33,7 +30,7 @@ void syncUtilU642Addr(uint64_t u64, char* host, size_t len, uint16_t* port) {
 
   struct in_addr addr;
   addr.s_addr = hostU32;
-  snprintf(host, len, "%s", inet_ntoa(addr));
+  snprintf(host, len, "%s", taosInetNtoa(addr));
   *port = (uint16_t)((u64 & 0x00000000FFFF0000) >> 16);
 }
 

source/libs/tdb/src/db/tdbBtree.c

@@ -27,7 +27,8 @@ typedef struct {
 } SBtIdx;
 
 // Btree page header definition
-typedef struct __attribute__((__packed__)) {
+#pragma pack (push,1) 
+typedef struct {
   uint8_t  flag;        // page flag
   int32_t  vlen;        // value length of current page, TDB_VARIANT_LEN for variant length
   uint16_t nPayloads;   // number of total payloads
@@ -36,6 +37,7 @@ typedef struct __attribute__((__packed__)) {
   pgoff_t  offPayload;  // payload offset
   pgno_t   rChildPgno;  // right most child page number
 } SBtPgHdr;
+#pragma pack(pop)
 
 typedef int (*BtreeCmprFn)(const void *, const void *);
 

source/libs/tdb/src/db/tdbEnv.c

@@ -30,11 +30,7 @@ struct STDbEnv {
   } pgfht;  // page file hash table;
 };
 
-#define TDB_ENV_PGF_HASH(fileid)        \
-  ({                                    \
-    uint8_t *tmp = (uint8_t *)(fileid); \
-    tmp[0] + tmp[1] + tmp[2];           \
-  })
+#define TDB_ENV_PGF_HASH(fileid) (((uint8_t *)(fileid))[0] + ((uint8_t *)(fileid))[1] + ((uint8_t *)(fileid))[2])
 
 static int tdbEnvDestroy(TENV *pEnv);
 

source/libs/tdb/src/db/tdbPgCache.c

@@ -106,11 +106,7 @@ int pgCacheClose(SPgCache *pPgCache) {
   return 0;
 }
 
-#define PG_CACHE_HASH(fileid, pgno)       \
-  ({                                      \
-    uint64_t *tmp = (uint64_t *)(fileid); \
-    (tmp[0] + tmp[1] + tmp[2] + (pgno));  \
-  })
+#define PG_CACHE_HASH(fileid, pgno) (((uint64_t *)(fileid))[0] + ((uint64_t *)(fileid))[1] + ((uint64_t *)(fileid))[2] + (pgno))
 
 SPage *pgCacheFetch(SPgCache *pPgCache, pgid_t pgid) {
   SPage *  pPage;

source/libs/tdb/src/inc/tdbPgFile.h

@@ -20,13 +20,15 @@
 extern "C" {
 #endif
 
-typedef struct __attribute__((__packed__)) {
+#pragma pack (push,1) 
+typedef struct {
   char    hdrInfo[16];  // info string
   pgsz_t  szPage;       // page size of current file
   int32_t cno;          // commit number counter
   pgno_t  freePgno;     // freelist page number
   uint8_t resv[100];    // reserved space
 } SPgFileHdr;
+#pragma pack(pop)
 
 #define TDB_PG_FILE_HDR_SIZE 128
 

source/libs/tfs/src/tfs.c

@@ -202,6 +202,12 @@ void tfsDirname(const STfsFile *pFile, char *dest) {
   tstrncpy(dest, taosDirName(tname), TSDB_FILENAME_LEN);
 }
 
+void tfsAbsoluteName(STfs *pTfs, SDiskID diskId, const char *rname, char *aname) {
+  STfsDisk *pDisk = TFS_DISK_AT(pTfs, diskId);
+  
+  snprintf(aname, TSDB_FILENAME_LEN, "%s%s%s", pDisk->path, TD_DIRSEP, rname);
+}
+
 int32_t tfsRemoveFile(const STfsFile *pFile) { return taosRemoveFile(pFile->aname); }
 
 int32_t tfsCopyFile(const STfsFile *pFile1, const STfsFile *pFile2) {

source/libs/transport/src/transCli.c

@@ -30,6 +30,7 @@ typedef struct SCliConn {
   void*        hostThrd;
   SConnBuffer  readBuf;
   void*        data;
+  SArray*      cliMsgs;
   queue        conn;
   uint64_t     expireTime;
   int          hThrdIdx;
@@ -106,6 +107,7 @@ static void cliAsyncCb(uv_async_t* handle);
 static SCliConn* cliCreateConn(SCliThrdObj* thrd);
 static void      cliDestroyConn(SCliConn* pConn, bool clear /*clear tcp handle or not*/);
 static void      cliDestroy(uv_handle_t* handle);
+static void      cliSend(SCliConn* pConn);
 
 // process data read from server, add decompress etc later
 static void cliHandleResp(SCliConn* conn);
@@ -156,8 +158,18 @@ static void         destroyThrdObj(SCliThrdObj* pThrd);
   } while (0)
 #define CONN_NO_PERSIST_BY_APP(conn) ((conn)->persist == false)
 
+#define REQUEST_NO_RESP(msg) ((msg)->noResp == 1)
+
 static void* cliWorkThread(void* arg);
 
+bool cliMaySendCachedMsg(SCliConn* conn) {
+  if (taosArrayGetSize(conn->cliMsgs) > 0) {
+    cliSend(conn);
+    return true;
+  } else {
+    return false;
+  }
+}
 void cliHandleResp(SCliConn* conn) {
   SCliThrdObj* pThrd = conn->hostThrd;
   STrans*      pTransInst = pThrd->pTransInst;
@@ -173,18 +185,18 @@ void cliHandleResp(SCliConn* conn) {
   transMsg.msgType = pHead->msgType;
   transMsg.ahandle = NULL;
 
-  SCliMsg*       pMsg = conn->data;
+  SCliMsg* pMsg = NULL;
+  if (taosArrayGetSize(conn->cliMsgs) > 0) {
+    pMsg = taosArrayGetP(conn->cliMsgs, 0);
+    taosArrayRemove(conn->cliMsgs, 0);
+  }
+
   STransConnCtx* pCtx = pMsg ? pMsg->ctx : NULL;
   if (pMsg == NULL && !CONN_NO_PERSIST_BY_APP(conn)) {
     transMsg.ahandle = pTransInst->mfp ? (*pTransInst->mfp)(pTransInst->parent, transMsg.msgType) : NULL;
   } else {
     transMsg.ahandle = pCtx ? pCtx->ahandle : NULL;
   }
-  // if (rpcMsg.ahandle == NULL) {
-  //  tDebug("%s cli conn %p handle except", CONN_GET_INST_LABEL(conn), conn);
-  //  return;
-  //}
-
   // buf's mem alread translated to transMsg.pCont
   transClearBuffer(&conn->readBuf);
 
@@ -195,8 +207,8 @@ void cliHandleResp(SCliConn* conn) {
   }
 
   tDebug("%s cli conn %p %s received from %s:%d, local info: %s:%d, msg size: %d", pTransInst->label, conn,
-         TMSG_INFO(pHead->msgType), inet_ntoa(conn->addr.sin_addr), ntohs(conn->addr.sin_port),
-         inet_ntoa(conn->locaddr.sin_addr), ntohs(conn->locaddr.sin_port), transMsg.contLen);
+         TMSG_INFO(pHead->msgType), taosInetNtoa(conn->addr.sin_addr), ntohs(conn->addr.sin_port),
+         taosInetNtoa(conn->locaddr.sin_addr), ntohs(conn->locaddr.sin_port), transMsg.contLen);
 
   conn->secured = pHead->secured;
 
@@ -214,12 +226,15 @@ void cliHandleResp(SCliConn* conn) {
     memcpy((char*)pCtx->pRsp, (char*)&transMsg, sizeof(transMsg));
     tsem_post(pCtx->pSem);
   }
+  destroyCmsg(pMsg);
+
+  if (cliMaySendCachedMsg(conn) == true) {
+    return;
+  }
 
   if (CONN_NO_PERSIST_BY_APP(conn)) {
     addConnToPool(pThrd->pool, conn);
   }
-  destroyCmsg(conn->data);
-  conn->data = NULL;
 
   uv_read_start((uv_stream_t*)conn->stream, cliAllocRecvBufferCb, cliRecvCb);
   // start thread's timer of conn pool if not active
@@ -229,7 +244,7 @@ void cliHandleResp(SCliConn* conn) {
 }
 
 void cliHandleExcept(SCliConn* pConn) {
-  if (pConn->data == NULL) {
+  if (taosArrayGetSize(pConn->cliMsgs) == 0) {
     if (pConn->broken == true || CONN_NO_PERSIST_BY_APP(pConn)) {
       transUnrefCliHandle(pConn);
       return;
@@ -238,7 +253,13 @@ void cliHandleExcept(SCliConn* pConn) {
   SCliThrdObj* pThrd = pConn->hostThrd;
   STrans*      pTransInst = pThrd->pTransInst;
 
-  SCliMsg*       pMsg = pConn->data;
+  do {
+    SCliMsg* pMsg = NULL;
+    if (taosArrayGetSize(pConn->cliMsgs) > 0) {
+      pMsg = taosArrayGetP(pConn->cliMsgs, 0);
+      taosArrayRemove(pConn->cliMsgs, 0);
+    }
+
     STransConnCtx* pCtx = pMsg ? pMsg->ctx : NULL;
 
     STransMsg transMsg = {0};
@@ -260,10 +281,10 @@ void cliHandleExcept(SCliConn* pConn) {
       memcpy((char*)(pCtx->pRsp), (char*)(&transMsg), sizeof(transMsg));
       tsem_post(pCtx->pSem);
     }
-  destroyCmsg(pConn->data);
-  pConn->data = NULL;
-
+    destroyCmsg(pMsg);
     tTrace("%s cli conn %p start to destroy", CONN_GET_INST_LABEL(pConn), pConn);
+  } while (taosArrayGetSize(pConn->cliMsgs) > 0);
+
   transUnrefCliHandle(pConn);
 }
 
@@ -398,6 +419,7 @@ static SCliConn* cliCreateConn(SCliThrdObj* pThrd) {
 
   conn->writeReq.data = conn;
   conn->connReq.data = conn;
+  conn->cliMsgs = taosArrayInit(2, sizeof(void*));
 
   QUEUE_INIT(&conn->conn);
   conn->hostThrd = pThrd;
@@ -417,32 +439,51 @@ static void cliDestroy(uv_handle_t* handle) {
   SCliConn* conn = handle->data;
   free(conn->ip);
   free(conn->stream);
+  taosArrayDestroy(conn->cliMsgs);
   tTrace("%s cli conn %p destroy successfully", CONN_GET_INST_LABEL(conn), conn);
   free(conn);
 }
-
+static bool cliHandleNoResp(SCliConn* conn) {
+  bool    res = false;
+  SArray* msgs = conn->cliMsgs;
+  if (taosArrayGetSize(msgs) > 0) {
+    SCliMsg* pMsg = taosArrayGetP(msgs, 0);
+    if (REQUEST_NO_RESP(&pMsg->msg)) {
+      taosArrayRemove(msgs, 0);
+      destroyCmsg(pMsg);
+      res = true;
+    }
+    if (res == true) {
+      if (cliMaySendCachedMsg(conn) == false) {
+        SCliThrdObj* thrd = conn->hostThrd;
+        addConnToPool(thrd->pool, conn);
+      }
+    }
+  }
+  return res;
+}
 static void cliSendCb(uv_write_t* req, int status) {
   SCliConn* pConn = req->data;
 
   if (status == 0) {
     tTrace("%s cli conn %p data already was written out", CONN_GET_INST_LABEL(pConn), pConn);
-    SCliMsg* pMsg = pConn->data;
-    if (pMsg == NULL) {
-      return;
-    }
-    destroyUserdata(&pMsg->msg);
   } else {
     tError("%s cli conn %p failed to write: %s", CONN_GET_INST_LABEL(pConn), pConn, uv_err_name(status));
     cliHandleExcept(pConn);
     return;
   }
+  if (cliHandleNoResp(pConn) == true) {
+    tTrace("%s cli conn %p no resp required", CONN_GET_INST_LABEL(pConn), pConn);
+    return;
+  }
   uv_read_start((uv_stream_t*)pConn->stream, cliAllocRecvBufferCb, cliRecvCb);
 }
 
 void cliSend(SCliConn* pConn) {
   CONN_HANDLE_BROKEN(pConn);
 
-  SCliMsg*       pCliMsg = pConn->data;
+  assert(taosArrayGetSize(pConn->cliMsgs) > 0);
+  SCliMsg*       pCliMsg = taosArrayGetP(pConn->cliMsgs, 0);
   STransConnCtx* pCtx = pCliMsg->ctx;
 
   SCliThrdObj* pThrd = pConn->hostThrd;
@@ -472,14 +513,16 @@ void cliSend(SCliConn* pConn) {
     msgLen += sizeof(STransUserMsg);
   }
 
+  pHead->resflag = REQUEST_NO_RESP(pMsg) ? 1 : 0;
   pHead->msgType = pMsg->msgType;
   pHead->msgLen = (int32_t)htonl((uint32_t)msgLen);
 
   uv_buf_t wb = uv_buf_init((char*)pHead, msgLen);
   tDebug("%s cli conn %p %s is send to %s:%d, local info %s:%d", CONN_GET_INST_LABEL(pConn), pConn,
-         TMSG_INFO(pHead->msgType), inet_ntoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port),
-         inet_ntoa(pConn->locaddr.sin_addr), ntohs(pConn->locaddr.sin_port));
+         TMSG_INFO(pHead->msgType), taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port),
+         taosInetNtoa(pConn->locaddr.sin_addr), ntohs(pConn->locaddr.sin_port));
 
+  pConn->writeReq.data = pConn;
   uv_write(&pConn->writeReq, (uv_stream_t*)pConn->stream, &wb, 1, cliSendCb);
 
   return;
@@ -502,8 +545,8 @@ void cliConnCb(uv_connect_t* req, int status) {
   uv_tcp_getsockname((uv_tcp_t*)pConn->stream, (struct sockaddr*)&pConn->locaddr, &addrlen);
 
   tTrace("%s cli conn %p connect to server successfully", CONN_GET_INST_LABEL(pConn), pConn);
-
   assert(pConn->stream == req->handle);
+
   cliSend(pConn);
 }
 
@@ -521,8 +564,11 @@ static void cliHandleRelease(SCliMsg* pMsg, SCliThrdObj* pThrd) {
   SCliConn* conn = pMsg->msg.handle;
   tDebug("%s cli conn %p release to inst", CONN_GET_INST_LABEL(conn), conn);
 
+  while (taosArrayGetSize(conn->cliMsgs) > 0) {
+    SCliMsg* pMsg = taosArrayGetP(conn->cliMsgs, 0);
     destroyCmsg(pMsg);
-  conn->data = NULL;
+    taosArrayRemove(conn->cliMsgs, 0);
+  }
 
   transDestroyBuffer(&conn->readBuf);
   if (conn->persist && T_REF_VAL_GET(conn) >= 2) {
@@ -561,14 +607,19 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
 
   SCliConn* conn = cliGetConn(pMsg, pThrd);
   if (conn != NULL) {
-    conn->data = pMsg;
     conn->hThrdIdx = pCtx->hThrdIdx;
 
+    if (taosArrayGetSize(conn->cliMsgs) > 0) {
+      taosArrayPush(conn->cliMsgs, &pMsg);
+      return;
+    }
+    taosArrayPush(conn->cliMsgs, &pMsg);
     transDestroyBuffer(&conn->readBuf);
     cliSend(conn);
   } else {
     conn = cliCreateConn(pThrd);
-    conn->data = pMsg;
+    taosArrayPush(conn->cliMsgs, &pMsg);
+
     conn->hThrdIdx = pCtx->hThrdIdx;
     conn->ip = strdup(pMsg->ctx->ip);
     conn->port = pMsg->ctx->port;

source/libs/transport/src/transSrv.c

@@ -202,7 +202,7 @@ static void uvHandleReq(SSrvConn* pConn) {
 
   STransMsgHead* pHead = (STransMsgHead*)p->msg;
   if (pHead->secured == 1) {
-    STransUserMsg* uMsg = (p->msg + p->msgLen - sizeof(STransUserMsg));
+    STransUserMsg* uMsg = (STransUserMsg*)((char*)p->msg + p->msgLen - sizeof(STransUserMsg));
     memcpy(pConn->user, uMsg->user, tListLen(uMsg->user));
     memcpy(pConn->secret, uMsg->secret, tListLen(uMsg->secret));
   }
@@ -226,15 +226,22 @@ static void uvHandleReq(SSrvConn* pConn) {
   transMsg.msgType = pHead->msgType;
   transMsg.code = pHead->code;
   transMsg.ahandle = NULL;
-  transMsg.handle = pConn;
+  transMsg.handle = NULL;
 
   transClearBuffer(&pConn->readBuf);
   pConn->inType = pHead->msgType;
-  transRefSrvHandle(pConn);
 
+  if (pHead->resflag == 0) {
+    transRefSrvHandle(pConn);
+    transMsg.handle = pConn;
     tDebug("server conn %p %s received from %s:%d, local info: %s:%d, msg size: %d", pConn, TMSG_INFO(transMsg.msgType),
-         inet_ntoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port), inet_ntoa(pConn->locaddr.sin_addr),
+           taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port), taosInetNtoa(pConn->locaddr.sin_addr),
            ntohs(pConn->locaddr.sin_port), transMsg.contLen);
+  } else {
+    tDebug("server conn %p %s received from %s:%d, local info: %s:%d, msg size: %d, no resp ", pConn,
+           TMSG_INFO(transMsg.msgType), taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port),
+           taosInetNtoa(pConn->locaddr.sin_addr), ntohs(pConn->locaddr.sin_port), transMsg.contLen);
+  }
 
   STrans* pTransInst = (STrans*)p->shandle;
   (*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL);
@@ -340,7 +347,7 @@ static void uvPrepareSendData(SSrvMsg* smsg, uv_buf_t* wb) {
     // impl later
   }
   tDebug("server conn %p %s is sent to %s:%d, local info: %s:%d", pConn, TMSG_INFO(pHead->msgType),
-         inet_ntoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port), inet_ntoa(pConn->locaddr.sin_addr),
+         taosInetNtoa(pConn->addr.sin_addr), ntohs(pConn->addr.sin_port), taosInetNtoa(pConn->locaddr.sin_addr),
          ntohs(pConn->locaddr.sin_port));
 
   pHead->msgLen = htonl(len);
@@ -367,8 +374,8 @@ static void uvStartSendResp(SSrvMsg* smsg) {
   transUnrefSrvHandle(pConn);
 
   if (taosArrayGetSize(pConn->srvMsgs) > 0) {
-    tDebug("server conn %p push data to client %s:%d, local info: %s:%d", pConn, inet_ntoa(pConn->addr.sin_addr),
-           ntohs(pConn->addr.sin_port), inet_ntoa(pConn->locaddr.sin_addr), ntohs(pConn->locaddr.sin_port));
+    tDebug("server conn %p push data to client %s:%d, local info: %s:%d", pConn, taosInetNtoa(pConn->addr.sin_addr),
+           ntohs(pConn->addr.sin_port), taosInetNtoa(pConn->locaddr.sin_addr), ntohs(pConn->locaddr.sin_port));
     taosArrayPush(pConn->srvMsgs, &smsg);
     return;
   }

source/libs/transport/test/transUT.cc

@@ -50,6 +50,7 @@ static void *ConstructArgForSpecificMsgType(void *parent, tmsg_t msgType) {
 }
 // server except
 static bool handleExcept(void *parent, tmsg_t msgType) {
+  //
   return msgType == TDMT_VND_QUERY || msgType == TDMT_VND_FETCH_RSP || msgType == TDMT_VND_RES_READY_RSP;
 }
 typedef void (*CB)(void *parent, SRpcMsg *pMsg, SEpSet *pEpSet);
@@ -360,7 +361,7 @@ TEST_F(TransEnv, cliPersistHandle) {
   tr->SetCliPersistFp(cliPersistHandle);
   SRpcMsg resp = {0};
   for (int i = 0; i < 10; i++) {
-    SRpcMsg req = {.handle = resp.handle};
+    SRpcMsg req = {.handle = resp.handle, .noResp = 0};
     req.msgType = 1;
     req.pCont = rpcMallocCont(10);
     req.contLen = 10;
@@ -447,6 +448,25 @@ TEST_F(TransEnv, srvPersistHandleExcept) {
   // conn broken
   //
 }
+TEST_F(TransEnv, cliPersistHandleExcept) {
+  tr->SetSrvContinueSend(processContinueSend);
+  tr->SetCliPersistFp(cliPersistHandle);
+  SRpcMsg resp = {0};
+  for (int i = 0; i < 5; i++) {
+    SRpcMsg req = {.handle = resp.handle};
+    req.msgType = 1;
+    req.pCont = rpcMallocCont(10);
+    req.contLen = 10;
+    tr->cliSendAndRecv(&req, &resp);
+    if (i > 2) {
+      tr->StopSrv();
+      break;
+    }
+  }
+  taosMsleep(2000);
+  // conn broken
+  //
+}
 
 TEST_F(TransEnv, multiCliPersistHandleExcept) {
   // conn broken
@@ -457,5 +477,15 @@ TEST_F(TransEnv, queryExcept) {
   // query and conn is broken
 }
 TEST_F(TransEnv, noResp) {
+  SRpcMsg resp = {0};
+  for (int i = 0; i < 5; i++) {
+    SRpcMsg req = {.noResp = 1};
+    req.msgType = 1;
+    req.pCont = rpcMallocCont(10);
+    req.contLen = 10;
+    tr->cliSendAndRecv(&req, &resp);
+  }
+  taosMsleep(2000);
+
   // no resp
 }

source/libs/wal/src/walMeta.c

@@ -19,9 +19,6 @@
 #include "tref.h"
 #include "walInt.h"
 
-#include <libgen.h>
-#include <regex.h>
-
 int64_t inline walGetFirstVer(SWal* pWal) { return pWal->vers.firstVer; }
 
 int64_t inline walGetSnaphostVer(SWal* pWal) { return pWal->vers.snapshotVer; }

source/os/CMakeLists.txt

@@ -3,10 +3,16 @@ add_library(os STATIC ${OS_SRC})
 target_include_directories(
     os
     PUBLIC "${CMAKE_SOURCE_DIR}/include/os"
-    PRIVATE "${CMAKE_SOURCE_DIR}/include"
-    PRIVATE "${CMAKE_SOURCE_DIR}/include/util"
-    PRIVATE "${CMAKE_SOURCE_DIR}/contrib/pthread-win32"
+    PUBLIC "${CMAKE_SOURCE_DIR}/include"
+    PUBLIC "${CMAKE_SOURCE_DIR}/include/util"
+    PUBLIC "${CMAKE_SOURCE_DIR}/contrib/pthread"
+    PUBLIC "${CMAKE_SOURCE_DIR}/contrib/gnuregex"
 )
+# iconv
+find_path(IconvApiIncludes iconv.h PATHS)
+if(NOT IconvApiIncludes)
+    add_definitions(-DDISALLOW_NCHAR_WITHOUT_ICONV) 
+endif ()
 target_link_libraries(
     os pthread dl rt m
 )

source/os/src/osFile.c

@@ -46,6 +46,23 @@ extern int openU(const char *, int, ...); /* MsvcLibX UTF-8 version of open */
 #define O_TEXT                    LINUX_FILE_NO_TEXT_OPTION
 #endif
 
+#if defined(WINDOWS)
+typedef int32_t FileFd;
+typedef int32_t SocketFd;
+#else
+typedef int32_t FileFd;
+typedef int32_t SocketFd;
+#endif
+
+typedef int32_t FileFd;
+
+typedef struct TdFile {
+  pthread_rwlock_t rwlock;
+  int      refId;
+  FileFd   fd;
+  FILE    *fp;
+} * TdFilePtr, TdFile;
+
 #define FILE_WITH_LOCK 1
 
 void taosGetTmpfilePath(const char *inputTmpDir, const char *fileNamePrefix, char *dstPath) {

source/os/src/osLocale.c

@@ -81,7 +81,7 @@ void taosSetSystemLocale(const char *inLocale, const char *inCharSet) {
   }
 
   if (!taosValidateEncodec(inCharSet)) {
-    printf("Invalid charset:%s, please set the valid charset in config file", inCharSet);
+    printf("Invalid charset:%s, please set the valid charset in config file\n", inCharSet);
     exit(-1);
   }
 }

source/os/src/osString.c

@@ -13,22 +13,69 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#define ALLOW_FORBID_FUNC
 #define _DEFAULT_SOURCE
+#include <malloc.h>
 #include "os.h"
-#include "tdef.h"
-#include <wchar.h>
-#include <wctype.h>
 
+#ifndef DISALLOW_NCHAR_WITHOUT_ICONV
+#include "iconv.h"
+#endif
+
+extern int wcwidth(wchar_t c);
+extern int wcswidth(const wchar_t *s, size_t n);
 
 int64_t taosStr2int64(const char *str) {
   char *endptr = NULL;
   return strtoll(str, &endptr, 10);
 }
 
-#ifdef USE_LIBICONV
-#include "iconv.h"
+int32_t tasoUcs4Compare(TdUcs4 *f1_ucs4, TdUcs4 *f2_ucs4, int32_t bytes) {
+  for (int32_t i = 0; i < bytes; i += sizeof(TdUcs4)) {
+    int32_t f1 = *(int32_t *)((char *)f1_ucs4 + i);
+    int32_t f2 = *(int32_t *)((char *)f2_ucs4 + i);
+
+    if ((f1 == 0 && f2 != 0) || (f1 != 0 && f2 == 0)) {
+      return f1 - f2;
+    } else if (f1 == 0 && f2 == 0) {
+      return 0;
+    }
+
+    if (f1 != f2) {
+      return f1 - f2;
+    }
+  }
 
-int32_t taosUcs4ToMbs(void *ucs4, int32_t ucs4_max_len, char *mbs) {
+  return 0;
+
+#if 0
+  int32_t ucs4_max_len = bytes + 4;
+  char *f1_mbs = calloc(bytes, 1);
+  char *f2_mbs = calloc(bytes, 1);
+  if (taosUcs4ToMbs(f1_ucs4, ucs4_max_len, f1_mbs) < 0) {
+    return -1;
+  }
+  if (taosUcs4ToMbs(f2_ucs4, ucs4_max_len, f2_mbs) < 0) {
+    return -1;
+  }
+  int32_t ret = strcmp(f1_mbs, f2_mbs);
+  free(f1_mbs);
+  free(f2_mbs);
+  return ret;
+#endif
+}
+
+
+TdUcs4* tasoUcs4Copy(TdUcs4 *target_ucs4, TdUcs4 *source_ucs4, int32_t len_ucs4) {
+  assert(malloc_usable_size(target_ucs4)>=len_ucs4*sizeof(TdUcs4));
+  return memcpy(target_ucs4, source_ucs4, len_ucs4*sizeof(TdUcs4));
+}
+
+int32_t taosUcs4ToMbs(TdUcs4 *ucs4, int32_t ucs4_max_len, char *mbs) {
+#ifdef DISALLOW_NCHAR_WITHOUT_ICONV
+  printf("Nchar cannot be read and written without iconv, please install iconv library and recompile TDengine.\n");
+  return -1;
+#else
   iconv_t cd = iconv_open(tsCharset, DEFAULT_UNICODE_ENCODEC);
   size_t  ucs4_input_len = ucs4_max_len;
   size_t  outLen = ucs4_max_len;
@@ -39,14 +86,19 @@ int32_t taosUcs4ToMbs(void *ucs4, int32_t ucs4_max_len, char *mbs) {
 
   iconv_close(cd);
   return (int32_t)(ucs4_max_len - outLen);
+#endif
 }
 
-bool taosMbsToUcs4(char *mbs, size_t mbsLength, char *ucs4, int32_t ucs4_max_len, int32_t *len) {
+bool taosMbsToUcs4(const char *mbs, size_t mbsLength, TdUcs4 *ucs4, int32_t ucs4_max_len, int32_t *len) { 
+#ifdef DISALLOW_NCHAR_WITHOUT_ICONV
+  printf("Nchar cannot be read and written without iconv, please install iconv library and recompile TDengine.\n");
+  return -1;
+#else
   memset(ucs4, 0, ucs4_max_len);
   iconv_t cd = iconv_open(DEFAULT_UNICODE_ENCODEC, tsCharset);
   size_t  ucs4_input_len = mbsLength;
   size_t  outLeft = ucs4_max_len;
-  if (iconv(cd, &mbs, &ucs4_input_len, &ucs4, &outLeft) == -1) {
+  if (iconv(cd, (char**)&mbs, &ucs4_input_len, (char**)&ucs4, &outLeft) == -1) {
     iconv_close(cd);
     return false;
   }
@@ -60,9 +112,14 @@ bool taosMbsToUcs4(char *mbs, size_t mbsLength, char *ucs4, int32_t ucs4_max_len
   }
 
   return true;
+#endif
 }
 
 bool taosValidateEncodec(const char *encodec) {
+#ifdef DISALLOW_NCHAR_WITHOUT_ICONV
+  printf("Nchar cannot be read and written without iconv, please install iconv library and recompile TDengine.\n");
+  return true;
+#else
   iconv_t cd = iconv_open(encodec, DEFAULT_UNICODE_ENCODEC);
   if (cd == (iconv_t)(-1)) {
     return false;
@@ -70,214 +127,11 @@ bool taosValidateEncodec(const char *encodec) {
 
   iconv_close(cd);
   return true;
-}
-
-#else
-
-int32_t taosUcs4ToMbs(void *ucs4, int32_t ucs4_max_len, char *mbs) {
-  mbstate_t state = {0};
-  int32_t   len = (int32_t)wcsnrtombs(NULL, (const wchar_t **)&ucs4, ucs4_max_len / 4, 0, &state);
-  if (len < 0) {
-    return -1;
-  }
-
-  memset(&state, 0, sizeof(state));
-  len = wcsnrtombs(mbs, (const wchar_t **)&ucs4, ucs4_max_len / 4, (size_t)len, &state);
-  if (len < 0) {
-    return -1;
-  }
-
-  return len;
-}
-
-bool taosMbsToUcs4(const char *mbs, size_t mbsLength, char *ucs4, int32_t ucs4_max_len, int32_t *len) {
-  memset(ucs4, 0, ucs4_max_len);
-  mbstate_t state = {0};
-  int32_t retlen = mbsnrtowcs((wchar_t *)ucs4, (const char **)&mbs, mbsLength, ucs4_max_len / 4, &state);
-  *len = retlen;
-
-  return retlen >= 0;
-}
-
-bool taosValidateEncodec(const char *encodec) {
-  return true;
-}
-
 #endif
-
-#if defined(_TD_WINDOWS_64) || defined(_TD_WINDOWS_32)
-
-/*
- * windows implementation
- */
-
-#ifdef HAVE_CONFIG_H
-#include <config.h>
-#endif
-
-#include <sys/types.h>
-#include <stdio.h>
-#include <assert.h>
-#include <errno.h>
-#include <stdint.h>
-
-#if STDC_HEADERS
-#include <stdlib.h>
-#else
-char *malloc(), *realloc();
-#endif
-
-/* Always add at least this many bytes when extending the buffer.  */
-#define MIN_CHUNK 64
-
-/* Read up to (and including) a TERMINATOR from STREAM into *LINEPTR
-+ OFFSET (and null-terminate it). *LINEPTR is a pointer returned from
-malloc (or NULL), pointing to *N characters of space.  It is realloc'd
-as necessary.  Return the number of characters read (not including the
-null terminator), or -1 on error or EOF.  On a -1 return, the caller
-should check feof(), if not then errno has been set to indicate
-the error.  */
-
-int32_t getstr(char **lineptr, size_t *n, FILE *stream, char terminator, int32_t offset) {
-  int32_t nchars_avail; /* Allocated but unused chars in *LINEPTR.  */
-  char *  read_pos;     /* Where we're reading into *LINEPTR. */
-  int32_t ret;
-
-  if (!lineptr || !n || !stream) {
-    errno = EINVAL;
-    return -1;
-  }
-
-  if (!*lineptr) {
-    *n = MIN_CHUNK;
-    *lineptr = malloc(*n);
-    if (!*lineptr) {
-      errno = ENOMEM;
-      return -1;
-    }
-  }
-
-  nchars_avail = (int32_t)(*n - offset);
-  read_pos = *lineptr + offset;
-
-  for (;;) {
-    int32_t          save_errno;
-    register int32_t c = getc(stream);
-
-    save_errno = errno;
-
-    /* We always want at least one char left in the buffer, since we
-    always (unless we get an error while reading the first char)
-    NUL-terminate the line buffer.  */
-
-    assert((*lineptr + *n) == (read_pos + nchars_avail));
-    if (nchars_avail < 2) {
-      if (*n > MIN_CHUNK)
-        *n *= 2;
-      else
-        *n += MIN_CHUNK;
-
-      nchars_avail = (int32_t)(*n + *lineptr - read_pos);
-      char* lineptr1 = realloc(*lineptr, *n);
-      if (!lineptr1) {
-        errno = ENOMEM;
-        return -1;
-      }
-      *lineptr = lineptr1;
-
-      read_pos = *n - nchars_avail + *lineptr;
-      assert((*lineptr + *n) == (read_pos + nchars_avail));
-    }
-
-    if (ferror(stream)) {
-      /* Might like to return partial line, but there is no
-      place for us to store errno.  And we don't want to just
-      lose errno.  */
-      errno = save_errno;
-      return -1;
-    }
-
-    if (c == EOF) {
-      /* Return partial line, if any.  */
-      if (read_pos == *lineptr)
-        return -1;
-      else
-        break;
-    }
-
-    *read_pos++ = c;
-    nchars_avail--;
-
-    if (c == terminator) /* Return the line.  */
-      break;
-  }
-
-  /* Done - NUL terminate and return the number of chars read.  */
-  *read_pos = '\0';
-
-  ret = (int32_t)(read_pos - (*lineptr + offset));
-  return ret;
-}
-
-int32_t tgetline(char **lineptr, size_t *n, FILE *stream) { return getstr(lineptr, n, stream, '\n', 0); }
-
-
-/*
- * Get next token from string *stringp, where tokens are possibly-empty
- * strings separated by characters from delim.
- *
- * Writes NULs into the string at *stringp to end tokens.
- * delim need not remain constant from call to call.
- * On return, *stringp points past the last NUL written (if there might
- * be further tokens), or is NULL (if there are definitely no moretokens).
- *
- * If *stringp is NULL, strsep returns NULL.
- */
-char *strsep(char **stringp, const char *delim) {
-  char *      s;
-  const char *spanp;
-  int32_t     c, sc;
-  char *tok;
-  if ((s = *stringp) == NULL)
-    return (NULL);
-  for (tok = s;;) {
-    c = *s++;
-    spanp = delim;
-    do {
-      if ((sc = *spanp++) == c) {
-        if (c == 0)
-          s = NULL;
-        else
-          s[-1] = 0;
-        *stringp = s;
-        return (tok);
-      }
-    } while (sc != 0);
-  }
-  /* NOTREACHED */
-}
-
-char *getpass(const char *prefix) {
-  static char passwd[TSDB_PASSWORD_LEN] = {0};
-  memset(passwd, 0, TSDB_PASSWORD_LEN);
-  //printf("%s", prefix);
-
-  int32_t index = 0;
-  char    ch;
-  while (index < TSDB_PASSWORD_LEN) {
-    ch = getch();
-    if (ch == '\n' || ch == '\r') {
-      break;
-    } else {
-      passwd[index++] = ch;
-    }
-  }
-
-  return passwd;
 }
 
-int32_t twcslen(const wchar_t *wcs) {
-  int32_t *wstr = (int32_t *)wcs;
+int32_t taosUcs4len(TdUcs4 *ucs4) {
+  TdUcs4 *wstr = (TdUcs4 *)ucs4;
   if (NULL == wstr) {
     return 0;
   }
@@ -292,73 +146,15 @@ int32_t twcslen(const wchar_t *wcs) {
 
   return n;
 }
-int32_t tasoUcs4Compare(void *f1_ucs4, void *f2_ucs4, int32_t bytes) {
-  for (int32_t i = 0; i < bytes; i += TSDB_NCHAR_SIZE) {
-    int32_t f1 = *(int32_t *)((char *)f1_ucs4 + i);
-    int32_t f2 = *(int32_t *)((char *)f2_ucs4 + i);
-
-    if ((f1 == 0 && f2 != 0) || (f1 != 0 && f2 == 0)) {
-      return f1 - f2;
-    } else if (f1 == 0 && f2 == 0) {
-      return 0;
-    }
-
-    if (f1 != f2) {
-      return f1 - f2;
-    }
-  }
-
-  return 0;
-
-#if 0
-  int32_t ucs4_max_len = bytes + 4;
-  char *f1_mbs = calloc(bytes, 1);
-  char *f2_mbs = calloc(bytes, 1);
-  if (taosUcs4ToMbs(f1_ucs4, ucs4_max_len, f1_mbs) < 0) {
-    return -1;
-  }
-  if (taosUcs4ToMbs(f2_ucs4, ucs4_max_len, f2_mbs) < 0) {
-    return -1;
-  }
-  int32_t ret = strcmp(f1_mbs, f2_mbs);
-  free(f1_mbs);
-  free(f2_mbs);
-  return ret;
-#endif
-}
 
-/* Copy memory to memory until the specified number of bytes
-has been copied, return pointer to following byte.
-Overlap is NOT handled correctly. */
-void *mempcpy(void *dest, const void *src, size_t len) {
-  return (char*)memcpy(dest, src, len) + len;
-}
+int32_t taosWcharWidth(TdWchar wchar) { return wcwidth(wchar); }
 
-/* Copy SRC to DEST, returning the address of the terminating '\0' in DEST.  */
-char *stpcpy (char *dest, const char *src) {
-  size_t len = strlen (src);
-  return (char*)memcpy(dest, src, len + 1) + len;
-}
-
-/* Copy no more than N characters of SRC to DEST, returning the address of
-   the terminating '\0' in DEST, if any, or else DEST + N.  */
-char *stpncpy (char *dest, const char *src, size_t n) {
-  size_t size = strnlen (src, n);
-  memcpy (dest, src, size);
-  dest += size;
-  if (size == n)
-    return dest;
-  return memset (dest, '\0', n - size);
-}
+int32_t taosWcharsWidth(TdWchar *pWchar, int32_t size) { return wcswidth(pWchar, size); }
 
-#else
+int32_t taosMbToWchar(TdWchar *pWchar, const char *pStr, int32_t size) { return mbtowc(pWchar, pStr, size); }
 
-/*
- * linux and darwin implementation
- */
+int32_t taosMbsToWchars(TdWchar *pWchars, const char *pStrs, int32_t size) { return mbstowcs(pWchars, pStrs, size); }
 
-int32_t tasoUcs4Compare(void *f1_ucs4, void *f2_ucs4, int32_t bytes, int8_t ncharSize) {
-  return wcsncmp((wchar_t *)f1_ucs4, (wchar_t *)f2_ucs4, bytes / ncharSize);
-}
+int32_t taosWcharToMb(char *pStr, TdWchar wchar) { return wctomb(pStr, wchar); }
 
-#endif
+int32_t taosWcharsToMbs(char *pStrs, TdWchar *pWchars, int32_t size) { return wcstombs(pStrs, pWchars, size); }

source/util/src/tcompare.c

@@ -208,7 +208,7 @@ int32_t compareLenPrefixedWStr(const void *pLeft, const void *pRight) {
   if (len1 != len2) {
     return len1 > len2 ? 1 : -1;
   } else {
-    int32_t ret = memcmp((wchar_t *)pLeft, (wchar_t *)pRight, len1);
+    int32_t ret = memcmp((TdUcs4 *)pLeft, (TdUcs4 *)pRight, len1);
     if (ret == 0) {
       return 0;
     } else {
@@ -295,10 +295,10 @@ int32_t patternMatch(const char *patterStr, const char *str, size_t size, const
   return (str[j] == 0 || j >= size) ? TSDB_PATTERN_MATCH : TSDB_PATTERN_NOMATCH;
 }
 
-int32_t WCSPatternMatch(const wchar_t *patterStr, const wchar_t *str, size_t size, const SPatternCompareInfo *pInfo) {
-  wchar_t c, c1;
-  wchar_t matchOne = L'_';  // "_"
-  wchar_t matchAll = L'%';  // "%"
+int32_t WCSPatternMatch(const TdUcs4 *patterStr, const TdUcs4 *str, size_t size, const SPatternCompareInfo *pInfo) {
+  TdUcs4 c, c1;
+  TdUcs4 matchOne = L'_';  // "_"
+  TdUcs4 matchAll = L'%';  // "%"
 
   int32_t i = 0;
   int32_t j = 0;
@@ -315,7 +315,7 @@ int32_t WCSPatternMatch(const wchar_t *patterStr, const wchar_t *str, size_t siz
         return TSDB_PATTERN_MATCH;
       }
 
-      wchar_t accept[3] = {towupper(c), towlower(c), 0};
+      TdUcs4 accept[3] = {towupper(c), towlower(c), 0};
       while (1) {
         size_t n = wcscspn(str, accept);
 
@@ -424,10 +424,10 @@ int32_t compareWStrPatternMatch(const void *pLeft, const void *pRight) {
 
   assert(varDataLen(pRight) <= TSDB_MAX_FIELD_LEN * TSDB_NCHAR_SIZE);
 
-  wchar_t *pattern = calloc(varDataLen(pRight) + 1, sizeof(wchar_t));
+  char *pattern = calloc(varDataLen(pRight) + TSDB_NCHAR_SIZE, 1);
   memcpy(pattern, varDataVal(pRight), varDataLen(pRight));
 
-  int32_t ret = WCSPatternMatch(pattern, varDataVal(pLeft), varDataLen(pLeft) / TSDB_NCHAR_SIZE, &pInfo);
+  int32_t ret = WCSPatternMatch((TdUcs4*)pattern, (TdUcs4*)varDataVal(pLeft), varDataLen(pLeft) / TSDB_NCHAR_SIZE, &pInfo);
   free(pattern);
 
   return (ret == TSDB_PATTERN_MATCH) ? 0 : 1;
@@ -647,7 +647,7 @@ int32_t doCompare(const char *f1, const char *f2, int32_t type, size_t size) {
       if (t1->len != t2->len) {
         return t1->len > t2->len ? 1 : -1;
       }
-      int32_t ret = memcmp((wchar_t *)t1, (wchar_t *)t2, t2->len);
+      int32_t ret = memcmp((TdUcs4 *)t1, (TdUcs4 *)t2, t2->len);
       if (ret == 0) {
         return ret;
       }

source/util/test/encodeTest.cpp

@@ -230,7 +230,7 @@ static int32_t tSStructA_v1_decode(SCoder *pCoder, SStructA_v1 *pSAV1) {
   const char *tstr;
   uint64_t    len;
   if (tDecodeCStrAndLen(pCoder, &tstr, &len) < 0) return -1;
-  pSAV1->A_c = (char *)TCODER_MALLOC(len + 1, pCoder);
+  TCODER_MALLOC(pSAV1->A_c, char*, len + 1, pCoder);
   memcpy(pSAV1->A_c, tstr, len + 1);
 
   tEndDecode(pCoder);
@@ -269,7 +269,7 @@ static int32_t tSStructA_v2_decode(SCoder *pCoder, SStructA_v2 *pSAV2) {
   const char *tstr;
   uint64_t    len;
   if (tDecodeCStrAndLen(pCoder, &tstr, &len) < 0) return -1;
-  pSAV2->A_c = (char *)TCODER_MALLOC(len + 1, pCoder);
+  TCODER_MALLOC(pSAV2->A_c, char*, len + 1, pCoder);
   memcpy(pSAV2->A_c, tstr, len + 1);
 
   // ------------------------NEW FIELDS DECODE-------------------------------
@@ -305,7 +305,7 @@ static int32_t tSFinalReq_v1_encode(SCoder *pCoder, const SFinalReq_v1 *ps1) {
 static int32_t tSFinalReq_v1_decode(SCoder *pCoder, SFinalReq_v1 *ps1) {
   if (tStartDecode(pCoder) < 0) return -1;
 
-  ps1->pA = (SStructA_v1 *)TCODER_MALLOC(sizeof(*(ps1->pA)), pCoder);
+  TCODER_MALLOC(ps1->pA, SStructA_v1*, sizeof(*(ps1->pA)), pCoder);
   if (tSStructA_v1_decode(pCoder, ps1->pA) < 0) return -1;
   if (tDecodeI32(pCoder, &ps1->v_a) < 0) return -1;
   if (tDecodeI8(pCoder, &ps1->v_b) < 0) return -1;
@@ -339,7 +339,7 @@ static int32_t tSFinalReq_v2_encode(SCoder *pCoder, const SFinalReq_v2 *ps2) {
 static int32_t tSFinalReq_v2_decode(SCoder *pCoder, SFinalReq_v2 *ps2) {
   if (tStartDecode(pCoder) < 0) return -1;
 
-  ps2->pA = (SStructA_v2 *)TCODER_MALLOC(sizeof(*(ps2->pA)), pCoder);
+  TCODER_MALLOC(ps2->pA, SStructA_v2*, sizeof(*(ps2->pA)), pCoder);
   if (tSStructA_v2_decode(pCoder, ps2->pA) < 0) return -1;
   if (tDecodeI32(pCoder, &ps2->v_a) < 0) return -1;
   if (tDecodeI8(pCoder, &ps2->v_b) < 0) return -1;

source/util/test/freelistTest.cpp

@@ -8,7 +8,8 @@ TEST(TD_UTIL_FREELIST_TEST, simple_test) {
   tFreeListInit(&fl);
 
   for (size_t i = 0; i < 1000; i++) {
-    void *ptr = TFL_MALLOC(1024, &fl);
+    void *ptr = NULL;
+    TFL_MALLOC(ptr, void*, 1024, &fl);
     GTEST_ASSERT_NE(ptr, nullptr);
   }
 

tests/pytest/crash_gen/service_manager.py

@@ -378,7 +378,7 @@ class TdeSubProcess:
     @classmethod
     def _stopForSure(cls, proc: Popen, sig: int):
         ''' 
-        Stop a process and all sub processes with a singal, and SIGKILL if necessary
+        Stop a process and all sub processes with a signal, and SIGKILL if necessary
         '''
         def doKillTdService(proc: Popen, sig: int):
             Logging.info("Killing sub-sub process {} with signal {}".format(proc.pid, sig))

tests/script/jenkins/basic.txt

@@ -18,4 +18,7 @@
 
 # ---- insert
 ./test.sh -f tsim/insert/basic0.sim
+
+# ---- query
+./test.sh -f tsim/query/interval.sim
 #======================b1-end===============

tests/script/tsim/insert/basic1.sim

+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start
+sleep 50
+sql connect
+
+print =============== create database
+sql create database d1
+sql show databases
+if $rows != 1 then 
+  return -1
+endi
+
+print $data00 $data01 $data02
+
+sql use d1
+
+print =============== create super table, include all type
+sql create table if not exists stb (ts timestamp, c1 bool, c2 tinyint, c3 smallint, c4 int, c5 bigint, c6 float, c7 double, c8 binary(16), c9 nchar(16), c10 timestamp, c11 tinyint unsigned, c12 smallint unsigned, c13 int unsigned, c14 bigint unsigned) tags (t1 bool, t2 tinyint, t3 smallint, t4 int, t5 bigint, t6 float, t7 double, t8 binary(16), t9 nchar(16), t10 timestamp, t11 tinyint unsigned, t12 smallint unsigned, t13 int unsigned, t14 bigint unsigned)
+
+sql create stable if not exists stb_1 (ts timestamp, i int) tags (j int)
+sql create table stb_2 (ts timestamp, i int) tags (j int)
+sql create stable stb_3 (ts timestamp, i int) tags (j int)
+
+sql show stables
+if $rows != 4 then 
+  return -1
+endi
+
+print =============== create child table
+sql create table c1 using stb tags(true, -1, -2, -3, -4, -6.0, -7.0, 'child tbl 1', 'child tbl 1', '2022-02-25 18:00:00.000', 10, 20, 30, 40)
+sql create table c2 using stb tags(false, -1, -2, -3, -4, -6.0, -7.0, 'child tbl 2', 'child tbl 2', '2022-02-25 18:00:00.000', 10, 20, 30, 40)
+
+sql show tables
+if $rows != 2 then 
+  return -1
+endi
+
+
+print =============== insert data, mode1: one row one table in sql
+print =============== insert data, mode1: mulit rows one table in sql
+print =============== insert data, mode1: one rows mulit table in sql
+print =============== insert data, mode1: mulit rows mulit table in sql
+sql insert into c1 values(now+0s, true, -1, -2, -3, -4, -6.0, -7.0, 'child tbl 1', 'child tbl 1', '2022-02-25 18:00:00.000', 10, 20, 30, 40)
+sql insert into c1 values(now+0s, true, -1, -2, -3, -4, -6.0, -7.0, 'child tbl 1', 'child tbl 1', '2022-02-25 18:00:00.000', 10, 20, 30, 40) (now+1s, true, -1, -2, -3, -4, -6.0, -7.0, 'child tbl 1', 'child tbl 1', '2022-02-25 18:00:00.000', 10, 20, 30, 40) (now+2s, true, -1, -2, -3, -4, -6.0, -7.0, 'child tbl 1', 'child tbl 1', '2022-02-25 18:00:00.000', 10, 20, 30, 40)
+
+print =============== query data
+sql select * from c1
+if $rows != 4 then 
+  return -1
+endi
+
+if $data01 != true then 
+  return -1
+endi
+
+if $data02 != -1 then 
+  return -1
+endi
+
+if $data03 != -2 then 
+  return -1
+endi
+
+print =============== query data from st
+sql select * from st
+if $rows != 4 then 
+  return -1
+endi
+
+print =============== stop and restart taosd
+system sh/exec.sh -n dnode1 -s stop -x SIGINT
+system sh/exec.sh -n dnode1 -s start
+
+sleep 2000
+print =============== query data
+sql select * from c1
+if $rows != 4 then 
+  return -1
+endi
+
+if $data01 != true then 
+  return -1
+endi
+
+if $data02 != -1 then 
+  return -1
+endi
+
+if $data03 != -2 then 
+  return -1
+endi
+
+system sh/exec.sh -n dnode1 -s stop -x SIGINT

tests/script/tsim/query/interval.sim

@@ -47,161 +47,137 @@ $i = 1
 $tb = $tbPrefix . $i
 
 sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb interval(1m)
-print ===> $rows
-if $rows < $rowNum then 
-  return -1
-endi
-if $data01 != 1 then 
-  return -1
-endi
-if $data05 != 1 then 
-  return -1
-endi
-
-print =============== step3
-$cc = 4 * 60000
-$ms = 1601481600000 + $cc
-sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb  where ts <= $ms interval(1m)
-print ===> $rows
-if $rows > 10 then 
-  return -1
-endi
-if $rows < 3 then 
-  return -1
-endi
-if $data01 != 1 then 
-  return -1
-endi
-if $data05 != 1 then 
-  return -1
-endi
-
-print =============== step4
-$cc = 40 * 60000
-$ms = 1601481600000 + $cc
-
-$cc = 1 * 60000
-$ms2 = 1601481600000 - $cc
-
-sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb  where ts <= $ms and ts > $ms2 interval(1m)
-print ===> $rows
-if $rows < 18 then 
-  return -1
-endi
-if $rows > 22 then 
-  return -1
-endi
-if $data01 != 1 then 
-  return -1
-endi
-if $data05 != 1 then 
-  return -1
-endi
-
-print =============== step5
-$cc = 40 * 60000
-$ms = 1601481600000 + $cc
-
-$cc = 1 * 60000
-$ms2 = 1601481600000 - $cc
-
-sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb  where ts <= $ms and ts > $ms2 interval(1m) fill(value,0)
-print ===> $rows
-if $rows < 30 then 
-  return -1
-endi
-if $rows > 50 then 
-  return -1
-endi
-if $data21 != 1 then 
-  return -1
-endi
-if $data25 != 1 then 
-  return -1
-endi
-
-print =============== step6
-sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt interval(1m)
-print ===> $rows
-if $rows < 18 then 
-  return -1
-endi
-if $rows > 22 then 
-  return -1
-endi
-if $data11 > 15 then 
-  return -1
-endi
-if $data11 < 5 then 
-  return -1
-endi
-
-print =============== step7
-$cc = 4 * 60000
-$ms = 1601481600000 + $cc
-sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt  where ts <= $ms interval(1m)
-print ===> $rows
-if $rows < 3 then 
-  return -1
-endi
-if $rows > 7 then 
-  return -1
-endi
-if $data11 > 15 then 
-  return -1
-endi
-if $data11 < 5 then 
-  return -1
-endi
-
-print =============== step8
-$cc = 40 * 60000
-$ms1 = 1601481600000 + $cc
-
-$cc = 1 * 60000
-$ms2 = 1601481600000 - $cc
-
-sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt  where ts <= $ms1 and ts > $ms2 interval(1m)
-print ===> $rows
-if $rows < 18 then 
-  return -1
-endi
-if $rows > 22 then 
-  return -1
-endi
-if $data11 > 15 then 
-  return -1
-endi
-if $data11 < 5 then 
-  return -1
-endi
-
-print =============== step9
-$cc = 40 * 60000
-$ms1 = 1601481600000 + $cc
-
-$cc = 1 * 60000
-$ms2 = 1601481600000 - $cc
-
-sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt  where ts <= $ms1 and ts > $ms2 interval(1m) fill(value, 0)
-if $rows < 30 then 
-  return -1
-endi
-if $rows > 50 then 
-  return -1
-endi
-if $data11 > 15 then 
-  return -1
-endi
-if $data11 < 5 then 
-  return -1
-endi
+print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb interval(1m)
+print ===> $rows $data01 $data05
+if $rows != $rowNum then 
+  return -1
+endi
+if $data00 != 1 then 
+  return -1
+endi
+if $data04 != 1 then 
+  return -1
+endi
+
+#print =============== step3
+#$cc = 4 * 60000
+#$ms = 1601481600000 + $cc
+#sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb  where ts <= $ms interval(1m)
+#print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb  where ts <= $ms interval(1m)
+#print ===> $rows $data01 $data05
+#if $rows != 5 then 
+#  return -1
+#endi
+#if $data00 != 1 then 
+#  return -1
+#endi
+#if $data04 != 1 then 
+#  return -1
+#endi
+
+#print =============== step4
+#$cc = 40 * 60000
+#$ms = 1601481600000 + $cc
+
+#$cc = 1 * 60000
+#$ms2 = 1601481600000 - $cc
+
+#sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb  where ts <= $ms and ts > $ms2 interval(1m)
+#print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb  where ts <= $ms and ts > $ms2 interval(1m)
+#print ===> $rows $data01 $data05
+#if $rows != 20 then 
+#  return -1
+#endi
+#if $data00 != 1 then 
+#  return -1
+#endi
+#if $data04 != 1 then 
+#  return -1
+#endi
+
+#print =============== step5
+#$cc = 40 * 60000
+#$ms = 1601481600000 + $cc
+
+#$cc = 1 * 60000
+#$ms2 = 1601481600000 - $cc
+
+#sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb  where ts <= $ms and ts > $ms2 interval(1m) fill(value,0)
+#print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $tb  where ts <= $ms and ts > $ms2 interval(1m) fill(value,0)
+#print ===> $rows $data21 $data25
+#if $rows != 42 then 
+#  return -1
+#endi
+#if $data20 != 1 then 
+#  return -1
+#endi
+#if $data24 != 1 then 
+#  return -1
+#endi
+
+#print =============== step6
+#sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt interval(1m)
+#print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt interval(1m)
+#print ===> $rows $data11
+#if $rows != 20 then 
+#  return -1
+#endi
+#if $data11 != 10 then 
+#  return -1
+#endi
+
+#print =============== step7
+#$cc = 4 * 60000
+#$ms = 1601481600000 + $cc
+#sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt  where ts <= $ms interval(1m)
+#print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt  where ts <= $ms interval(1m)
+#print ===> $rows $data11
+#if $rows != 5 then 
+#  return -1
+#endi
+#if $data11 != 10 then 
+#  return -1
+#endi
+
+#print =============== step8
+#$cc = 40 * 60000
+#$ms1 = 1601481600000 + $cc
+#
+#$cc = 1 * 60000
+#$ms2 = 1601481600000 - $cc
+#
+#sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt  where ts <= $ms1 and ts > $ms2 interval(1m)
+#print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt  where ts <= $ms1 and ts > $ms2 interval(1m)
+#print ===> $rows $data11
+#if $rows != 20 then 
+#  return -1
+#endi
+#if $data11 != 10 then 
+#  return -1
+#endi
+#
+#print =============== step9
+#$cc = 40 * 60000
+#$ms1 = 1601481600000 + $cc
+#
+#$cc = 1 * 60000
+#$ms2 = 1601481600000 - $cc
+#
+#sql select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt  where ts <= $ms1 and ts > $ms2 interval(1m) fill(value, 0)
+#print ===> select count(tbcol), sum(tbcol), max(tbcol), min(tbcol), count(tbcol) from $mt  where ts <= $ms1 and ts > $ms2 interval(1m) fill(value, 0)
+#print ===> $rows $data11
+#if $rows != 42 then 
+#  return -1
+#endi
+#if $data11 != 10 then 
+#  return -1
+#endi
 
 print =============== clear
-sql drop database $db
-sql show databases
-if $rows != 0 then 
-  return -1
-endi
+#sql drop database $db
+#sql show databases
+#if $rows != 0 then 
+#  return -1
+#endi
 
-system sh/exec.sh -n dnode1 -s stop -x SIGINT
+#system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file

tools/shell/src/backup/shellDarwin.c

@@ -28,7 +28,6 @@
 int indicator = 1;
 struct termios oldtio;
 
-extern int wcwidth(wchar_t c);
 void insertChar(Command *cmd, char *c, int size);
 
 
@@ -426,7 +425,7 @@ void showOnScreen(Command *cmd) {
     w.ws_row = 30;
   }
 
-  wchar_t wc;
+  TdWchar wc;
   int size = 0;
 
   // Print out the command.
@@ -441,11 +440,11 @@ void showOnScreen(Command *cmd) {
   int remain_column = w.ws_col;
   /* size = cmd->commandSize + prompt_size; */
   for (char *str = total_string; size < cmd->commandSize + prompt_size;) {
-    int ret = mbtowc(&wc, str, MB_CUR_MAX);
+    int ret = taosMbToWchar(&wc, str, MB_CUR_MAX);
     if (ret < 0) break;
     size += ret;
     /* assert(size >= 0); */
-    int width = wcwidth(wc);
+    int width = taosWcharWidth(wc);
     if (remain_column > width) {
       printf("%lc", wc);
       remain_column -= width;

tools/shell/src/shellCommand.c

@@ -21,8 +21,6 @@
 
 #include <regex.h>
 
-extern int wcwidth(wchar_t c);
-extern int wcswidth(const wchar_t *s, size_t n);
 typedef struct {
   char widthInString;
   char widthOnScreen;
@@ -43,7 +41,7 @@ int countPrefixOnes(unsigned char c) {
 void getPrevCharSize(const char *str, int pos, int *size, int *width) {
   assert(pos > 0);
 
-  wchar_t wc;
+  TdWchar wc;
   *size = 0;
   *width = 0;
 
@@ -53,25 +51,25 @@ void getPrevCharSize(const char *str, int pos, int *size, int *width) {
     if (str[pos] > 0 || countPrefixOnes((unsigned char )str[pos]) > 1) break;
   }
 
-  int rc = mbtowc(&wc, str + pos, MB_CUR_MAX);
+  int rc = taosMbToWchar(&wc, str + pos, MB_CUR_MAX);
   assert(rc == *size);
 
-  *width = wcwidth(wc);
+  *width = taosWcharWidth(wc);
 }
 
 void getNextCharSize(const char *str, int pos, int *size, int *width) {
   assert(pos >= 0);
 
-  wchar_t wc;
-  *size = mbtowc(&wc, str + pos, MB_CUR_MAX);
-  *width = wcwidth(wc);
+  TdWchar wc;
+  *size = taosMbToWchar(&wc, str + pos, MB_CUR_MAX);
+  *width = taosWcharWidth(wc);
 }
 
 void insertChar(Command *cmd, char *c, int size) {
   assert(cmd->cursorOffset <= cmd->commandSize && cmd->endOffset >= cmd->screenOffset);
 
-  wchar_t wc;
-  if (mbtowc(&wc, c, size) < 0) return;
+  TdWchar wc;
+  if (taosMbToWchar(&wc, c, size) < 0) return;
 
   clearScreen(cmd->endOffset + prompt_size, cmd->screenOffset + prompt_size);
   /* update the buffer */
@@ -81,8 +79,8 @@ void insertChar(Command *cmd, char *c, int size) {
   /* update the values */
   cmd->commandSize += size;
   cmd->cursorOffset += size;
-  cmd->screenOffset += wcwidth(wc);
-  cmd->endOffset += wcwidth(wc);
+  cmd->screenOffset += taosWcharWidth(wc);
+  cmd->endOffset += taosWcharWidth(wc);
   showOnScreen(cmd);
 }
 
@@ -249,10 +247,10 @@ int isReadyGo(Command *cmd) {
 }
 
 void getMbSizeInfo(const char *str, int *size, int *width) {
-  wchar_t *wc = (wchar_t *)calloc(sizeof(wchar_t), MAX_COMMAND_SIZE);
+  TdWchar *wc = (TdWchar *)calloc(sizeof(TdWchar), MAX_COMMAND_SIZE);
   *size = strlen(str);
-  mbstowcs(wc, str, MAX_COMMAND_SIZE);
-  *width = wcswidth(wc, MAX_COMMAND_SIZE);
+  taosMbsToWchars(wc, str, MAX_COMMAND_SIZE);
+  *width = taosWcharsWidth(wc, MAX_COMMAND_SIZE);
   free(wc);
 }
 

tools/shell/src/shellEngine.c

@@ -560,12 +560,12 @@ static int dumpResultToFile(const char *fname, TAOS_RES *tres) {
 }
 
 static void shellPrintNChar(const char *str, int length, int width) {
-  wchar_t tail[3];
+  TdWchar tail[3];
   int     pos = 0, cols = 0, totalCols = 0, tailLen = 0;
 
   while (pos < length) {
-    wchar_t wc;
-    int     bytes = mbtowc(&wc, str + pos, MB_CUR_MAX);
+    TdWchar wc;
+    int     bytes = taosMbToWchar(&wc, str + pos, MB_CUR_MAX);
     if (bytes == 0) {
       break;
     }
@@ -577,7 +577,7 @@ static void shellPrintNChar(const char *str, int length, int width) {
 #ifdef WINDOWS
     int w = bytes;
 #else
-    int w = wcwidth(wc);
+    int w = taosWcharWidth(wc);
 #endif
     if (w <= 0) {
       continue;

tools/shell/src/shellLinux.c

@@ -31,7 +31,6 @@
 int indicator = 1;
 struct termios oldtio;
 
-extern int wcwidth(wchar_t c);
 void insertChar(Command *cmd, char *c, int size);
 const char *argp_program_version = version;
 const char *argp_program_bug_address = "<support@taosdata.com>";
@@ -456,7 +455,7 @@ void showOnScreen(Command *cmd) {
     w.ws_row = 30;
   }
 
-  wchar_t wc;
+  TdWchar wc;
   int size = 0;
 
   // Print out the command.
@@ -471,11 +470,11 @@ void showOnScreen(Command *cmd) {
   int remain_column = w.ws_col;
   /* size = cmd->commandSize + prompt_size; */
   for (char *str = total_string; size < cmd->commandSize + prompt_size;) {
-    int ret = mbtowc(&wc, str, MB_CUR_MAX);
+    int ret = taosMbToWchar(&wc, str, MB_CUR_MAX);
     if (ret < 0) break;
     size += ret;
     /* assert(size >= 0); */
-    int width = wcwidth(wc);
+    int width = taosWcharWidth(wc);
     if (remain_column > width) {
       printf("%lc", wc);
       remain_column -= width;