Merge branch '3.0' into cpwu/3.0

Jenkinsfile2

@@ -117,27 +117,29 @@ def pre_test(){
 def pre_test_win(){
     bat '''
         hostname
+        ipconfig
+        set
         date /t
         time /t
-        taskkill /f /t /im python.exe
-        taskkill /f /t /im bash.exe
         rd /s /Q C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\debug
-        exit 0
     '''
     bat '''
         cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
         git reset --hard
         git fetch || git fetch
+    '''
+    bat '''
         cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
         git reset --hard
         git fetch || git fetch
-        git checkout -f
     '''
     script {
         if (env.CHANGE_TARGET == 'master') {
             bat '''
                 cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
                 git checkout master
+            '''
+            bat '''
                 cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
                 git checkout master
             '''
@@ -145,6 +147,8 @@ def pre_test_win(){
             bat '''
                 cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
                 git checkout 2.0
+            '''
+            bat '''
                 cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
                 git checkout 2.0
             '''
@@ -152,6 +156,8 @@ def pre_test_win(){
             bat '''
                 cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
                 git checkout 3.0
+            '''
+            bat '''
                 cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
                 git checkout 3.0
             '''
@@ -159,6 +165,8 @@ def pre_test_win(){
             bat '''
                 cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
                 git checkout develop
+            '''
+            bat '''
                 cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
                 git checkout develop
             '''
@@ -169,30 +177,52 @@ def pre_test_win(){
             bat '''
                 cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
                 git pull
-                git log -5
+            '''
+            bat '''
                 cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
                 git pull
-                git fetch origin +refs/pull/${CHANGE_ID}/merge
+            '''
+            bat '''
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
+                git fetch origin +refs/pull/%CHANGE_ID%/merge
+            '''
+            bat '''
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
                 git checkout -qf FETCH_HEAD
-                git log -5
             '''
         } else if (env.CHANGE_URL =~ /\/TDinternal\//) {
             bat '''
                 cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
                 git pull
-                git fetch origin +refs/pull/${CHANGE_ID}/merge
+            '''
+            bat '''
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
+                git fetch origin +refs/pull/%CHANGE_ID%/merge
+            '''
+            bat '''
+                cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
                 git checkout -qf FETCH_HEAD
-                git log -5
+            '''
+            bat '''
                 cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
                 git pull
-                git log -5
             '''
         } else {
-            sh '''
-                echo "unmatched reposiotry ${CHANGE_URL}"
+            bat '''
+                echo "unmatched reposiotry %CHANGE_URL%"
             '''
         }
     }
+    bat '''
+        cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
+        git branch
+        git log -5
+    '''
+    bat '''
+        cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
+        git branch
+        git log -5
+    '''
     bat '''
         cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal\\community
         git submodule update --init --recursive
@@ -205,10 +235,15 @@ def pre_test_build_win() {
         cd C:\\workspace\\%EXECUTOR_NUMBER%\\TDinternal
         mkdir debug
         cd debug
+        time /t
         call "C:\\Program Files (x86)\\Microsoft Visual Studio\\2017\\Community\\VC\\Auxiliary\\Build\\vcvarsall.bat" x64
         set CL=/MP8
-        cmake .. -G "NMake Makefiles JOM"
-        jom -j 4 || exit 8
+        echo ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> cmake"
+        time /t
+        cmake .. -G "NMake Makefiles JOM" || exit 7
+        echo ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> jom -j 6"
+        time /t
+        jom -j 6 || exit 8
         time /t
     '''
     return 1
@@ -226,6 +261,13 @@ pipeline {
     stages {
         stage('run test') {
             parallel {
+                stage('windows test') {
+                    agent{label " windows10_01 || windows10_02 || windows10_03 || windows10_04 "}
+                    steps {
+                        pre_test_win()
+                        pre_test_build_win()
+                    }
+                }
                 stage('linux test') {
                     agent{label " slave3_0 || slave15 || slave16 || slave17 "}
                     options { skipDefaultCheckout() } 

cmake/cmake.install

@@ -5,22 +5,27 @@ IF (TD_LINUX)
 ELSEIF (TD_WINDOWS)
   SET(CMAKE_INSTALL_PREFIX C:/TDengine)
 
-  INSTALL(DIRECTORY ${TD_SOURCE_DIR}/src/connector/go     DESTINATION connector)
-  INSTALL(DIRECTORY ${TD_SOURCE_DIR}/src/connector/nodejs DESTINATION connector)
-  INSTALL(DIRECTORY ${TD_SOURCE_DIR}/src/connector/python DESTINATION connector)
-  INSTALL(DIRECTORY ${TD_SOURCE_DIR}/src/connector/C\# DESTINATION connector)
-  INSTALL(DIRECTORY ${TD_SOURCE_DIR}/examples DESTINATION .)
+  # INSTALL(DIRECTORY ${TD_SOURCE_DIR}/src/connector/go     DESTINATION connector)
+  # INSTALL(DIRECTORY ${TD_SOURCE_DIR}/src/connector/nodejs DESTINATION connector)
+  # INSTALL(DIRECTORY ${TD_SOURCE_DIR}/src/connector/python DESTINATION connector)
+  # INSTALL(DIRECTORY ${TD_SOURCE_DIR}/src/connector/C\# DESTINATION connector)
+  # INSTALL(DIRECTORY ${TD_SOURCE_DIR}/examples DESTINATION .)
   INSTALL(FILES ${TD_SOURCE_DIR}/packaging/cfg/taos.cfg DESTINATION cfg)
-  INSTALL(FILES ${TD_SOURCE_DIR}/src/inc/taos.h DESTINATION include)
-  INSTALL(FILES ${TD_SOURCE_DIR}/src/inc/taoserror.h DESTINATION include)
+  INSTALL(FILES ${TD_SOURCE_DIR}/include/client/taos.h DESTINATION include)
+  INSTALL(FILES ${TD_SOURCE_DIR}/include/util/taoserror.h DESTINATION include)
   INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos.lib DESTINATION driver)
   INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos_static.lib DESTINATION driver)
-  INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos.exp DESTINATION driver)
   INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos.dll DESTINATION driver)
+  INSTALL(FILES ${EXECUTABLE_OUTPUT_PATH}/taos.exe DESTINATION .)
+  INSTALL(FILES ${EXECUTABLE_OUTPUT_PATH}/taosd.exe DESTINATION .)
+  INSTALL(FILES ${EXECUTABLE_OUTPUT_PATH}/udfd.exe DESTINATION .)
 
   IF (TD_MVN_INSTALLED)
     INSTALL(FILES ${LIBRARY_OUTPUT_PATH}/taos-jdbcdriver-2.0.38-dist.jar DESTINATION connector/jdbc)
   ENDIF ()
+  SET(TD_MAKE_INSTALL_SH "${TD_SOURCE_DIR}/packaging/tools/make_install.bat")
+  INSTALL(CODE "MESSAGE(\"make install script: ${TD_MAKE_INSTALL_SH}\")")
+  INSTALL(CODE "execute_process(COMMAND ${TD_MAKE_INSTALL_SH} :needAdmin ${TD_SOURCE_DIR} ${PROJECT_BINARY_DIR} Windows ${TD_VER_NUMBER})")
 ELSEIF (TD_DARWIN)
   SET(TD_MAKE_INSTALL_SH "${TD_SOURCE_DIR}/packaging/tools/make_install.sh")
   INSTALL(CODE "MESSAGE(\"make install script: ${TD_MAKE_INSTALL_SH}\")")

contrib/CMakeLists.txt

@@ -226,10 +226,10 @@ endif(${BUILD_WITH_NURAFT})
 if(${BUILD_PTHREAD})
     set(CMAKE_BUILD_TYPE release)
     add_definitions(-DPTW32_STATIC_LIB)
-    add_subdirectory(pthread)
+    add_subdirectory(pthread EXCLUDE_FROM_ALL)
     set_target_properties(libpthreadVC3 PROPERTIES OUTPUT_NAME pthread)
-    add_library(pthread STATIC IMPORTED GLOBAL)
-    SET_PROPERTY(TARGET pthread PROPERTY IMPORTED_LOCATION ${LIBRARY_OUTPUT_PATH}/pthread.lib)
+    add_library(pthread INTERFACE)
+    target_link_libraries(pthread INTERFACE libpthreadVC3)
 endif()
 
 # iconv

example/src/tmq.c

@@ -167,7 +167,7 @@ tmq_t* build_consumer() {
   tmq_conf_set(conf, "td.connect.pass", "taosdata");
   /*tmq_conf_set(conf, "td.connect.db", "abc1");*/
   tmq_conf_set(conf, "msg.with.table.name", "true");
-  tmq_conf_set_offset_commit_cb(conf, tmq_commit_cb_print, NULL);
+  tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL);
   tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
   assert(tmq);
   return tmq;
@@ -176,6 +176,7 @@ tmq_t* build_consumer() {
 tmq_list_t* build_topic_list() {
   tmq_list_t* topic_list = tmq_list_new();
   tmq_list_append(topic_list, "topic_ctb_column");
+  /*tmq_list_append(topic_list, "tmq_test_db_multi_insert_topic");*/
   return topic_list;
 }
 
@@ -190,7 +191,7 @@ void basic_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
   int32_t cnt = 0;
   /*clock_t startTime = clock();*/
   while (running) {
-    TAOS_RES* tmqmessage = tmq_consumer_poll(tmq, 500);
+    TAOS_RES* tmqmessage = tmq_consumer_poll(tmq, 0);
     if (tmqmessage) {
       cnt++;
       /*printf("get data\n");*/
@@ -238,7 +239,7 @@ void sync_consume_loop(tmq_t* tmq, tmq_list_t* topics) {
       msg_process(tmqmessage);
       taos_free_result(tmqmessage);
 
-      tmq_commit(tmq, NULL, 1);
+      tmq_commit_async(tmq, NULL, tmq_commit_cb_print, NULL);
       /*if ((++msg_count % MIN_COMMIT_COUNT) == 0) tmq_commit(tmq, NULL, 0);*/
     }
   }

include/client/taos.h

@@ -232,9 +232,11 @@ DLL_EXPORT tmq_resp_err_t tmq_unsubscribe(tmq_t *tmq);
 DLL_EXPORT tmq_resp_err_t tmq_subscription(tmq_t *tmq, tmq_list_t **topics);
 DLL_EXPORT TAOS_RES      *tmq_consumer_poll(tmq_t *tmq, int64_t wait_time);
 DLL_EXPORT tmq_resp_err_t tmq_consumer_close(tmq_t *tmq);
-DLL_EXPORT tmq_resp_err_t tmq_commit(tmq_t *tmq, const tmq_topic_vgroup_list_t *offsets, int32_t async);
+DLL_EXPORT tmq_resp_err_t tmq_commit_sync(tmq_t *tmq, const tmq_topic_vgroup_list_t *offsets);
+DLL_EXPORT void tmq_commit_async(tmq_t *tmq, const tmq_topic_vgroup_list_t *offsets, tmq_commit_cb *cb, void *param);
+
 #if 0
-DLL_EXPORT tmq_resp_err_t tmq_commit_message(tmq_t* tmq, const tmq_message_t* tmqmessage, int32_t async);
+DLL_EXPORT tmq_resp_err_t tmq_commit(tmq_t *tmq, const tmq_topic_vgroup_list_t *offsets, int32_t async);
 DLL_EXPORT tmq_resp_err_t tmq_seek(tmq_t *tmq, const tmq_topic_vgroup_t *offset);
 #endif
 
@@ -251,7 +253,7 @@ typedef enum tmq_conf_res_t tmq_conf_res_t;
 DLL_EXPORT tmq_conf_t    *tmq_conf_new();
 DLL_EXPORT tmq_conf_res_t tmq_conf_set(tmq_conf_t *conf, const char *key, const char *value);
 DLL_EXPORT void           tmq_conf_destroy(tmq_conf_t *conf);
-DLL_EXPORT void           tmq_conf_set_offset_commit_cb(tmq_conf_t *conf, tmq_commit_cb *cb, void *param);
+DLL_EXPORT void           tmq_conf_set_auto_commit_cb(tmq_conf_t *conf, tmq_commit_cb *cb, void *param);
 
 /* -------------------------TMQ MSG HANDLE INTERFACE---------------------- */
 

include/common/tdataformat.h

@@ -29,27 +29,42 @@ extern "C" {
 typedef struct SSchema       SSchema;
 typedef struct STColumn      STColumn;
 typedef struct STSchema      STSchema;
+typedef struct SColVal       SColVal;
 typedef struct STSRow2       STSRow2;
 typedef struct STSRowBuilder STSRowBuilder;
-typedef struct SKVIdx        SKVIdx;
-
-// STSchema
-
-// STSRow2
-int32_t tEncodeTSRow(SEncoder *pEncoder, const STSRow2 *pRow);
-int32_t tDecodeTSRow(SDecoder *pDecoder, STSRow2 *pRow);
+typedef struct STagVal       STagVal;
+typedef struct STag          STag;
 
 // STSchema
 int32_t tTSchemaCreate(int32_t sver, SSchema *pSchema, int32_t nCols, STSchema **ppTSchema);
 void    tTSchemaDestroy(STSchema *pTSchema);
 
+// SColVal
+#define ColValNONE               ((SColVal){.type = COL_VAL_NONE, .nData = 0, .pData = NULL})
+#define ColValNULL               ((SColVal){.type = COL_VAL_NULL, .nData = 0, .pData = NULL})
+#define ColValDATA(nData, pData) ((SColVal){.type = COL_VAL_DATA, .nData = (nData), .pData = (pData)})
+
+// STSRow2
+int32_t tPutTSRow(uint8_t *p, STSRow2 *pRow);
+int32_t tGetTSRow(uint8_t *p, STSRow2 *pRow);
+int32_t tTSRowDup(const STSRow2 *pRow, STSRow2 **ppRow);
+void    tTSRowFree(STSRow2 *pRow);
+int32_t tTSRowGet(const STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
+
 // STSRowBuilder
-int32_t tTSRowBuilderInit(STSRowBuilder *pBuilder, int32_t sver, SSchema *pSchema, int32_t nCols);
+int32_t tTSRowBuilderInit(STSRowBuilder *pBuilder, int32_t sver, int32_t nCols, SSchema *pSchema);
 void    tTSRowBuilderClear(STSRowBuilder *pBuilder);
 void    tTSRowBuilderReset(STSRowBuilder *pBuilder);
-int32_t tTSRowBuilderPut(STSRowBuilder *pBuilder, int32_t cid, const uint8_t *pData, uint32_t nData);
+int32_t tTSRowBuilderPut(STSRowBuilder *pBuilder, int32_t cid, uint8_t *pData, uint32_t nData);
 int32_t tTSRowBuilderGetRow(STSRowBuilder *pBuilder, const STSRow2 **ppRow);
 
+// STag
+int32_t tTagNew(STagVal *pTagVals, int16_t nTag, STag **ppTag);
+void    tTagFree(STag *pTag);
+void    tTagGet(STag *pTag, int16_t cid, int8_t type, uint8_t **ppData, int32_t *nData);
+int32_t tEncodeTag(SEncoder *pEncoder, STag *pTag);
+int32_t tDecodeTag(SDecoder *pDecoder, const STag **ppTag);
+
 // STRUCT =================
 struct STColumn {
   col_id_t colId;
@@ -68,31 +83,47 @@ struct STSchema {
   STColumn columns[];
 };
 
+#define TSROW_HAS_NONE ((uint8_t)0x1)
+#define TSROW_HAS_NULL ((uint8_t)0x2U)
+#define TSROW_HAS_VAL  ((uint8_t)0x4U)
+#define TSROW_KV_ROW   ((uint8_t)0x10U)
 struct STSRow2 {
   TSKEY    ts;
-  uint32_t flags;
-  union {
+  uint8_t  flags;
   int32_t  sver;
-    int32_t ncols;
-  };
   uint32_t nData;
-  const uint8_t *pData;
+  uint8_t *pData;
 };
 
 struct STSRowBuilder {
-  STColumn *pTColumn;
   STSchema *pTSchema;
+  int32_t   szBitMap1;
+  int32_t   szBitMap2;
   int32_t   szKVBuf;
   uint8_t  *pKVBuf;
   int32_t   szTPBuf;
   uint8_t  *pTPBuf;
-  int32_t   nCols;
-  int32_t   kvVLen;
-  int32_t   tpVLen;
+  int32_t   iCol;
+  int32_t   vlenKV;
+  int32_t   vlenTP;
   STSRow2   row;
 };
 
-#if 1  //====================================
+typedef enum { COL_VAL_NONE = 0, COL_VAL_NULL = 1, COL_VAL_DATA = 2 } EColValT;
+struct SColVal {
+  EColValT type;
+  uint32_t nData;
+  uint8_t *pData;
+};
+
+struct STagVal {
+  int16_t  cid;
+  int8_t   type;
+  uint32_t nData;
+  uint8_t *pData;
+};
+
+#if 1  //================================================================================================================================================
 // Imported since 3.0 and use bitmap to demonstrate None/Null/Norm, while use Null/Norm below 3.0 without of bitmap.
 #define TD_SUPPORT_BITMAP
 #define TD_SUPPORT_READ2

include/common/tglobal.h

@@ -32,6 +32,7 @@ extern char     tsLocalEp[];
 extern uint16_t tsServerPort;
 extern int32_t  tsVersion;
 extern int32_t  tsStatusInterval;
+extern int32_t  tsNumOfSupportVnodes;
 
 // common
 extern int32_t tsMaxShellConns;
@@ -45,7 +46,7 @@ extern bool    tsPrintAuth;
 extern int64_t tsTickPerMin[3];
 
 // multi-process
-extern bool    tsMultiProcess;
+extern int32_t tsMultiProcess;
 extern int32_t tsMnodeShmSize;
 extern int32_t tsVnodeShmSize;
 extern int32_t tsQnodeShmSize;

include/common/tmsg.h

@@ -258,6 +258,7 @@ typedef struct {
   char*   tblFName;
   int32_t numOfRows;
   int32_t affectedRows;
+  int64_t sver;
 } SSubmitBlkRsp;
 
 typedef struct {
@@ -276,8 +277,8 @@ void    tFreeSSubmitRsp(SSubmitRsp* pRsp);
 
 #define COL_SMA_ON   ((int8_t)0x1)
 #define COL_IDX_ON   ((int8_t)0x2)
-#define COL_VAL_SET ((int8_t)0x4)
-
+#define COL_SET_NULL ((int8_t)0x10)
+#define COL_SET_VAL  ((int8_t)0x20)
 typedef struct SSchema {
   int8_t   type;
   int8_t   flags;
@@ -286,6 +287,9 @@ typedef struct SSchema {
   char     name[TSDB_COL_NAME_LEN];
 } SSchema;
 
+#define COL_IS_SET(FLG)  ((FLG) & (COL_SET_VAL | COL_SET_NULL) != 0)
+#define COL_CLR_SET(FLG) ((FLG) &= (~(COL_SET_VAL | COL_SET_NULL)))
+
 #define IS_BSMA_ON(s) (((s)->flags & 0x01) == COL_SMA_ON)
 
 #define SSCHMEA_TYPE(s)  ((s)->type)
@@ -297,6 +301,8 @@ typedef struct SSchema {
 typedef struct {
   int32_t  nCols;
   int32_t  sver;
+  int32_t  tagVer;
+  int32_t  colVer;
   SSchema* pSchema;
 } SSchemaWrapper;
 
@@ -305,6 +311,8 @@ static FORCE_INLINE SSchemaWrapper* tCloneSSchemaWrapper(const SSchemaWrapper* p
   if (pSW == NULL) return pSW;
   pSW->nCols = pSchemaWrapper->nCols;
   pSW->sver = pSchemaWrapper->sver;
+  pSW->tagVer = pSchemaWrapper->tagVer;
+  pSW->colVer = pSchemaWrapper->colVer;
   pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema));
   if (pSW->pSchema == NULL) {
     taosMemoryFree(pSW);
@@ -360,6 +368,8 @@ static FORCE_INLINE int32_t taosEncodeSSchemaWrapper(void** buf, const SSchemaWr
   int32_t tlen = 0;
   tlen += taosEncodeVariantI32(buf, pSW->nCols);
   tlen += taosEncodeVariantI32(buf, pSW->sver);
+  tlen += taosEncodeVariantI32(buf, pSW->tagVer);
+  tlen += taosEncodeVariantI32(buf, pSW->colVer);
   for (int32_t i = 0; i < pSW->nCols; i++) {
     tlen += taosEncodeSSchema(buf, &pSW->pSchema[i]);
   }
@@ -369,6 +379,8 @@ static FORCE_INLINE int32_t taosEncodeSSchemaWrapper(void** buf, const SSchemaWr
 static FORCE_INLINE void* taosDecodeSSchemaWrapper(const void* buf, SSchemaWrapper* pSW) {
   buf = taosDecodeVariantI32(buf, &pSW->nCols);
   buf = taosDecodeVariantI32(buf, &pSW->sver);
+  buf = taosDecodeVariantI32(buf, &pSW->tagVer);
+  buf = taosDecodeVariantI32(buf, &pSW->colVer);
   pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema));
   if (pSW->pSchema == NULL) {
     return NULL;
@@ -383,6 +395,8 @@ static FORCE_INLINE void* taosDecodeSSchemaWrapper(const void* buf, SSchemaWrapp
 static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SEncoder* pEncoder, const SSchemaWrapper* pSW) {
   if (tEncodeI32v(pEncoder, pSW->nCols) < 0) return -1;
   if (tEncodeI32v(pEncoder, pSW->sver) < 0) return -1;
+  if (tEncodeI32v(pEncoder, pSW->tagVer) < 0) return -1;
+  if (tEncodeI32v(pEncoder, pSW->colVer) < 0) return -1;
   for (int32_t i = 0; i < pSW->nCols; i++) {
     if (tEncodeSSchema(pEncoder, &pSW->pSchema[i]) < 0) return -1;
   }
@@ -393,6 +407,8 @@ static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SEncoder* pEncoder, const SSch
 static FORCE_INLINE int32_t tDecodeSSchemaWrapper(SDecoder* pDecoder, SSchemaWrapper* pSW) {
   if (tDecodeI32v(pDecoder, &pSW->nCols) < 0) return -1;
   if (tDecodeI32v(pDecoder, &pSW->sver) < 0) return -1;
+  if (tDecodeI32v(pDecoder, &pSW->tagVer) < 0) return -1;
+  if (tDecodeI32v(pDecoder, &pSW->colVer) < 0) return -1;
 
   pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema));
   if (pSW->pSchema == NULL) return -1;
@@ -403,6 +419,21 @@ static FORCE_INLINE int32_t tDecodeSSchemaWrapper(SDecoder* pDecoder, SSchemaWra
   return 0;
 }
 
+static FORCE_INLINE int32_t tDecodeSSchemaWrapperEx(SDecoder* pDecoder, SSchemaWrapper* pSW) {
+  if (tDecodeI32v(pDecoder, &pSW->nCols) < 0) return -1;
+  if (tDecodeI32v(pDecoder, &pSW->sver) < 0) return -1;
+  if (tDecodeI32v(pDecoder, &pSW->tagVer) < 0) return -1;
+  if (tDecodeI32v(pDecoder, &pSW->colVer) < 0) return -1;
+
+  pSW->pSchema = (SSchema*)tDecoderMalloc(pDecoder, pSW->nCols * sizeof(SSchema));
+  if (pSW->pSchema == NULL) return -1;
+  for (int32_t i = 0; i < pSW->nCols; i++) {
+    if (tDecodeSSchema(pDecoder, &pSW->pSchema[i]) < 0) return -1;
+  }
+
+  return 0;
+}
+
 STSchema* tdGetSTSChemaFromSSChema(SSchema** pSchema, int32_t nCols);
 
 typedef struct {
@@ -438,6 +469,7 @@ int32_t tDeserializeSMDropStbReq(void* buf, int32_t bufLen, SMDropStbReq* pReq);
 typedef struct {
   char    name[TSDB_TABLE_FNAME_LEN];
   int8_t  alterType;
+  int32_t verInBlock;
   int32_t numOfFields;
   SArray* pFields;
   int32_t ttl;
@@ -1463,6 +1495,7 @@ typedef struct {
 typedef struct {
   int64_t consumerId;
   char    cgroup[TSDB_CGROUP_LEN];
+  char    clientId[256];
   SArray* topicNames;  // SArray<char**>
 } SCMSubscribeReq;
 
@@ -1470,6 +1503,7 @@ static FORCE_INLINE int32_t tSerializeSCMSubscribeReq(void** buf, const SCMSubsc
   int32_t tlen = 0;
   tlen += taosEncodeFixedI64(buf, pReq->consumerId);
   tlen += taosEncodeString(buf, pReq->cgroup);
+  tlen += taosEncodeString(buf, pReq->clientId);
 
   int32_t topicNum = taosArrayGetSize(pReq->topicNames);
   tlen += taosEncodeFixedI32(buf, topicNum);
@@ -1483,6 +1517,7 @@ static FORCE_INLINE int32_t tSerializeSCMSubscribeReq(void** buf, const SCMSubsc
 static FORCE_INLINE void* tDeserializeSCMSubscribeReq(void* buf, SCMSubscribeReq* pReq) {
   buf = taosDecodeFixedI64(buf, &pReq->consumerId);
   buf = taosDecodeStringTo(buf, pReq->cgroup);
+  buf = taosDecodeStringTo(buf, pReq->clientId);
 
   int32_t topicNum;
   buf = taosDecodeFixedI32(buf, &topicNum);
@@ -1613,6 +1648,15 @@ typedef struct {
 int32_t tSerializeSMDropTopicReq(void* buf, int32_t bufLen, SMDropTopicReq* pReq);
 int32_t tDeserializeSMDropTopicReq(void* buf, int32_t bufLen, SMDropTopicReq* pReq);
 
+typedef struct {
+  char   topic[TSDB_TOPIC_FNAME_LEN];
+  char   cgroup[TSDB_CGROUP_LEN];
+  int8_t igNotExists;
+} SMDropCgroupReq;
+
+int32_t tSerializeSMDropCgroupReq(void* buf, int32_t bufLen, SMDropCgroupReq* pReq);
+int32_t tDeserializeSMDropCgroupReq(void* buf, int32_t bufLen, SMDropCgroupReq* pReq);
+
 typedef struct {
   char    name[TSDB_TABLE_FNAME_LEN];
   int8_t  alterType;

include/common/tmsgcb.h

@@ -25,6 +25,7 @@ extern "C" {
 typedef struct SRpcMsg        SRpcMsg;
 typedef struct SEpSet         SEpSet;
 typedef struct SMgmtWrapper   SMgmtWrapper;
+typedef struct SRpcHandleInfo SRpcHandleInfo;
 
 typedef enum {
   QUERY_QUEUE,
@@ -37,41 +38,36 @@ typedef enum {
   QUEUE_MAX,
 } EQueueType;
 
-typedef int32_t (*PutToQueueFp)(void *pMgmt, SRpcMsg* pReq);
-typedef int32_t (*GetQueueSizeFp)(void *pMgmt, int32_t vgId, EQueueType qtype);
-typedef int32_t (*SendReqFp)(SMgmtWrapper* pWrapper, const SEpSet* epSet, SRpcMsg* pReq);
-typedef int32_t (*SendMnodeReqFp)(SMgmtWrapper* pWrapper, SRpcMsg* pReq);
-typedef void (*SendRspFp)(SMgmtWrapper* pWrapper, const SRpcMsg* pRsp);
-typedef void (*SendMnodeRecvFp)(SMgmtWrapper* pWrapper, SRpcMsg* pReq, SRpcMsg* pRsp);
-typedef void (*SendRedirectRspFp)(SMgmtWrapper* pWrapper, const SRpcMsg* pRsp, const SEpSet* pNewEpSet);
-typedef void (*RegisterBrokenLinkArgFp)(SMgmtWrapper* pWrapper, SRpcMsg* pMsg);
-typedef void (*ReleaseHandleFp)(SMgmtWrapper* pWrapper, void* handle, int8_t type);
-typedef void (*ReportStartup)(SMgmtWrapper* pWrapper, const char* name, const char* desc);
+typedef int32_t (*PutToQueueFp)(void* pMgmt, SRpcMsg* pMsg);
+typedef int32_t (*GetQueueSizeFp)(void* pMgmt, int32_t vgId, EQueueType qtype);
+typedef int32_t (*SendReqFp)(const SEpSet* pEpSet, SRpcMsg* pMsg);
+typedef void (*SendRspFp)(SRpcMsg* pMsg);
+typedef void (*SendRedirectRspFp)(SRpcMsg* pMsg, const SEpSet* pNewEpSet);
+typedef void (*RegisterBrokenLinkArgFp)(SRpcMsg* pMsg);
+typedef void (*ReleaseHandleFp)(SRpcHandleInfo* pHandle, int8_t type);
+typedef void (*ReportStartup)(const char* name, const char* desc);
 
 typedef struct {
-  SMgmtWrapper*           pWrapper;
-  void*                   pMgmt;
+  void*                   mgmt;
   void*                   clientRpc;
   PutToQueueFp            queueFps[QUEUE_MAX];
   GetQueueSizeFp          qsizeFp;
   SendReqFp               sendReqFp;
   SendRspFp               sendRspFp;
-  SendMnodeRecvFp         sendMnodeRecvFp;
   SendRedirectRspFp       sendRedirectRspFp;
   RegisterBrokenLinkArgFp registerBrokenLinkArgFp;
   ReleaseHandleFp         releaseHandleFp;
   ReportStartup           reportStartupFp;
 } SMsgCb;
 
-void    tmsgSetDefaultMsgCb(const SMsgCb* pMsgCb);
-int32_t tmsgPutToQueue(const SMsgCb* pMsgCb, EQueueType qtype, SRpcMsg* pReq);
-int32_t tmsgGetQueueSize(const SMsgCb* pMsgCb, int32_t vgId, EQueueType qtype);
-int32_t tmsgSendReq(const SMsgCb* pMsgCb, const SEpSet* epSet, SRpcMsg* pReq);
-void    tmsgSendRsp(SRpcMsg* pRsp);
-void    tmsgSendMnodeRecv(SRpcMsg* pReq, SRpcMsg* pRsp);
-void    tmsgSendRedirectRsp(SRpcMsg* pRsp, const SEpSet* pNewEpSet);
-void    tmsgRegisterBrokenLinkArg(const SMsgCb* pMsgCb, SRpcMsg* pMsg);
-void    tmsgReleaseHandle(void* handle, int8_t type);
+void    tmsgSetDefault(const SMsgCb* msgcb);
+int32_t tmsgPutToQueue(const SMsgCb* msgcb, EQueueType qtype, SRpcMsg* pMsg);
+int32_t tmsgGetQueueSize(const SMsgCb* msgcb, int32_t vgId, EQueueType qtype);
+int32_t tmsgSendReq(const SEpSet* epSet, SRpcMsg* pMsg);
+void    tmsgSendRsp(SRpcMsg* pMsg);
+void    tmsgSendRedirectRsp(SRpcMsg* pMsg, const SEpSet* pNewEpSet);
+void    tmsgRegisterBrokenLinkArg(SRpcMsg* pMsg);
+void    tmsgReleaseHandle(SRpcHandleInfo* pHandle, int8_t type);
 void    tmsgReportStartup(const char* name, const char* desc);
 
 #ifdef __cplusplus

include/common/ttypes.h

@@ -179,6 +179,8 @@ typedef struct {
     }                                          \
   } while (0)
 
+//TODO: use varchar(0) to represent NULL type
+#define IS_NULL_TYPE(_t)             ((_t) == TSDB_DATA_TYPE_NULL)
 #define IS_SIGNED_NUMERIC_TYPE(_t)   ((_t) >= TSDB_DATA_TYPE_TINYINT && (_t) <= TSDB_DATA_TYPE_BIGINT)
 #define IS_UNSIGNED_NUMERIC_TYPE(_t) ((_t) >= TSDB_DATA_TYPE_UTINYINT && (_t) <= TSDB_DATA_TYPE_UBIGINT)
 #define IS_FLOAT_TYPE(_t)            ((_t) == TSDB_DATA_TYPE_FLOAT || (_t) == TSDB_DATA_TYPE_DOUBLE)

include/dnode/mgmt/dnode.h

@@ -22,66 +22,28 @@
 extern "C" {
 #endif
 
-/* ------------------------ TYPES EXPOSED ---------------- */
-typedef struct SDnode SDnode;
-
 /**
- * @brief Initialize the environment
+ * @brief Initialize the dnode
  *
+ * @param rtype for internal debug usage, default is 0
  * @return int32_t 0 for success and -1 for failure
  */
-int32_t dmInit();
+int32_t dmInit(int8_t rtype);
 
 /**
- * @brief Clear the environment
+ * @brief Cleanup the dnode
  */
 void dmCleanup();
 
-/* ------------------------ SDnode ----------------------- */
-typedef struct {
-  int32_t   numOfSupportVnodes;
-  uint16_t  serverPort;
-  char      dataDir[PATH_MAX];
-  char      localEp[TSDB_EP_LEN];
-  char      localFqdn[TSDB_FQDN_LEN];
-  char      firstEp[TSDB_EP_LEN];
-  char      secondEp[TSDB_EP_LEN];
-  SDiskCfg *disks;
-  int32_t   numOfDisks;
-  int8_t    ntype;
-} SDnodeOpt;
-
-typedef enum { DND_EVENT_START = 0, DND_EVENT_STOP = 1, DND_EVENT_CHILD = 2 } EDndEvent;
-
 /**
- * @brief Initialize and start the dnode.
- *
- * @param pOption Option of the dnode.
- * @return SDnode* The dnode object.
+ * @brief Run dnode.
  */
-SDnode *dmCreate(const SDnodeOpt *pOption);
+int32_t dmRun();
 
 /**
- * @brief Stop and cleanup the dnode.
- *
- * @param pDnode The dnode object to close.
- */
-void dmClose(SDnode *pDnode);
-
-/**
- * @brief Run dnode until specific event is receive.
- *
- * @param pDnode The dnode object to run.
- */
-int32_t dmRun(SDnode *pDnode);
-
-/**
- * @brief Handle event in the dnode.
- *
- * @param pDnode The dnode object to close.
- * @param event The event to handle.
+ * @brief Stop dnode.
  */
-void dmSetEvent(SDnode *pDnode, EDndEvent event);
+void dmStop();
 
 #ifdef __cplusplus
 }

include/dnode/mnode/mnode.h

@@ -88,7 +88,7 @@ int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad);
  * @param pMsg The request msg.
  * @return int32_t 0 for success, -1 for failure.
  */
-int32_t mndProcessMsg(SNodeMsg *pMsg);
+int32_t mndProcessMsg(SRpcMsg *pMsg);
 
 /**
  * @brief Generate machine code

include/dnode/mnode/sdb/sdb.h

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

include/libs/catalog/catalog.h

@@ -59,6 +59,11 @@ typedef struct SMetaData {
   SArray    *pQnodeList;  // qnode list, SArray<SQueryNodeAddr>
 } SMetaData;
 
+typedef struct STbSVersion {
+  char* tbFName;
+  int32_t sver;
+} STbSVersion;
+
 typedef struct SCatalogCfg {
   uint32_t maxTblCacheNum;
   uint32_t maxDBCacheNum;
@@ -165,6 +170,8 @@ int32_t catalogUpdateSTableMeta(SCatalog* pCatalog, STableMetaRsp *rspMsg);
  */
 int32_t catalogRefreshDBVgInfo(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const char* dbFName);
 
+int32_t catalogChkTbMetaVersion(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, SArray* pTables);
+
 /**
  * Force refresh a table's local cached meta data. 
  * @param pCatalog (input, got with catalogGetHandle)

include/libs/executor/executor.h

@@ -81,7 +81,7 @@ int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numO
  * @param isAdd
  * @return
  */
-int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, SArray* tableIdList, bool isAdd);
+int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bool isAdd);
 
 /**
  * Create the exec task object according to task json
@@ -95,6 +95,15 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, SArray* tableIdList, bool isA
 int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, struct SSubplan* pPlan,
                         qTaskInfo_t* pTaskInfo, DataSinkHandle* handle, EOPTR_EXEC_MODEL model);
 
+/**
+ *
+ * @param tinfo
+ * @param sversion
+ * @param tversion
+ * @return
+ */
+int32_t qGetQueriedTableSchemaVersion(qTaskInfo_t tinfo, char* dbName, char* tableName, int32_t* sversion, int32_t* tversion);
+
 /**
  * The main task execution function, including query on both table and multiple tables,
  * which are decided according to the tag or table name query conditions
@@ -169,7 +178,7 @@ int32_t qUpdateQueriedTableIdList(qTaskInfo_t tinfo, int64_t uid, int32_t type);
 
 void qProcessFetchRsp(void* parent, struct SRpcMsg* pMsg, struct SEpSet* pEpSet);
 
-int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, int32_t *resNum, SExplainExecInfo **pRes);
+int32_t qGetExplainExecInfo(qTaskInfo_t tinfo, int32_t* resNum, SExplainExecInfo** pRes);
 
 #ifdef __cplusplus
 }

include/libs/function/function.h

@@ -174,22 +174,8 @@ typedef struct SqlFunctionCtx {
   SResultDataInfo        resDataInfo;
   uint32_t               order;  // data block scanner order: asc|desc
   uint8_t                scanFlag;  // record current running step, default: 0
-  ////////////////////////////////////////////////////////////////
-  int32_t          startRow;   // start row index
-  int32_t          size;       // handled processed row number
-  SColumnInfoData* pInput;
-  SColumnDataAgg   agg;
-  int16_t          inputType;    // TODO remove it
-  int16_t          inputBytes;   // TODO remove it
-  bool             hasNull;      // null value exist in current block, TODO remove it
-  bool             requireNull;  // require null in some function, TODO remove it
-  int32_t          columnIndex;  // TODO remove it
-  bool             isAggSet;
-  int64_t          startTs;       // timestamp range of current query when function is executed on a specific data block, TODO remove it
-  bool             stableQuery;
-  /////////////////////////////////////////////////////////////////
   int16_t                functionId;    // function id
-  char *           pOutput;       // final result output buffer, point to sdata->data
+  char                  *pOutput;       // final result output buffer, point to sdata->data
   int32_t                numOfParams;
   SFunctParam           *param;         // input parameter, e.g., top(k, 20), the number of results for top query is kept in param
   int64_t               *ptsList;       // corresponding timestamp array list

include/libs/function/functionMgt.h

@@ -142,6 +142,8 @@ void fmFuncMgtDestroy();
 
 int32_t fmGetFuncInfo(SFmGetFuncInfoParam* pParam, SFunctionNode* pFunc);
 
+bool fmIsBuiltinFunc(const char* pFunc);
+
 bool fmIsAggFunc(int32_t funcId);
 bool fmIsScalarFunc(int32_t funcId);
 bool fmIsNonstandardSQLFunc(int32_t funcId);

include/libs/nodes/cmdnodes.h

@@ -78,7 +78,7 @@ typedef struct SAlterDatabaseStmt {
 
 typedef struct STableOptions {
   ENodeType  type;
-  char       comment[TSDB_STB_COMMENT_LEN];
+  char       comment[TSDB_TB_COMMENT_LEN];
   int32_t    delay;
   float      filesFactor;
   SNodeList* pRollupFuncs;
@@ -90,7 +90,7 @@ typedef struct SColumnDefNode {
   ENodeType type;
   char      colName[TSDB_COL_NAME_LEN];
   SDataType dataType;
-  char      comments[TSDB_STB_COMMENT_LEN];
+  char      comments[TSDB_TB_COMMENT_LEN];
   bool      sma;
 } SColumnDefNode;
 

include/libs/nodes/nodes.h

@@ -208,6 +208,7 @@ typedef enum ENodeType {
   QUERY_NODE_PHYSICAL_PLAN_SORT,
   QUERY_NODE_PHYSICAL_PLAN_INTERVAL,
   QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL,
+  QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL,
   QUERY_NODE_PHYSICAL_PLAN_FILL,
   QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW,
   QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW,
@@ -240,6 +241,7 @@ typedef struct SNodeList {
 
 #define SNodeptr void*
 
+int32_t  nodesNodeSize(ENodeType type);
 SNodeptr nodesMakeNode(ENodeType type);
 void     nodesDestroyNode(SNodeptr pNode);
 

include/libs/nodes/querynodes.h

@@ -48,6 +48,7 @@ typedef struct SExprNode {
   ENodeType type;
   SDataType resType;
   char      aliasName[TSDB_COL_NAME_LEN];
+  char      userAlias[TSDB_COL_NAME_LEN];
   SArray*   pAssociation;
 } SExprNode;
 
@@ -247,6 +248,7 @@ typedef struct SSetOperator {
   SNode*           pRight;
   SNodeList*       pOrderByList;  // SOrderByExprNode
   SNode*           pLimit;
+  char             stmtName[TSDB_TABLE_NAME_LEN];
 } SSetOperator;
 
 typedef enum ESqlClause {
@@ -325,7 +327,7 @@ typedef struct SQuery {
   bool           showRewrite;
   int32_t        placeholderNum;
   SArray*        pPlaceholderValues;
-  SNode*         pContainPlaceholderRoot;
+  SNode*         pPrepareRoot;
 } SQuery;
 
 void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker walker, void* pContext);

include/libs/qcom/query.h

@@ -223,19 +223,19 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
 #define qDebug(...)                                               \
   do {                                                            \
     if (qDebugFlag & DEBUG_DEBUG) {                               \
-      taosPrintLog("QRY ", DEBUG_DEBUG, tsLogEmbedded ? 255 : qDebugFlag, __VA_ARGS__); \
+      taosPrintLog("QRY ", DEBUG_DEBUG, qDebugFlag, __VA_ARGS__); \
     }                                                             \
   } while (0)
 #define qTrace(...)                                               \
   do {                                                            \
     if (qDebugFlag & DEBUG_TRACE) {                               \
-      taosPrintLog("QRY ", DEBUG_TRACE, tsLogEmbedded ? 255 : qDebugFlag, __VA_ARGS__); \
+      taosPrintLog("QRY ", DEBUG_TRACE, qDebugFlag, __VA_ARGS__); \
     }                                                             \
   } while (0)
 #define qDebugL(...)                                                     \
   do {                                                                   \
     if (qDebugFlag & DEBUG_DEBUG) {                                      \
-      taosPrintLongString("QRY ", DEBUG_DEBUG, tsLogEmbedded ? 255 : qDebugFlag, __VA_ARGS__); \
+      taosPrintLongString("QRY ", DEBUG_DEBUG, qDebugFlag, __VA_ARGS__); \
     }                                                                    \
   } while (0)
 

include/libs/scheduler/scheduler.h

@@ -72,7 +72,7 @@ int32_t schedulerInit(SSchedulerCfg *cfg);
  * @param nodeList  Qnode/Vnode address list, element is SQueryNodeAddr
  * @return
  */
-int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryPlan *pDag, int64_t *pJob, const char *sql, int64_t startTs, bool needRes, SQueryResult *pRes);
+int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryPlan *pDag, int64_t *pJob, const char *sql, int64_t startTs, SQueryResult *pRes);
 
 /**
  * Process the query job, generated according to the query physical plan.

include/libs/transport/trpc.h

@@ -26,38 +26,41 @@ extern "C" {
 
 #define TAOS_CONN_SERVER 0
 #define TAOS_CONN_CLIENT 1
+#define IsReq(pMsg)      (pMsg->msgType & 1U)
 
 extern int tsRpcHeadSize;
 
-typedef struct SRpcConnInfo {
+typedef struct {
   uint32_t clientIp;
   uint16_t clientPort;
-  uint32_t serverIp;
   char     user[TSDB_USER_LEN];
 } SRpcConnInfo;
 
-typedef struct SRpcMsg {
-  tmsg_t  msgType;
-  void *  pCont;
-  int     contLen;
-  int32_t code;
-  void *  handle;         // rpc handle returned to app
-  void *  ahandle;        // app handle set by client
+typedef struct SRpcHandleInfo {
+  // rpc info
+  void   *handle;         // rpc handle returned to app
   int64_t refId;          // refid, used by server
-  int     noResp;         // has response or not(default 0, 0: resp, 1: no resp);
-  int     persistHandle;  // persist handle or not
+  int32_t noResp;         // has response or not(default 0, 0: resp, 1: no resp);
+  int32_t persistHandle;  // persist handle or not
 
-} SRpcMsg;
+  // app info
+  void *ahandle;  // app handle set by client
+  void *wrapper;  // wrapper handle
+  void *node;     // node mgmt handle
 
-typedef struct {
-  char     user[TSDB_USER_LEN];
-  uint32_t clientIp;
-  uint16_t clientPort;
-  SRpcMsg  rpcMsg;
+  // resp info
+  void   *rsp;
   int32_t rspLen;
-  void *   pRsp;
-  void *   pNode;
-} SNodeMsg;
+} SRpcHandleInfo;
+
+typedef struct SRpcMsg {
+  tmsg_t         msgType;
+  void          *pCont;
+  int32_t        contLen;
+  int32_t        code;
+  SRpcHandleInfo info;
+  SRpcConnInfo   conn;
+} SRpcMsg;
 
 typedef void (*RpcCfp)(void *parent, SRpcMsg *, SEpSet *rf);
 typedef int (*RpcAfp)(void *parent, char *tableId, char *spi, char *encrypt, char *secret, char *ckey);

include/os/osString.h

@@ -38,6 +38,13 @@ typedef int32_t TdUcs4;
     #define wcsncpy WCSNCPY_FUNC_TAOS_FORBID
     #define wchar_t WCHAR_T_TYPE_TAOS_FORBID
     #define strcasestr STR_CASE_STR_FORBID
+    #define strtoll STR_TO_LL_FUNC_TAOS_FORBID
+    #define strtoull STR_TO_ULL_FUNC_TAOS_FORBID
+    #define strtol STR_TO_L_FUNC_TAOS_FORBID
+    #define strtoul STR_TO_UL_FUNC_TAOS_FORBID
+    #define strtod STR_TO_LD_FUNC_TAOS_FORBID
+    #define strtold STR_TO_D_FUNC_TAOS_FORBID
+    #define strtof STR_TO_F_FUNC_TAOS_FORBID
 #endif
 
 #ifdef WINDOWS
@@ -72,6 +79,17 @@ int32_t taosWcharsToMbs(char *pStrs, TdWchar *pWchars, int32_t size);
 
 char *taosStrCaseStr(const char *str, const char *pattern);
 
+int64_t taosStr2Int64(const char *str, char** pEnd, int32_t radix);
+uint64_t taosStr2UInt64(const char *str, char** pEnd, int32_t radix);
+int32_t taosStr2Int32(const char *str, char** pEnd, int32_t radix);
+uint32_t taosStr2UInt32(const char *str, char** pEnd, int32_t radix);
+int16_t taosStr2Int16(const char *str, char** pEnd, int32_t radix);
+uint16_t taosStr2UInt16(const char *str, char** pEnd, int32_t radix);
+int8_t taosStr2Int8(const char *str, char** pEnd, int32_t radix);
+uint8_t taosStr2UInt8(const char *str, char** pEnd, int32_t radix);
+double taosStr2Double(const char *str, char** pEnd);
+float taosStr2Float(const char *str, char** pEnd);
+
 #ifdef __cplusplus
 }
 #endif

include/util/taoserror.h

@@ -646,6 +646,9 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_INVALID_ALTER_TABLE       TAOS_DEF_ERROR_CODE(0, 0x2649)
 #define TSDB_CODE_PAR_CANNOT_DROP_PRIMARY_KEY   TAOS_DEF_ERROR_CODE(0, 0x264A)
 #define TSDB_CODE_PAR_INVALID_MODIFY_COL        TAOS_DEF_ERROR_CODE(0, 0x264B)
+#define TSDB_CODE_PAR_INVALID_TBNAME            TAOS_DEF_ERROR_CODE(0, 0x264C)
+#define TSDB_CODE_PAR_INVALID_FUNCTION_NAME     TAOS_DEF_ERROR_CODE(0, 0x264D)
+#define TSDB_CODE_PAR_COMMENT_TOO_LONG          TAOS_DEF_ERROR_CODE(0, 0x264E)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)
@@ -668,6 +671,8 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_UDF_INVALID_STATE             TAOS_DEF_ERROR_CODE(0, 0x2906)
 #define TSDB_CODE_UDF_INVALID_INPUT             TAOS_DEF_ERROR_CODE(0, 0x2907)
 #define TSDB_CODE_UDF_NO_FUNC_HANDLE            TAOS_DEF_ERROR_CODE(0, 0x2908)
+#define TSDB_CODE_UDF_INVALID_BUFSIZE           TAOS_DEF_ERROR_CODE(0, 0x2909)
+#define TSDB_CODE_UDF_INVALID_OUTPUT_TYPE       TAOS_DEF_ERROR_CODE(0, 0x290A)
 
 #define TSDB_CODE_SML_INVALID_PROTOCOL_TYPE     TAOS_DEF_ERROR_CODE(0, 0x3000)
 #define TSDB_CODE_SML_INVALID_PRECISION_TYPE    TAOS_DEF_ERROR_CODE(0, 0x3001)

include/util/tdef.h

@@ -219,7 +219,7 @@ typedef enum ELogicConditionType {
 #define TSDB_MAX_ALLOWED_SQL_LEN (1 * 1024 * 1024u)  // sql length should be less than 1mb
 
 #define TSDB_APP_NAME_LEN   TSDB_UNI_LEN
-#define TSDB_STB_COMMENT_LEN 1024
+#define TSDB_TB_COMMENT_LEN 1025
 
 /**
  *  In some scenarios uint16_t (0~65535) is used to store the row len.

include/util/tencode.h

@@ -456,52 +456,189 @@ static FORCE_INLINE void* tDecoderMalloc(SDecoder* pCoder, int32_t size) {
   return p;
 }
 
-static FORCE_INLINE int32_t tPutBinary(uint8_t* p, const uint8_t* pData, uint32_t nData) {
-  int      n = 0;
-  uint32_t v = nData;
+// ===========================================
+#define tPutV(p, v)                    \
+  do {                                 \
+    int32_t n = 0;                     \
+    for (;;) {                         \
+      if (v <= 0x7f) {                 \
+        if (p) p[n] = v;               \
+        n++;                           \
+        break;                         \
+      }                                \
+      if (p) p[n] = (v & 0x7f) | 0x80; \
+      n++;                             \
+      v >>= 7;                         \
+    }                                  \
+    return n;                          \
+  } while (0)
 
-  for (;;) {
-    if (v <= 0x7f) {
-      if (p) p[n] = v;
-      n++;
-      break;
-    }
+#define tGetV(p, v)                              \
+  do {                                           \
+    int32_t n = 0;                               \
+    if (v) *v = 0;                               \
+    for (;;) {                                   \
+      if (p[n] <= 0x7f) {                        \
+        if (v) (*v) |= (p[n] << (7 * n));        \
+        n++;                                     \
+        break;                                   \
+      }                                          \
+      if (v) (*v) |= ((p[n] & 0x7f) << (7 * n)); \
+      n++;                                       \
+    }                                            \
+    return n;                                    \
+  } while (0)
 
-    if (p) p[n] = (v & 0x7f) | 0x80;
-    n++;
-    v >>= 7;
-  }
+// PUT
+static FORCE_INLINE int32_t tPutU8(uint8_t* p, uint8_t v) {
+  if (p) ((uint8_t*)p)[0] = v;
+  return sizeof(uint8_t);
+}
 
-  if (p) {
-    memcpy(p + n, pData, nData);
-  }
+static FORCE_INLINE int32_t tPutI8(uint8_t* p, int8_t v) {
+  if (p) ((int8_t*)p)[0] = v;
+  return sizeof(int8_t);
+}
+
+static FORCE_INLINE int32_t tPutU16(uint8_t* p, uint16_t v) {
+  if (p) ((uint16_t*)p)[0] = v;
+  return sizeof(uint16_t);
+}
+
+static FORCE_INLINE int32_t tPutI16(uint8_t* p, int16_t v) {
+  if (p) ((int16_t*)p)[0] = v;
+  return sizeof(int16_t);
+}
+
+static FORCE_INLINE int32_t tPutU32(uint8_t* p, uint32_t v) {
+  if (p) ((uint32_t*)p)[0] = v;
+  return sizeof(uint32_t);
+}
+
+static FORCE_INLINE int32_t tPutI32(uint8_t* p, int32_t v) {
+  if (p) ((int32_t*)p)[0] = v;
+  return sizeof(int32_t);
+}
+
+static FORCE_INLINE int32_t tPutU64(uint8_t* p, uint64_t v) {
+  if (p) ((uint64_t*)p)[0] = v;
+  return sizeof(uint64_t);
+}
+
+static FORCE_INLINE int32_t tPutI64(uint8_t* p, int64_t v) {
+  if (p) ((int64_t*)p)[0] = v;
+  return sizeof(int64_t);
+}
+
+static FORCE_INLINE int32_t tPutU16v(uint8_t* p, uint16_t v) { tPutV(p, v); }
+
+static FORCE_INLINE int32_t tPutI16v(uint8_t* p, int16_t v) { return tPutU16v(p, ZIGZAGE(int16_t, v)); }
+
+static FORCE_INLINE int32_t tPutU32v(uint8_t* p, uint32_t v) { tPutV(p, v); }
+
+static FORCE_INLINE int32_t tPutI32v(uint8_t* p, int32_t v) { return tPutU32v(p, ZIGZAGE(int32_t, v)); }
+
+static FORCE_INLINE int32_t tPutU64v(uint8_t* p, uint64_t v) { tPutV(p, v); }
+
+static FORCE_INLINE int32_t tPutI64v(uint8_t* p, int64_t v) { return tPutU64v(p, ZIGZAGE(int64_t, v)); }
+
+// GET
+static FORCE_INLINE int32_t tGetU8(uint8_t* p, uint8_t* v) {
+  if (v) *v = ((uint8_t*)p)[0];
+  return sizeof(uint8_t);
+}
+
+static FORCE_INLINE int32_t tGetI8(uint8_t* p, int8_t* v) {
+  if (v) *v = ((int8_t*)p)[0];
+  return sizeof(int8_t);
+}
+
+static FORCE_INLINE int32_t tGetU16(uint8_t* p, uint16_t* v) {
+  if (v) *v = ((uint16_t*)p)[0];
+  return sizeof(uint16_t);
+}
+
+static FORCE_INLINE int32_t tGetI16(uint8_t* p, int16_t* v) {
+  if (v) *v = ((int16_t*)p)[0];
+  return sizeof(int16_t);
+}
+
+static FORCE_INLINE int32_t tGetU32(uint8_t* p, uint32_t* v) {
+  if (v) *v = ((uint32_t*)p)[0];
+  return sizeof(uint32_t);
+}
+
+static FORCE_INLINE int32_t tGetI32(uint8_t* p, int32_t* v) {
+  if (v) *v = ((int32_t*)p)[0];
+  return sizeof(int32_t);
+}
+
+static FORCE_INLINE int32_t tGetU64(uint8_t* p, uint64_t* v) {
+  if (v) *v = ((uint64_t*)p)[0];
+  return sizeof(uint64_t);
+}
+
+static FORCE_INLINE int32_t tGetI64(uint8_t* p, int64_t* v) {
+  if (v) *v = ((int64_t*)p)[0];
+  return sizeof(int64_t);
+}
+
+static FORCE_INLINE int32_t tGetU16v(uint8_t* p, uint16_t* v) { tGetV(p, v); }
+
+static FORCE_INLINE int32_t tGetI16v(uint8_t* p, int16_t* v) {
+  int32_t  n;
+  uint16_t tv;
+
+  n = tGetU16v(p, &tv);
+  if (v) *v = ZIGZAGD(int16_t, tv);
+
+  return n;
+}
+
+static FORCE_INLINE int32_t tGetU32v(uint8_t* p, uint32_t* v) { tGetV(p, v); }
+
+static FORCE_INLINE int32_t tGetI32v(uint8_t* p, int32_t* v) {
+  int32_t  n;
+  uint32_t tv;
+
+  n = tGetU32v(p, &tv);
+  if (v) *v = ZIGZAGD(int32_t, tv);
+
+  return n;
+}
+
+static FORCE_INLINE int32_t tGetU64v(uint8_t* p, uint64_t* v) { tGetV(p, v); }
+
+static FORCE_INLINE int32_t tGetI64v(uint8_t* p, int64_t* v) {
+  int32_t  n;
+  uint64_t tv;
+
+  n = tGetU64v(p, &tv);
+  if (v) *v = ZIGZAGD(int64_t, tv);
+
+  return n;
+}
+
+// =====================
+static FORCE_INLINE int32_t tPutBinary(uint8_t* p, uint8_t* pData, uint32_t nData) {
+  int n = 0;
+
+  n += tPutU32v(p ? p + n : p, nData);
+  if (p) memcpy(p + n, pData, nData);
   n += nData;
 
   return n;
 }
 
-static FORCE_INLINE int32_t tGetBinary(const uint8_t* p, const uint8_t** ppData, uint32_t* nData) {
+static FORCE_INLINE int32_t tGetBinary(uint8_t* p, uint8_t** ppData, uint32_t* nData) {
   int32_t  n = 0;
-  uint32_t tv = 0;
-  uint32_t t;
-
-  for (;;) {
-    if (p[n] <= 0x7f) {
-      t = p[n];
-      tv |= (t << (7 * n));
-      n++;
-      break;
-    }
-
-    t = p[n] & 0x7f;
-    tv |= (t << (7 * n));
-    n++;
-  }
+  uint32_t nt;
 
-  if (nData) *nData = n;
+  n += tGetU32v(p, &nt);
+  if (nData) *nData = nt;
   if (ppData) *ppData = p + n;
+  n += nt;
 
-  n += tv;
   return n;
 }
 

include/util/tprocess.h

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
-
-#ifndef _TD_UTIL_PROCESS_H_
-#define _TD_UTIL_PROCESS_H_
-
-#include "os.h"
-#include "tqueue.h"
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-typedef enum { PROC_FUNC_REQ = 1, PROC_FUNC_RSP, PROC_FUNC_REGIST, PROC_FUNC_RELEASE } EProcFuncType;
-
-typedef struct SProcObj SProcObj;
-typedef void *(*ProcMallocFp)(int32_t contLen, EQItype itype);
-typedef void *(*ProcFreeFp)(void *pCont);
-typedef void (*ProcConsumeFp)(void *parent, void *pHead, int16_t headLen, void *pBody, int32_t bodyLen,
-                              EProcFuncType ftype);
-
-typedef struct {
-  ProcConsumeFp childConsumeFp;
-  ProcMallocFp  childMallocHeadFp;
-  ProcFreeFp    childFreeHeadFp;
-  ProcMallocFp  childMallocBodyFp;
-  ProcFreeFp    childFreeBodyFp;
-  ProcConsumeFp parentConsumeFp;
-  ProcMallocFp  parentMallocHeadFp;
-  ProcFreeFp    parentFreeHeadFp;
-  ProcMallocFp  parentMallocBodyFp;
-  ProcFreeFp    parentFreeBodyFp;
-  SShm          shm;
-  void         *parent;
-  const char   *name;
-  bool          isChild;
-} SProcCfg;
-
-SProcObj *taosProcInit(const SProcCfg *pCfg);
-void      taosProcCleanup(SProcObj *pProc);
-int32_t   taosProcRun(SProcObj *pProc);
-void      taosProcStop(SProcObj *pProc);
-
-int32_t taosProcPutToChildQ(SProcObj *pProc, const void *pHead, int16_t headLen, const void *pBody, int32_t bodyLen,
-                            void *handle, int64_t handleRef, EProcFuncType ftype);
-int64_t taosProcRemoveHandle(SProcObj *pProc, void *handle);
-void    taosProcCloseHandles(SProcObj *pProc, void (*HandleFp)(void *handle));
-void    taosProcPutToParentQ(SProcObj *pProc, const void *pHead, int16_t headLen, const void *pBody, int32_t bodyLen,
-                             EProcFuncType ftype);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /*_TD_UTIL_PROCESS_H_*/

include/util/tskiplist.h

@@ -56,10 +56,10 @@ typedef enum { SSkipListPutSuccess = 0, SSkipListPutEarlyStop = 1, SSkipListPutS
 
 typedef struct SSkipList {
   uint32_t           seed;
+  uint16_t           len;
   __compar_fn_t      comparFn;
   __sl_key_fn_t      keyFn;
   TdThreadRwlock    *lock;
-  uint16_t           len;
   uint8_t            maxLevel;
   uint8_t            flags;
   uint8_t            type;  // static info above

packaging/deb/makedeb.sh

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

packaging/release.sh

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

packaging/rpm/tdengine.spec

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

packaging/tools/install.sh

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

packaging/tools/install_client.sh

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

packaging/tools/make_install.bat

+@echo off
+goto %1
+:needAdmin
+mshta vbscript:createobject("shell.application").shellexecute("%~s0",":hasAdmin","","runas",1)(window.close)&goto :eof
+:hasAdmin
+cp -f C:\\TDengine\\driver\\taos.dll C:\\Windows\\System32
\ No newline at end of file

packaging/tools/makearbi.sh

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

packaging/tools/makeclient.sh

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

packaging/tools/makepkg.sh

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

source/client/inc/clientInt.h

@@ -234,6 +234,10 @@ static FORCE_INLINE SReqResultInfo* tmqGetNextResInfo(TAOS_RES* res, bool conver
     if (msg->rsp.withSchema) {
       SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(msg->rsp.blockSchema, msg->resIter);
       setResSchemaInfo(&msg->resInfo, pSW->pSchema, pSW->nCols);
+      taosMemoryFreeClear(msg->resInfo.row);
+      taosMemoryFreeClear(msg->resInfo.pCol);
+      taosMemoryFreeClear(msg->resInfo.length);
+      taosMemoryFreeClear(msg->resInfo.convertBuf);
     }
     setQueryResultFromRsp(&msg->resInfo, pRetrieve, convertUcs4);
     return &msg->resInfo;

source/client/src/clientEnv.c

@@ -83,6 +83,15 @@ void closeTransporter(STscObj *pTscObj) {
   rpcClose(pTscObj->pAppInfo->pTransporter);
 }
 
+static bool clientRpcRfp(int32_t code) {
+  if (code == TSDB_CODE_RPC_REDIRECT) {
+    return true;
+  } else {
+    return false;
+  }
+}
+
+
 // TODO refactor
 void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
   SRpcInit rpcInit;
@@ -91,6 +100,7 @@ void *openTransporter(const char *user, const char *auth, int32_t numOfThread) {
   rpcInit.label = "TSC";
   rpcInit.numOfThreads = numOfThread;
   rpcInit.cfp = processMsgFromServer;
+  rpcInit.rfp = clientRpcRfp;
   rpcInit.sessions = 1024;
   rpcInit.connType = TAOS_CONN_CLIENT;
   rpcInit.user = (char *)user;

source/client/src/clientImpl.c

@@ -291,7 +291,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
 
   SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf};
   int32_t      code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr,
-                                       pRequest->metric.start, NULL != pRes, &res);
+                                       pRequest->metric.start, &res);
   if (code != TSDB_CODE_SUCCESS) {
     if (pRequest->body.queryJob != 0) {
       schedulerFreeJob(pRequest->body.queryJob);
@@ -310,9 +310,7 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
     }
   }
 
-  if (pRes) {
   *pRes = res.res;
-  }
 
   pRequest->code = res.code;
   terrno = res.code;
@@ -324,7 +322,60 @@ int32_t getQueryPlan(SRequestObj* pRequest, SQuery* pQuery, SArray** pNodeList)
   return getPlan(pRequest, pQuery, &pRequest->body.pDag, *pNodeList);
 }
 
+int32_t validateSversion(SRequestObj* pRequest, void* res) {
+  SArray* pArray = NULL;
+  int32_t code = 0;
+  
+  if (TDMT_VND_SUBMIT == pRequest->type) {
+    SSubmitRsp* pRsp = (SSubmitRsp*)res;
+    if (pRsp->nBlocks <= 0) {
+      return TSDB_CODE_SUCCESS;
+    }
+
+    pArray = taosArrayInit(pRsp->nBlocks, sizeof(STbSVersion));
+    if (NULL == pArray) {
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+    
+    for (int32_t i = 0; i < pRsp->nBlocks; ++i) {
+      SSubmitBlkRsp *blk = pRsp->pBlocks + i;
+      STbSVersion tbSver = {.tbFName = blk->tblFName, .sver = blk->sver};
+      taosArrayPush(pArray, &tbSver);
+    }
+  } else if (TDMT_VND_QUERY == pRequest->type) {
+
+  }
+
+  SCatalog* pCatalog = NULL;
+  CHECK_CODE_GOTO(catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog), _return);
+
+  SEpSet epset = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp);
+
+  code = catalogChkTbMetaVersion(pCatalog, pRequest->pTscObj->pAppInfo->pTransporter, &epset, pArray);
+
+_return:
+
+  taosArrayDestroy(pArray);
+
+  return code;
+}
+
+void freeRequestRes(SRequestObj* pRequest, void* res) {
+  if (NULL == res) {
+    return;
+  }
+  
+  if (TDMT_VND_SUBMIT == pRequest->type) {
+    tFreeSSubmitRsp((SSubmitRsp*)res);
+  } else if (TDMT_VND_QUERY == pRequest->type) {
+
+  }
+}
+
 SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, int32_t code, bool keepQuery, void** res) {
+  void* pRes = NULL;
+
   if (TSDB_CODE_SUCCESS == code) {
     switch (pQuery->execMode) {
       case QUERY_EXEC_MODE_LOCAL:
@@ -337,7 +388,10 @@ SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, int32_t code
         SArray* pNodeList = taosArrayInit(4, sizeof(struct SQueryNodeAddr));
         code = getPlan(pRequest, pQuery, &pRequest->body.pDag, pNodeList);
         if (TSDB_CODE_SUCCESS == code) {
-          code = scheduleQuery(pRequest, pRequest->body.pDag, pNodeList, res);
+          code = scheduleQuery(pRequest, pRequest->body.pDag, pNodeList, &pRes);
+          if (NULL != pRes) {
+            code = validateSversion(pRequest, pRes);
+          }
         }
         taosArrayDestroy(pNodeList);
         break;
@@ -356,6 +410,12 @@ SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, int32_t code
 
   if (NULL != pRequest && TSDB_CODE_SUCCESS != code) {
     pRequest->code = terrno;
+    freeRequestRes(pRequest, pRes);
+    pRes = NULL;
+  }
+
+  if (res) {
+    *res = pRes;
   }
 
   return pRequest;
@@ -584,8 +644,8 @@ bool persistConnForSpecificMsg(void* parenct, tmsg_t msgType) {
 }
 
 void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
-  SMsgSendInfo* pSendInfo = (SMsgSendInfo*)pMsg->ahandle;
-  assert(pMsg->ahandle != NULL);
+  SMsgSendInfo* pSendInfo = (SMsgSendInfo*)pMsg->info.ahandle;
+  assert(pMsg->info.ahandle != NULL);
 
   if (pSendInfo->requestObjRefId != 0) {
     SRequestObj* pRequest = (SRequestObj*)taosAcquireRef(clientReqRefPool, pSendInfo->requestObjRefId);
@@ -616,7 +676,7 @@ void processMsgFromServer(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
     taosReleaseRef(clientReqRefPool, pSendInfo->requestObjRefId);
   }
 
-  SDataBuf buf = {.len = pMsg->contLen, .pData = NULL, .handle = pMsg->handle};
+  SDataBuf buf = {.len = pMsg->contLen, .pData = NULL, .handle = pMsg->info.handle};
 
   if (pMsg->contLen > 0) {
     buf.pData = taosMemoryCalloc(1, pMsg->contLen);
@@ -957,7 +1017,7 @@ TSDB_SERVER_STATUS taos_check_server_status(const char* fqdn, int port, char* de
   void*              clientRpc = NULL;
   SServerStatusRsp   statusRsp = {0};
   SEpSet             epSet = {.inUse = 0, .numOfEps = 1};
-  SRpcMsg            rpcMsg = {.ahandle = (void*)0x9526, .msgType = TDMT_DND_SERVER_STATUS};
+  SRpcMsg            rpcMsg = {.info.ahandle = (void*)0x9526, .msgType = TDMT_DND_SERVER_STATUS};
   SRpcMsg            rpcRsp = {0};
   SRpcInit           rpcInit = {0};
   char               pass[TSDB_PASSWORD_LEN + 1] = {0};

source/client/src/clientSml.c

@@ -580,7 +580,7 @@ static bool smlParseNumber(SSmlKv *kvVal, SSmlMsgBuf *msg){
   const char *pVal = kvVal->value;
   int32_t len = kvVal->length;
   char *endptr = NULL;
-  double result = strtod(pVal, &endptr);
+  double result = taosStr2Double(pVal, &endptr);
   if(pVal == endptr){
     smlBuildInvalidDataMsg(msg, "invalid data", pVal);
     return false;
@@ -714,25 +714,31 @@ static bool smlIsNchar(const char *pVal, uint16_t len) {
 
 static int64_t smlGetTimeValue(const char *value, int32_t len, int8_t type) {
   char *endPtr = NULL;
-  double ts = (double)strtoll(value, &endPtr, 10);
+  int64_t tsInt64 = taosStr2Int64(value, &endPtr, 10);
   if(value + len != endPtr){
     return -1;
   }
+  double ts = tsInt64;
   switch (type) {
     case TSDB_TIME_PRECISION_HOURS:
       ts *= (3600 * 1e9);
+      tsInt64 *= (3600 * 1e9);
       break;
     case TSDB_TIME_PRECISION_MINUTES:
       ts *= (60 * 1e9);
+      tsInt64 *= (60 * 1e9);
       break;
     case TSDB_TIME_PRECISION_SECONDS:
       ts *= (1e9);
+      tsInt64 *= (1e9);
       break;
     case TSDB_TIME_PRECISION_MILLI:
       ts *= (1e6);
+      tsInt64 *= (1e6);
       break;
     case TSDB_TIME_PRECISION_MICRO:
       ts *= (1e3);
+      tsInt64 *= (1e3);
       break;
     case TSDB_TIME_PRECISION_NANO:
       break;
@@ -743,7 +749,7 @@ static int64_t smlGetTimeValue(const char *value, int32_t len, int8_t type) {
     return -1;
   }
 
-  return (int64_t)ts;
+  return tsInt64;
 }
 
 static int64_t smlGetTimeNow(int8_t precision) {

source/client/src/clientStmt.c

@@ -218,6 +218,13 @@ int32_t stmtParseSql(STscStmt* pStmt) {
 
   pStmt->bInfo.needParse = false;
 
+  if (pStmt->sql.pQuery->pRoot && 0 == pStmt->sql.type) {
+    pStmt->sql.type = STMT_TYPE_INSERT;
+  } else if (pStmt->sql.pQuery->pPrepareRoot) {
+    pStmt->sql.type = STMT_TYPE_QUERY;
+  }
+
+/*  
   switch (nodeType(pStmt->sql.pQuery->pRoot)) {
     case QUERY_NODE_VNODE_MODIF_STMT:
       if (0 == pStmt->sql.type) {
@@ -231,6 +238,7 @@ int32_t stmtParseSql(STscStmt* pStmt) {
       tscError("not supported stmt type %d", nodeType(pStmt->sql.pQuery->pRoot));
       STMT_ERR_RET(TSDB_CODE_TSC_STMT_CLAUSE_ERROR);
   }
+*/
 
   return TSDB_CODE_SUCCESS;
 }
@@ -252,8 +260,8 @@ int32_t stmtCleanBindInfo(STscStmt* pStmt) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool freeRequest) {
-  if (STMT_TYPE_QUERY != pStmt->sql.type || freeRequest) {
+int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool deepClean) {
+  if (STMT_TYPE_QUERY != pStmt->sql.type || deepClean) {
     taos_free_result(pStmt->exec.pRequest);
     pStmt->exec.pRequest = NULL;
   }
@@ -272,7 +280,11 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool freeRequest) {
       continue;
     }
 
+    if (STMT_TYPE_MULTI_INSERT == pStmt->sql.type) {
       qFreeStmtDataBlock(pBlocks);
+    } else {
+      qDestroyStmtDataBlock(pBlocks);
+    }
     taosHashRemove(pStmt->exec.pBlockHash, key, keyLen);
 
     pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
@@ -312,11 +324,11 @@ int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
   taosHashCleanup(pStmt->sql.pTableCache);
   pStmt->sql.pTableCache = NULL;
 
-  memset(&pStmt->sql, 0, sizeof(pStmt->sql));
-
   STMT_ERR_RET(stmtCleanExecInfo(pStmt, false, true));
   STMT_ERR_RET(stmtCleanBindInfo(pStmt));
 
+  memset(&pStmt->sql, 0, sizeof(pStmt->sql));
+
   return TSDB_CODE_SUCCESS;
 }
 

source/client/src/tmq.c

@@ -182,10 +182,14 @@ typedef struct {
 
 typedef struct {
   tmq_t*         tmq;
-  int32_t        async;
+  int8_t         async;
+  int8_t         automatic;
+  int8_t         freeOffsets;
+  tmq_commit_cb* userCb;
   tsem_t         rspSem;
   tmq_resp_err_t rspErr;
   SArray*        offsets;
+  void*          userParam;
 } SMqCommitCbParam;
 
 tmq_conf_t* tmq_conf_new() {
@@ -314,6 +318,142 @@ static int32_t tmqMakeTopicVgKey(char* dst, const char* topicName, int32_t vg) {
   return sprintf(dst, "%s:%d", topicName, vg);
 }
 
+int32_t tmqCommitCb(void* param, const SDataBuf* pMsg, int32_t code) {
+  SMqCommitCbParam* pParam = (SMqCommitCbParam*)param;
+  pParam->rspErr = code == 0 ? TMQ_RESP_ERR__SUCCESS : TMQ_RESP_ERR__FAIL;
+  if (pParam->async) {
+    if (pParam->automatic && pParam->tmq->commitCb) {
+      pParam->tmq->commitCb(pParam->tmq, pParam->rspErr, (tmq_topic_vgroup_list_t*)pParam->offsets,
+                            pParam->tmq->commitCbUserParam);
+    } else if (!pParam->automatic && pParam->userCb) {
+      pParam->userCb(pParam->tmq, pParam->rspErr, (tmq_topic_vgroup_list_t*)pParam->offsets, pParam->userParam);
+    }
+
+    if (pParam->freeOffsets) {
+      taosArrayDestroy(pParam->offsets);
+    }
+
+    taosMemoryFree(pParam);
+  } else {
+    tsem_post(&pParam->rspSem);
+  }
+  return 0;
+}
+
+int32_t tmqCommitInner(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, int8_t automatic, int8_t async,
+                       tmq_commit_cb* userCb, void* userParam) {
+  SMqCMCommitOffsetReq req;
+  SArray*              pOffsets = NULL;
+  void*                buf = NULL;
+  SMqCommitCbParam*    pParam = NULL;
+  SMsgSendInfo*        sendInfo = NULL;
+  int8_t               freeOffsets;
+  int32_t              code = -1;
+
+  if (offsets == NULL) {
+    freeOffsets = 1;
+    pOffsets = taosArrayInit(0, sizeof(SMqOffset));
+    for (int32_t i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
+      SMqClientTopic* pTopic = taosArrayGet(tmq->clientTopics, i);
+      for (int32_t j = 0; j < taosArrayGetSize(pTopic->vgs); j++) {
+        SMqClientVg* pVg = taosArrayGet(pTopic->vgs, j);
+        SMqOffset    offset;
+        tstrncpy(offset.topicName, pTopic->topicName, TSDB_TOPIC_FNAME_LEN);
+        tstrncpy(offset.cgroup, tmq->groupId, TSDB_CGROUP_LEN);
+        offset.vgId = pVg->vgId;
+        offset.offset = pVg->currentOffset;
+        taosArrayPush(pOffsets, &offset);
+      }
+    }
+  } else {
+    freeOffsets = 0;
+    pOffsets = (SArray*)&offsets->container;
+  }
+
+  req.num = (int32_t)pOffsets->size;
+  req.offsets = pOffsets->pData;
+
+  SEncoder encoder;
+
+  tEncoderInit(&encoder, NULL, 0);
+  code = tEncodeSMqCMCommitOffsetReq(&encoder, &req);
+  if (code < 0) {
+    goto END;
+  }
+  int32_t tlen = encoder.pos;
+  buf = taosMemoryMalloc(tlen);
+  if (buf == NULL) {
+    tEncoderClear(&encoder);
+    goto END;
+  }
+  tEncoderClear(&encoder);
+
+  tEncoderInit(&encoder, buf, tlen);
+  tEncodeSMqCMCommitOffsetReq(&encoder, &req);
+  tEncoderClear(&encoder);
+
+  pParam = taosMemoryCalloc(1, sizeof(SMqCommitCbParam));
+  if (pParam == NULL) {
+    goto END;
+  }
+  pParam->tmq = tmq;
+  pParam->automatic = automatic;
+  pParam->async = async;
+  pParam->offsets = pOffsets;
+  pParam->freeOffsets = freeOffsets;
+  pParam->userCb = userCb;
+  pParam->userParam = userParam;
+  if (!async) tsem_init(&pParam->rspSem, 0, 0);
+
+  sendInfo = taosMemoryMalloc(sizeof(SMsgSendInfo));
+  if (sendInfo == NULL) goto END;
+  sendInfo->msgInfo = (SDataBuf){
+      .pData = buf,
+      .len = tlen,
+      .handle = NULL,
+  };
+
+  sendInfo->requestId = generateRequestId();
+  sendInfo->requestObjRefId = 0;
+  sendInfo->param = pParam;
+  sendInfo->fp = tmqCommitCb;
+  sendInfo->msgType = TDMT_MND_MQ_COMMIT_OFFSET;
+
+  SEpSet epSet = getEpSet_s(&tmq->pTscObj->pAppInfo->mgmtEp);
+
+  int64_t transporterId = 0;
+  asyncSendMsgToServer(tmq->pTscObj->pAppInfo->pTransporter, &epSet, &transporterId, sendInfo);
+
+  if (!async) {
+    tsem_wait(&pParam->rspSem);
+    code = pParam->rspErr;
+    tsem_destroy(&pParam->rspSem);
+    taosMemoryFree(pParam);
+  }
+
+  // avoid double free if msg is sent
+  buf = NULL;
+
+  code = 0;
+END:
+  if (buf) taosMemoryFree(buf);
+  /*if (pParam) taosMemoryFree(pParam);*/
+  /*if (sendInfo) taosMemoryFree(sendInfo);*/
+
+  if (code != 0 && async) {
+    if (automatic) {
+      tmq->commitCb(tmq, TMQ_RESP_ERR__FAIL, (tmq_topic_vgroup_list_t*)pOffsets, tmq->commitCbUserParam);
+    } else {
+      userCb(tmq, TMQ_RESP_ERR__FAIL, (tmq_topic_vgroup_list_t*)pOffsets, userParam);
+    }
+  }
+
+  if (!async && freeOffsets) {
+    taosArrayDestroy(pOffsets);
+  }
+  return code;
+}
+
 void tmqAssignDelayedHbTask(void* param, void* tmrId) {
   tmq_t*  tmq = (tmq_t*)param;
   int8_t* pTaskType = taosAllocateQitem(sizeof(int8_t), DEF_QITEM);
@@ -350,7 +490,8 @@ int32_t tmqHandleAllDelayedTask(tmq_t* tmq) {
       tmqAskEp(tmq, true);
       taosTmrReset(tmqAssignDelayedHbTask, 1000, tmq, tmqMgmt.timer, &tmq->hbTimer);
     } else if (*pTaskType == TMQ_DELAYED_TASK__COMMIT) {
-      tmq_commit(tmq, NULL, true);
+      /*tmq_commit(tmq, NULL, true);*/
+      tmqCommitInner(tmq, NULL, 1, 1, tmq->commitCb, tmq->commitCbUserParam);
       taosTmrReset(tmqAssignDelayedCommitTask, tmq->autoCommitInterval, tmq, tmqMgmt.timer, &tmq->commitTimer);
     } else if (*pTaskType == TMQ_DELAYED_TASK__REPORT) {
     } else {
@@ -385,29 +526,8 @@ void tmqClearUnhandleMsg(tmq_t* tmq) {
 int32_t tmqSubscribeCb(void* param, const SDataBuf* pMsg, int32_t code) {
   SMqSubscribeCbParam* pParam = (SMqSubscribeCbParam*)param;
   pParam->rspErr = code;
-  tmq_t* tmq = pParam->tmq;
-  tsem_post(&pParam->rspSem);
-  return 0;
-}
-
-int32_t tmqCommitCb(void* param, const SDataBuf* pMsg, int32_t code) {
-  SMqCommitCbParam* pParam = (SMqCommitCbParam*)param;
-  pParam->rspErr = code == 0 ? TMQ_RESP_ERR__SUCCESS : TMQ_RESP_ERR__FAIL;
-  if (pParam->tmq->commitCb) {
-    pParam->tmq->commitCb(pParam->tmq, pParam->rspErr, NULL, pParam->tmq->commitCbUserParam);
-  }
-  if (!pParam->async)
+  /*tmq_t* tmq = pParam->tmq;*/
   tsem_post(&pParam->rspSem);
-  else {
-    if (pParam->offsets) {
-      taosArrayDestroy(pParam->offsets);
-    }
-    tsem_destroy(&pParam->rspSem);
-    /*if (pParam->pArray) {*/
-    /*taosArrayDestroy(pParam->pArray);*/
-    /*}*/
-    taosMemoryFree(pParam);
-  }
   return 0;
 }
 
@@ -541,6 +661,8 @@ FAIL:
 }
 
 tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, int32_t async) {
+  return tmqCommitInner(tmq, offsets, 0, async, tmq->commitCb, tmq->commitCbUserParam);
+#if 0
   // TODO: add read write lock
   SRequestObj*   pRequest = NULL;
   tmq_resp_err_t resp = TMQ_RESP_ERR__SUCCESS;
@@ -627,6 +749,7 @@ tmq_resp_err_t tmq_commit(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, in
   }
 
   return resp;
+#endif
 }
 
 tmq_resp_err_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
@@ -707,10 +830,12 @@ tmq_resp_err_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
   }
 
   // init hb timer
+  if (tmq->hbTimer == NULL) {
     tmq->hbTimer = taosTmrStart(tmqAssignDelayedHbTask, 1000, tmq, tmqMgmt.timer);
+  }
 
   // init auto commit timer
-  if (tmq->autoCommit) {
+  if (tmq->autoCommit && tmq->commitTimer == NULL) {
     tmq->commitTimer = taosTmrStart(tmqAssignDelayedCommitTask, tmq->autoCommitInterval, tmq, tmqMgmt.timer);
   }
 
@@ -723,7 +848,7 @@ FAIL:
   return code;
 }
 
-void tmq_conf_set_offset_commit_cb(tmq_conf_t* conf, tmq_commit_cb* cb, void* param) {
+void tmq_conf_set_auto_commit_cb(tmq_conf_t* conf, tmq_commit_cb* cb, void* param) {
   //
   conf->commitCb = cb;
   conf->commitCbUserParam = param;
@@ -1310,7 +1435,7 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t wait_time) {
 
   while (1) {
     tmqHandleAllDelayedTask(tmq);
-    tmqPollImpl(tmq, wait_time);
+    if (tmqPollImpl(tmq, wait_time) < 0) return NULL;
 
     rspObj = tmqHandleAllRsp(tmq, wait_time, false);
     if (rspObj) {
@@ -1333,9 +1458,18 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t wait_time) {
 
 tmq_resp_err_t tmq_consumer_close(tmq_t* tmq) {
   if (tmq->status == TMQ_CONSUMER_STATUS__READY) {
+    tmq_resp_err_t rsp = tmq_commit_sync(tmq, NULL);
+    if (rsp == TMQ_RESP_ERR__SUCCESS) {
+      // TODO: free resources
+      return TMQ_RESP_ERR__SUCCESS;
+    } else {
+      return TMQ_RESP_ERR__FAIL;
+    }
+
     tmq_list_t* lst = tmq_list_new();
-    tmq_resp_err_t rsp = tmq_subscribe(tmq, lst);
+    rsp = tmq_subscribe(tmq, lst);
     tmq_list_destroy(lst);
+
     if (rsp == TMQ_RESP_ERR__SUCCESS) {
       // TODO: free resources
       return TMQ_RESP_ERR__SUCCESS;
@@ -1384,3 +1518,10 @@ const char* tmq_get_table_name(TAOS_RES* res) {
   }
   return NULL;
 }
+DLL_EXPORT void tmq_commit_async(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets, tmq_commit_cb* cb, void* param) {
+  tmqCommitInner(tmq, offsets, 0, 1, cb, param);
+}
+
+DLL_EXPORT tmq_resp_err_t tmq_commit_sync(tmq_t* tmq, const tmq_topic_vgroup_list_t* offsets) {
+  return tmqCommitInner(tmq, offsets, 0, 0, NULL, NULL);
+}

source/client/test/smlTest.cpp

@@ -1188,3 +1188,36 @@ TEST(testCase, smlParseTelnetLine_diff_json_type2_Test) {
   destroyRequest(request);
   smlDestroyInfo(info);
 }
+
+TEST(testCase, sml_TD15662_Test) {
+  TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  ASSERT_NE(taos, nullptr);
+
+  TAOS_RES *pRes = taos_query(taos, "create database if not exists db_15662 precision 'ns'");
+  taos_free_result(pRes);
+
+  pRes = taos_query(taos, "use db_15662");
+  taos_free_result(pRes);
+
+  SRequestObj *request = (SRequestObj *)createRequest((STscObj *)taos, NULL, NULL, TSDB_SQL_INSERT);
+  ASSERT_NE(request, nullptr);
+
+  SSmlHandle *info = smlBuildSmlInfo(taos, request, TSDB_SML_LINE_PROTOCOL, TSDB_SML_TIMESTAMP_NANO_SECONDS);
+  ASSERT_NE(info, nullptr);
+
+  const char *sql[] = {
+      "iyyyje,id=iyyyje_41943_1303,t0=t,t1=127i8,t2=32767i16,t3=2147483647i32,t4=9223372036854775807i64,t5=11.12345f32,t6=22.123456789f64,t7=\"binaryTagValue\",t8=L\"ncharTagValue\" c0=false,c1=127i8,c2=32767i16,c3=2147483647i32,c4=9223372036854775807i64,c5=11.12345f32,c6=22.123456789f64,c7=\"binaryColValue\",c8=L\"ncharColValue\",c9=7u64 1626006833639000000",
+  };
+  int ret = smlProcess(info, (char **)sql, sizeof(sql) / sizeof(sql[0]));
+  ASSERT_EQ(ret, 0);
+
+  // case 1
+  TAOS_RES *res = taos_query(taos, "select * from t_a5615048edae55218a22a149edebdc82");
+  ASSERT_NE(res, nullptr);
+
+  TAOS_ROW row = taos_fetch_row(res);
+  int64_t ts = *(int64_t*)row[0];
+  ASSERT_EQ(ts, 1626006833639000000);
+
+  taos_free_result(res);
+}
\ No newline at end of file

source/common/src/systable.c

@@ -263,7 +263,7 @@ static const SSysDbTableSchema topicSchema[] = {
 static const SSysDbTableSchema consumerSchema[] = {
     {.name = "consumer_id", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
     {.name = "consumer_group", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
-    {.name = "app_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "status", .bytes = 20 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "topics", .bytes = TSDB_TOPIC_FNAME_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT},

source/common/src/tdatablock.c

@@ -600,10 +600,11 @@ int32_t blockDataFromBuf(SSDataBlock* pBlock, const char* buf) {
 }
 
 int32_t blockDataFromBuf1(SSDataBlock* pBlock, const char* buf, size_t capacity) {
-  pBlock->info.rows = *(int32_t*)buf;
+  pBlock->info.rows = *(int32_t*) buf;
+  pBlock->info.groupId = *(uint64_t*) (buf + sizeof(int32_t));
 
   int32_t     numOfCols = pBlock->info.numOfCols;
-  const char* pStart = buf + sizeof(uint32_t);
+  const char* pStart = buf + sizeof(uint32_t) + sizeof(uint64_t);
 
   for (int32_t i = 0; i < numOfCols; ++i) {
     SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, i);
@@ -1224,7 +1225,27 @@ SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock, bool copyData) {
 }
 
 size_t blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize) {
-  return (int32_t)((pageSize - blockDataGetSerialMetaSize(pBlock)) / blockDataGetSerialRowSize(pBlock));
+  int32_t payloadSize = pageSize - blockDataGetSerialMetaSize(pBlock);
+
+  int32_t rowSize = pBlock->info.rowSize;
+
+  int32_t nRows = payloadSize / rowSize;
+
+  // the true value must be less than the value of nRows
+  int32_t additional = 0;
+  for(int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
+    SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, i);
+    if (IS_VAR_DATA_TYPE(pCol->info.type)) {
+      additional += nRows * sizeof(int32_t);
+    } else {
+      additional += BitmapLen(nRows);
+    }
+  }
+
+  int32_t newRows = (payloadSize - additional) / rowSize;
+  ASSERT(newRows <= nRows && newRows > 1);
+
+  return newRows;
 }
 
 void colDataDestroy(SColumnInfoData* pColData) {

source/common/src/tglobal.c

@@ -30,6 +30,7 @@ char     tsLocalEp[TSDB_EP_LEN] = {0};  // Local End Point, hostname:port
 uint16_t tsServerPort = 6030;
 int32_t  tsVersion = 30000000;
 int32_t  tsStatusInterval = 1;  // second
+int32_t  tsNumOfSupportVnodes = 256;
 
 // common
 int32_t tsMaxShellConns = 50000;
@@ -38,7 +39,7 @@ bool    tsEnableSlaveQuery = true;
 bool    tsPrintAuth = false;
 
 // multi process
-bool    tsMultiProcess = false;
+int32_t tsMultiProcess = 0;
 int32_t tsMnodeShmSize = TSDB_MAX_WAL_SIZE * 2 + 128;
 int32_t tsVnodeShmSize = TSDB_MAX_WAL_SIZE * 10 + 128;
 int32_t tsQnodeShmSize = TSDB_MAX_WAL_SIZE * 4 + 128;
@@ -362,7 +363,7 @@ static int32_t taosAddSystemCfg(SConfig *pCfg) {
 static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddDir(pCfg, "dataDir", tsDataDir, 0) != 0) return -1;
   if (cfgAddFloat(pCfg, "minimalDataDirGB", 2.0f, 0.001f, 10000000, 0) != 0) return -1;
-  if (cfgAddInt32(pCfg, "supportVnodes", 256, 0, 4096, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "supportVnodes", tsNumOfSupportVnodes, 0, 4096, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "maxShellConns", tsMaxShellConns, 10, 50000000, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "statusInterval", tsStatusInterval, 1, 30, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "minSlidingTime", tsMinSlidingTime, 10, 1000000, 0) != 0) return -1;
@@ -378,7 +379,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddBool(pCfg, "slaveQuery", tsEnableSlaveQuery, 0) != 0) return -1;
   if (cfgAddBool(pCfg, "deadLockKillQuery", tsDeadLockKillQuery, 0) != 0) return -1;
 
-  if (cfgAddBool(pCfg, "multiProcess", tsMultiProcess, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "multiProcess", tsMultiProcess, 0, 2, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "mnodeShmSize", tsMnodeShmSize, TSDB_MAX_WAL_SIZE + 128, INT32_MAX, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "vnodeShmSize", tsVnodeShmSize, TSDB_MAX_WAL_SIZE + 128, INT32_MAX, 0) != 0) return -1;
   if (cfgAddInt32(pCfg, "qnodeShmSize", tsQnodeShmSize, TSDB_MAX_WAL_SIZE + 128, INT32_MAX, 0) != 0) return -1;
@@ -552,6 +553,7 @@ static void taosSetSystemCfg(SConfig *pCfg) {
 
 static int32_t taosSetServerCfg(SConfig *pCfg) {
   tsDataSpace.reserved = cfgGetItem(pCfg, "minimalDataDirGB")->fval;
+  tsNumOfSupportVnodes = cfgGetItem(pCfg, "supportVnodes")->i32;
   tsMaxShellConns = cfgGetItem(pCfg, "maxShellConns")->i32;
   tsStatusInterval = cfgGetItem(pCfg, "statusInterval")->i32;
   tsMinSlidingTime = cfgGetItem(pCfg, "minSlidingTime")->i32;
@@ -565,7 +567,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
   tsRetrieveBlockingModel = cfgGetItem(pCfg, "retrieveBlockingModel")->bval;
   tsPrintAuth = cfgGetItem(pCfg, "printAuth")->bval;
   tsEnableSlaveQuery = cfgGetItem(pCfg, "slaveQuery")->bval;
-  tsDeadLockKillQuery = cfgGetItem(pCfg, "deadLockKillQuery")->bval;
+  tsDeadLockKillQuery = cfgGetItem(pCfg, "deadLockKillQuery")->i32;
 
   tsMultiProcess = cfgGetItem(pCfg, "multiProcess")->bval;
   tsMnodeShmSize = cfgGetItem(pCfg, "mnodeShmSize")->i32;

source/common/src/tmsg.c

@@ -600,6 +600,7 @@ int32_t tSerializeSMAlterStbReq(void *buf, int32_t bufLen, SMAlterStbReq *pReq)
   if (tStartEncode(&encoder) < 0) return -1;
   if (tEncodeCStr(&encoder, pReq->name) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->alterType) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->verInBlock) < 0) return -1;
   if (tEncodeI32(&encoder, pReq->numOfFields) < 0) return -1;
   for (int32_t i = 0; i < pReq->numOfFields; ++i) {
     SField *pField = taosArrayGet(pReq->pFields, i);
@@ -626,6 +627,7 @@ int32_t tDeserializeSMAlterStbReq(void *buf, int32_t bufLen, SMAlterStbReq *pReq
   if (tStartDecode(&decoder) < 0) return -1;
   if (tDecodeCStrTo(&decoder, pReq->name) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->alterType) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->verInBlock) < 0) return -1;
   if (tDecodeI32(&decoder, &pReq->numOfFields) < 0) return -1;
   pReq->pFields = taosArrayInit(pReq->numOfFields, sizeof(SField));
   if (pReq->pFields == NULL) {
@@ -2625,6 +2627,35 @@ int32_t tDeserializeSMDropTopicReq(void *buf, int32_t bufLen, SMDropTopicReq *pR
   return 0;
 }
 
+int32_t tSerializeSMDropCgroupReq(void *buf, int32_t bufLen, SMDropCgroupReq *pReq) {
+  SEncoder encoder = {0};
+  tEncoderInit(&encoder, buf, bufLen);
+
+  if (tStartEncode(&encoder) < 0) return -1;
+  if (tEncodeCStr(&encoder, pReq->topic) < 0) return -1;
+  if (tEncodeCStr(&encoder, pReq->cgroup) < 0) return -1;
+  if (tEncodeI8(&encoder, pReq->igNotExists) < 0) return -1;
+  tEndEncode(&encoder);
+
+  int32_t tlen = encoder.pos;
+  tEncoderClear(&encoder);
+  return tlen;
+}
+
+int32_t tDeserializeSMDropCgroupReq(void *buf, int32_t bufLen, SMDropCgroupReq *pReq) {
+  SDecoder decoder = {0};
+  tDecoderInit(&decoder, buf, bufLen);
+
+  if (tStartDecode(&decoder) < 0) return -1;
+  if (tDecodeCStrTo(&decoder, pReq->topic) < 0) return -1;
+  if (tDecodeCStrTo(&decoder, pReq->cgroup) < 0) return -1;
+  if (tDecodeI8(&decoder, &pReq->igNotExists) < 0) return -1;
+  tEndDecode(&decoder);
+
+  tDecoderClear(&decoder);
+  return 0;
+}
+
 int32_t tSerializeSCMCreateTopicReq(void *buf, int32_t bufLen, const SCMCreateTopicReq *pReq) {
   int32_t sqlLen = 0;
   int32_t astLen = 0;
@@ -4087,12 +4118,11 @@ static int32_t tEncodeSSubmitBlkRsp(SEncoder *pEncoder, const SSubmitBlkRsp *pBl
 
   if (tEncodeI32(pEncoder, pBlock->code) < 0) return -1;
   if (tEncodeI8(pEncoder, pBlock->hashMeta) < 0) return -1;
-  if (pBlock->hashMeta) {
   if (tEncodeI64(pEncoder, pBlock->uid) < 0) return -1;
   if (tEncodeCStr(pEncoder, pBlock->tblFName) < 0) return -1;
-  }
   if (tEncodeI32v(pEncoder, pBlock->numOfRows) < 0) return -1;
   if (tEncodeI32v(pEncoder, pBlock->affectedRows) < 0) return -1;
+  if (tEncodeI64v(pEncoder, pBlock->sver) < 0) return -1;
 
   tEndEncode(pEncoder);
   return 0;
@@ -4103,14 +4133,13 @@ static int32_t tDecodeSSubmitBlkRsp(SDecoder *pDecoder, SSubmitBlkRsp *pBlock) {
 
   if (tDecodeI32(pDecoder, &pBlock->code) < 0) return -1;
   if (tDecodeI8(pDecoder, &pBlock->hashMeta) < 0) return -1;
-  if (pBlock->hashMeta) {
   if (tDecodeI64(pDecoder, &pBlock->uid) < 0) return -1;
   pBlock->tblFName = taosMemoryCalloc(TSDB_TABLE_FNAME_LEN, 1);
   if (NULL == pBlock->tblFName) return -1;
   if (tDecodeCStrTo(pDecoder, pBlock->tblFName) < 0) return -1;
-  }
   if (tDecodeI32v(pDecoder, &pBlock->numOfRows) < 0) return -1;
   if (tDecodeI32v(pDecoder, &pBlock->affectedRows) < 0) return -1;
+  if (tDecodeI64v(pDecoder, &pBlock->sver) < 0) return -1;
 
   tEndDecode(pDecoder);
   return 0;

source/common/src/tmsgcb.c

@@ -17,94 +17,46 @@
 #include "tmsgcb.h"
 #include "taoserror.h"
 
-static SMsgCb tsDefaultMsgCb;
+static SMsgCb defaultMsgCb;
 
-void tmsgSetDefaultMsgCb(const SMsgCb* pMsgCb) {
-  // if (tsDefaultMsgCb.pWrapper == NULL) {
-  tsDefaultMsgCb = *pMsgCb;
-  //}
-}
-
-int32_t tmsgPutToQueue(const SMsgCb* pMsgCb, EQueueType qtype, SRpcMsg* pReq) {
-  PutToQueueFp fp = pMsgCb->queueFps[qtype];
-  if (fp != NULL) {
-    return (*fp)(pMsgCb->pMgmt, pReq);
-  } else {
-    terrno = TSDB_CODE_INVALID_PTR;
-    return -1;
-  }
-}
+void tmsgSetDefault(const SMsgCb* msgcb) { defaultMsgCb = *msgcb; }
 
-int32_t tmsgGetQueueSize(const SMsgCb* pMsgCb, int32_t vgId, EQueueType qtype) {
-  GetQueueSizeFp fp = pMsgCb->qsizeFp;
-  if (fp != NULL) {
-    return (*fp)(pMsgCb->pMgmt, vgId, qtype);
-  } else {
-    terrno = TSDB_CODE_INVALID_PTR;
-    return -1;
-  }
+int32_t tmsgPutToQueue(const SMsgCb* msgcb, EQueueType qtype, SRpcMsg* pMsg) {
+  PutToQueueFp fp = msgcb->queueFps[qtype];
+  return (*fp)(msgcb->mgmt, pMsg);
 }
 
-int32_t tmsgSendReq(const SMsgCb* pMsgCb, const SEpSet* epSet, SRpcMsg* pReq) {
-  SendReqFp fp = pMsgCb->sendReqFp;
-  if (fp != NULL) {
-    return (*fp)(pMsgCb->pWrapper, epSet, pReq);
-  } else {
-    terrno = TSDB_CODE_INVALID_PTR;
-    return -1;
-  }
+int32_t tmsgGetQueueSize(const SMsgCb* msgcb, int32_t vgId, EQueueType qtype) {
+  GetQueueSizeFp fp = msgcb->qsizeFp;
+  return (*fp)(msgcb->mgmt, vgId, qtype);
 }
 
-void tmsgSendRsp(SRpcMsg* pRsp) {
-  SendRspFp fp = tsDefaultMsgCb.sendRspFp;
-  if (fp != NULL) {
-    return (*fp)(tsDefaultMsgCb.pWrapper, pRsp);
-  } else {
-    terrno = TSDB_CODE_INVALID_PTR;
-  }
+int32_t tmsgSendReq(const SEpSet* epSet, SRpcMsg* pMsg) {
+  SendReqFp fp = defaultMsgCb.sendReqFp;
+  return (*fp)(epSet, pMsg);
 }
 
-void tmsgSendRedirectRsp(SRpcMsg* pRsp, const SEpSet* pNewEpSet) {
-  SendRedirectRspFp fp = tsDefaultMsgCb.sendRedirectRspFp;
-  if (fp != NULL) {
-    (*fp)(tsDefaultMsgCb.pWrapper, pRsp, pNewEpSet);
-  } else {
-    terrno = TSDB_CODE_INVALID_PTR;
-  }
+void tmsgSendRsp(SRpcMsg* pMsg) {
+  SendRspFp fp = defaultMsgCb.sendRspFp;
+  return (*fp)(pMsg);
 }
 
-void tmsgSendMnodeRecv(SRpcMsg* pReq, SRpcMsg* pRsp) {
-  SendMnodeRecvFp fp = tsDefaultMsgCb.sendMnodeRecvFp;
-  if (fp != NULL) {
-    (*fp)(tsDefaultMsgCb.pWrapper, pReq, pRsp);
-  } else {
-    terrno = TSDB_CODE_INVALID_PTR;
-  }
+void tmsgSendRedirectRsp(SRpcMsg* pMsg, const SEpSet* pNewEpSet) {
+  SendRedirectRspFp fp = defaultMsgCb.sendRedirectRspFp;
+  (*fp)(pMsg, pNewEpSet);
 }
 
-void tmsgRegisterBrokenLinkArg(const SMsgCb* pMsgCb, SRpcMsg* pMsg) {
-  RegisterBrokenLinkArgFp fp = pMsgCb->registerBrokenLinkArgFp;
-  if (fp != NULL) {
-    (*fp)(pMsgCb->pWrapper, pMsg);
-  } else {
-    terrno = TSDB_CODE_INVALID_PTR;
-  }
+void tmsgRegisterBrokenLinkArg(SRpcMsg* pMsg) {
+  RegisterBrokenLinkArgFp fp = defaultMsgCb.registerBrokenLinkArgFp;
+  (*fp)(pMsg);
 }
 
-void tmsgReleaseHandle(void* handle, int8_t type) {
-  ReleaseHandleFp fp = tsDefaultMsgCb.releaseHandleFp;
-  if (fp != NULL) {
-    (*fp)(tsDefaultMsgCb.pWrapper, handle, type);
-  } else {
-    terrno = TSDB_CODE_INVALID_PTR;
-  }
+void tmsgReleaseHandle(SRpcHandleInfo* pHandle, int8_t type) {
+  ReleaseHandleFp fp = defaultMsgCb.releaseHandleFp;
+  (*fp)(pHandle, type);
 }
 
 void tmsgReportStartup(const char* name, const char* desc) {
-  ReportStartup fp = tsDefaultMsgCb.reportStartupFp;
-  if (fp != NULL && tsDefaultMsgCb.pWrapper != NULL) {
-    (*fp)(tsDefaultMsgCb.pWrapper, name, desc);
-  } else {
-    terrno = TSDB_CODE_INVALID_PTR;
-  }
+  ReportStartup fp = defaultMsgCb.reportStartupFp;
+  (*fp)(name, desc);
 }
\ No newline at end of file

source/common/src/tname.c

@@ -250,7 +250,7 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
       return -1;
     }
 
-    dst->acctId = strtoll(str, NULL, 10);
+    dst->acctId = taosStr2Int32(str, NULL, 10);
   }
 
   if ((type & T_NAME_DB) == T_NAME_DB) {

source/common/src/trow.c

@@ -1063,7 +1063,7 @@ bool tdSTpRowGetVal(STSRow *pRow, col_id_t colId, col_type_t colType, int32_t fl
 
 int32_t tdGetColDataOfRow(SCellVal *pVal, SDataCol *pCol, int32_t row, int8_t bitmapMode) {
   if (isAllRowsNone(pCol)) {
-    pVal->valType = TD_VTYPE_NULL;
+    pVal->valType = TD_VTYPE_NONE;
 #ifdef TD_SUPPORT_READ2
     pVal->val = (void *)getNullValue(pCol->type);
 #else

source/common/src/ttime.c

@@ -374,39 +374,135 @@ char getPrecisionUnit(int32_t precision) {
 }
 
 int64_t convertTimePrecision(int64_t time, int32_t fromPrecision, int32_t toPrecision) {
-  assert(fromPrecision == TSDB_TIME_PRECISION_MILLI || fromPrecision == TSDB_TIME_PRECISION_MICRO ||
+  assert(fromPrecision == TSDB_TIME_PRECISION_MILLI ||
+         fromPrecision == TSDB_TIME_PRECISION_MICRO ||
          fromPrecision == TSDB_TIME_PRECISION_NANO);
-  assert(toPrecision == TSDB_TIME_PRECISION_MILLI || toPrecision == TSDB_TIME_PRECISION_MICRO ||
+  assert(toPrecision == TSDB_TIME_PRECISION_MILLI ||
+         toPrecision == TSDB_TIME_PRECISION_MICRO ||
          toPrecision == TSDB_TIME_PRECISION_NANO);
-  static double factors[3][3] = {{1., 1000., 1000000.}, {1.0 / 1000, 1., 1000.}, {1.0 / 1000000, 1.0 / 1000, 1.}};
-  return (int64_t)((double)time * factors[fromPrecision][toPrecision]);
+  double tempResult = (double)time;
+  switch(fromPrecision) {
+    case TSDB_TIME_PRECISION_MILLI: {
+      switch (toPrecision) {
+        case TSDB_TIME_PRECISION_MILLI:
+          return time;
+        case TSDB_TIME_PRECISION_MICRO:
+          tempResult *= 1000;
+          time *= 1000;
+          goto end_;
+        case TSDB_TIME_PRECISION_NANO:
+          tempResult *= 1000000;
+          time *= 1000000;
+          goto end_;
+      }
+    } // end from milli
+    case TSDB_TIME_PRECISION_MICRO: {
+      switch (toPrecision) {
+        case TSDB_TIME_PRECISION_MILLI:
+          return time / 1000;
+        case TSDB_TIME_PRECISION_MICRO:
+          return time;
+        case TSDB_TIME_PRECISION_NANO:
+          tempResult *= 1000;
+          time *= 1000;
+          goto end_;
+      }
+    } //end from micro
+    case TSDB_TIME_PRECISION_NANO: {
+      switch (toPrecision) {
+        case TSDB_TIME_PRECISION_MILLI:
+          return time / 1000000;
+        case TSDB_TIME_PRECISION_MICRO:
+          return time / 1000;
+        case TSDB_TIME_PRECISION_NANO:
+          return time;
+      }
+    } //end from nano
+    default: {
+      assert(0);
+      return time;  // only to pass windows compilation
+    }
+  } //end switch fromPrecision
+end_:
+  if (tempResult >= (double)INT64_MAX) return INT64_MAX;
+  if (tempResult <= (double)INT64_MIN) return INT64_MIN;  // INT64_MIN means NULL
+  return time;
 }
 
+// !!!!notice:there are precision problems, double lose precison if time is too large, for example: 1626006833631000000*1.0 = double = 1626006833631000064
+//int64_t convertTimePrecision(int64_t time, int32_t fromPrecision, int32_t toPrecision) {
+//  assert(fromPrecision == TSDB_TIME_PRECISION_MILLI || fromPrecision == TSDB_TIME_PRECISION_MICRO ||
+//         fromPrecision == TSDB_TIME_PRECISION_NANO);
+//  assert(toPrecision == TSDB_TIME_PRECISION_MILLI || toPrecision == TSDB_TIME_PRECISION_MICRO ||
+//         toPrecision == TSDB_TIME_PRECISION_NANO);
+//  static double factors[3][3] = {{1., 1000., 1000000.}, {1.0 / 1000, 1., 1000.}, {1.0 / 1000000, 1.0 / 1000, 1.}};
+//  ((double)time * factors[fromPrecision][toPrecision]);
+//}
+
+
+// !!!!notice: double lose precison if time is too large, for example: 1626006833631000000*1.0 = double = 1626006833631000064
 int64_t convertTimeFromPrecisionToUnit(int64_t time, int32_t fromPrecision, char toUnit) {
   assert(fromPrecision == TSDB_TIME_PRECISION_MILLI || fromPrecision == TSDB_TIME_PRECISION_MICRO ||
          fromPrecision == TSDB_TIME_PRECISION_NANO);
-  static double factors[3] = {1000000., 1000., 1.};
+  int64_t factors[3] = {NANOSECOND_PER_MSEC, NANOSECOND_PER_USEC, 1};
+  double tmp = time;
   switch (toUnit) {
-    case 's':
-      return time * factors[fromPrecision] / NANOSECOND_PER_SEC;
+    case 's':{
+      tmp /= (NANOSECOND_PER_SEC/factors[fromPrecision]);     // the result of division is an integer
+      time /= (NANOSECOND_PER_SEC/factors[fromPrecision]);
+      break;
+    }
     case 'm':
-      return time * factors[fromPrecision] / NANOSECOND_PER_MINUTE;
+      tmp /= (NANOSECOND_PER_MINUTE/factors[fromPrecision]);  // the result of division is an integer
+      time /= (NANOSECOND_PER_MINUTE/factors[fromPrecision]);
+      break;
     case 'h':
-      return time * factors[fromPrecision] / NANOSECOND_PER_HOUR;
+      tmp /= (NANOSECOND_PER_HOUR/factors[fromPrecision]);    // the result of division is an integer
+      time /= (NANOSECOND_PER_HOUR/factors[fromPrecision]);
+      break;
     case 'd':
-      return time * factors[fromPrecision] / NANOSECOND_PER_DAY;
+      tmp /= (NANOSECOND_PER_DAY/factors[fromPrecision]);     // the result of division is an integer
+      time /= (NANOSECOND_PER_DAY/factors[fromPrecision]);
+      break;
     case 'w':
-      return time * factors[fromPrecision] / NANOSECOND_PER_WEEK;
+      tmp /= (NANOSECOND_PER_WEEK/factors[fromPrecision]);    // the result of division is an integer
+      time /= (NANOSECOND_PER_WEEK/factors[fromPrecision]);
+      break;
     case 'a':
-      return time * factors[fromPrecision] / NANOSECOND_PER_MSEC;
+      tmp /= (NANOSECOND_PER_MSEC/factors[fromPrecision]);    // the result of division is an integer
+      time /= (NANOSECOND_PER_MSEC/factors[fromPrecision]);
+      break;
     case 'u':
-      return time * factors[fromPrecision] / NANOSECOND_PER_USEC;
+      // the result of (NANOSECOND_PER_USEC/(double)factors[fromPrecision]) maybe a double
+      switch (fromPrecision) {
+        case TSDB_TIME_PRECISION_MILLI:{
+          tmp *= 1000;
+          time *= 1000;
+          break;
+        }
+        case TSDB_TIME_PRECISION_MICRO:{
+          tmp /= 1;
+          time /= 1;
+          break;
+        }
+        case TSDB_TIME_PRECISION_NANO:{
+          tmp /= 1000;
+          time /= 1000;
+          break;
+        }
+      }
+      break;
     case 'b':
-      return time * factors[fromPrecision];
+      tmp *= factors[fromPrecision];
+      time *= factors[fromPrecision];
+      break;
     default: {
       return -1;
     }
   }
+  if (tmp >= (double)INT64_MAX) return INT64_MAX;
+  if (tmp <= (double)INT64_MIN) return INT64_MIN;
+  return time;
 }
 
 int32_t convertStringToTimestamp(int16_t type, char *inputData, int64_t timePrec, int64_t *timeVal) {
@@ -494,7 +590,7 @@ int32_t parseAbsoluteDuration(const char* token, int32_t tokenlen, int64_t* dura
   char* endPtr = NULL;
 
   /* get the basic numeric value */
-  int64_t timestamp = strtoll(token, &endPtr, 10);
+  int64_t timestamp = taosStr2Int64(token, &endPtr, 10);
   if (errno != 0) {
     return -1;
   }
@@ -512,7 +608,7 @@ int32_t parseNatualDuration(const char* token, int32_t tokenLen, int64_t* durati
   errno = 0;
 
   /* get the basic numeric value */
-  *duration = strtoll(token, NULL, 10);
+  *duration = taosStr2Int64(token, NULL, 10);
   if (errno != 0) {
     return -1;
   }

source/common/src/tvariant.c

@@ -39,7 +39,7 @@ int32_t toInteger(const char *z, int32_t n, int32_t base, int64_t *value) {
   errno = 0;
   char *endPtr = NULL;
 
-  *value = strtoll(z, &endPtr, base);
+  *value = taosStr2Int64(z, &endPtr, base);
   if (errno == ERANGE || errno == EINVAL || endPtr - z != n) {
     errno = 0;
     return -1;
@@ -58,7 +58,7 @@ int32_t toUInteger(const char *z, int32_t n, int32_t base, uint64_t *value) {
     return -1;
   }
 
-  *value = strtoull(z, &endPtr, base);
+  *value = taosStr2UInt64(z, &endPtr, base);
   if (errno == ERANGE || errno == EINVAL || endPtr - z != n) {
     errno = 0;
     return -1;
@@ -434,7 +434,7 @@ static FORCE_INLINE int32_t convertToDouble(char *pStr, int32_t len, double *val
   //    return -1;
   //  }
   //
-  //  *value = strtod(pStr, NULL);
+  //  *value = taosStr2Double(pStr, NULL);
   return 0;
 }
 
@@ -911,7 +911,7 @@ int32_t taosVariantTypeSetType(SVariant *pVariant, char type) {
     case TSDB_DATA_TYPE_DOUBLE: {
       if (pVariant->nType == TSDB_DATA_TYPE_BINARY) {
         errno = 0;
-        double v = strtod(pVariant->pz, NULL);
+        double v = taosStr2Double(pVariant->pz, NULL);
         if ((errno == ERANGE && v == -1) || (isinf(v) || isnan(v))) {
           taosMemoryFree(pVariant->pz);
           return -1;

source/common/test/CMakeLists.txt

@@ -17,6 +17,15 @@ TARGET_INCLUDE_DIRECTORIES(
         PRIVATE "${TD_SOURCE_DIR}/source/libs/common/inc"
 )
 
+# dataformatTest.cpp 
+add_executable(dataformatTest "")
+target_sources(
+    dataformatTest
+    PRIVATE
+    "dataformatTest.cpp"
+)
+target_link_libraries(dataformatTest gtest gtest_main util)
+
 # tmsg test
 # add_executable(tmsgTest "")
 # target_sources(tmsgTest 

source/common/test/dataformatTest.cpp

+#include "gtest/gtest.h"
\ No newline at end of file

source/common/test/trowTest.cpp

-#include <gtest/gtest.h>
-
-#include "trow.h"
-
-TEST(td_row_test, build_row_to_target) {
-#if 0
-  char        dst[1024];
-  SRow*       pRow = (SRow*)dst;
-  int         ncols = 10;
-  col_id_t    cid;
-  void*       pData;
-  SRowBuilder rb = trbInit(TD_OR_ROW_BUILDER, NULL, 0, pRow, 1024);
-
-  trbSetRowInfo(&rb, false, 0);
-  trbSetRowTS(&rb, 1637550210000);
-  for (int c = 0; c < ncols; c++) {
-    cid = c;
-    if (trbWriteCol(&rb, pData, cid) < 0) {
-      // TODO
-    }
-  }
-#endif
-}
\ No newline at end of file

source/common/type/type.c

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
\ No newline at end of file

source/common/type/typeBigint.c

-/*
- * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
- *
- * This program is free software: you can use, redistribute, and/or modify
- * it under the terms of the GNU Affero General Public License, version 3
- * or later ("AGPL"), as published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
- * FITNESS FOR A PARTICULAR PURPOSE.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
- */
\ No newline at end of file