refactor: merge from 3.0

cmake/addr2line_CMakeLists.txt.in

+
+# addr2line
+ExternalProject_Add(addr2line
+        GIT_REPOSITORY https://github.com/davea42/libdwarf-addr2line.git
+        GIT_TAG master
+        SOURCE_DIR "${TD_CONTRIB_DIR}/addr2line"
+        BINARY_DIR "${TD_CONTRIB_DIR}/addr2line"
+        CONFIGURE_COMMAND "" 
+        BUILD_COMMAND ""  
+        INSTALL_COMMAND ""
+        TEST_COMMAND ""
+        )

cmake/cmake.define

@@ -15,12 +15,12 @@ MESSAGE(STATUS "Project executable files output path: " ${EXECUTABLE_OUTPUT_PATH
 MESSAGE(STATUS "Project library files output path: " ${LIBRARY_OUTPUT_PATH})
 
 find_package(Git QUIET)
-if(GIT_FOUND AND EXISTS "${PROJECT_SOURCE_DIR}/.git")
+if(GIT_FOUND AND EXISTS "${TD_SOURCE_DIR}/.git")
 # Update submodules as needed
     option(GIT_SUBMODULE "Check submodules during build" ON)
     if(GIT_SUBMODULE)
         message(STATUS "Submodule update")
-        execute_process(COMMAND ${GIT_EXECUTABLE} submodule update --init --recursive
+        execute_process(COMMAND cd ${TD_SOURCE_DIR} && ${GIT_EXECUTABLE} submodule update --init --recursive
                         WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}
                         RESULT_VARIABLE GIT_SUBMOD_RESULT)
         if(NOT GIT_SUBMOD_RESULT EQUAL "0")
@@ -29,7 +29,7 @@ if(GIT_FOUND AND EXISTS "${PROJECT_SOURCE_DIR}/.git")
     endif()
 endif()
 
-if(NOT EXISTS "${PROJECT_SOURCE_DIR}/tools/taos-tools/CMakeLists.txt")
+if(NOT EXISTS "${TD_SOURCE_DIR}/tools/taos-tools/CMakeLists.txt")
     message(WARNING "The submodules were not downloaded! GIT_SUBMODULE was turned off or failed. Please update submodules manually if you need build them.")
 endif()
 
@@ -75,6 +75,14 @@ IF (TD_WINDOWS)
     SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${COMMON_FLAGS}")
 
 ELSE ()
+    IF (${COVER} MATCHES "true")
+        MESSAGE(STATUS "Test coverage mode, add extra flags")
+        SET(GCC_COVERAGE_COMPILE_FLAGS "-fprofile-arcs -ftest-coverage")
+        SET(GCC_COVERAGE_LINK_FLAGS    "-lgcov --coverage")
+        SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} ${GCC_COVERAGE_COMPILE_FLAGS} ${GCC_COVERAGE_LINK_FLAGS}")
+        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${GCC_COVERAGE_COMPILE_FLAGS} ${GCC_COVERAGE_LINK_FLAGS}")
+    ENDIF ()
+
     IF (${SANITIZER} MATCHES "true")
         SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS}     -Werror -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fno-sanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=null -fno-sanitize=alignment -static-libasan -g3")
         SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror -Wno-literal-suffix -Werror=return-type -fPIC -gdwarf-2 -fsanitize=address -fsanitize=undefined -fno-sanitize-recover=all -fsanitize=float-divide-by-zero -fsanitize=float-cast-overflow -fno-sanitize=null -fno-sanitize=alignment -static-libasan -g3")

cmake/cmake.options

@@ -48,6 +48,12 @@ IF(${TD_WINDOWS})
 
 ENDIF ()
 
+option(
+    BUILD_ADDR2LINE
+    "If build addr2line"
+    OFF
+    )
+
 option(
     BUILD_TEST 
     "If build unit tests using googletest" 

cmake/libdwarf_CMakeLists.txt.in

+
+# libdwarf
+ExternalProject_Add(libdwarf
+        GIT_REPOSITORY https://github.com/davea42/libdwarf-code.git
+        GIT_TAG libdwarf-0.3.1
+        SOURCE_DIR "${TD_CONTRIB_DIR}/libdwarf"
+        BINARY_DIR "${TD_CONTRIB_DIR}/libdwarf"
+        CONFIGURE_COMMAND "" 
+        BUILD_COMMAND ""  
+        INSTALL_COMMAND ""
+        TEST_COMMAND ""
+        )

contrib/CMakeLists.txt

@@ -98,6 +98,12 @@ if(${BUILD_WITH_NURAFT})
     cat("${TD_SUPPORT_DIR}/nuraft_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
 endif(${BUILD_WITH_NURAFT})
 
+# addr2line
+if(${BUILD_ADDR2LINE})
+    cat("${TD_SUPPORT_DIR}/libdwarf_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+    cat("${TD_SUPPORT_DIR}/addr2line_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+endif(${BUILD_ADDR2LINE})
+
 # download dependencies
 configure_file(${CONTRIB_TMP_FILE} "${TD_CONTRIB_DIR}/deps-download/CMakeLists.txt")
 execute_process(COMMAND "${CMAKE_COMMAND}" -G "${CMAKE_GENERATOR}" .
@@ -327,7 +333,48 @@ if(${BUILD_WITH_SQLITE})
     endif(NOT TD_WINDOWS)
 endif(${BUILD_WITH_SQLITE})
 
-# pthread
+# addr2line
+if(${BUILD_ADDR2LINE})
+    check_include_file( "sys/types.h"     HAVE_SYS_TYPES_H) 
+    check_include_file( "sys/stat.h"      HAVE_SYS_STAT_H ) 
+    check_include_file( "inttypes.h"      HAVE_INTTYPES_H   ) 
+    check_include_file( "stddef.h"        HAVE_STDDEF_H   ) 
+    check_include_file( "stdlib.h"        HAVE_STDLIB_H   ) 
+    check_include_file( "string.h"        HAVE_STRING_H   ) 
+    check_include_file( "memory.h"        HAVE_MEMORY_H   ) 
+    check_include_file( "strings.h"       HAVE_STRINGS_H  ) 
+    check_include_file( "stdint.h"        HAVE_STDINT_H   )
+    check_include_file( "unistd.h"        HAVE_UNISTD_H   )
+    check_include_file( "sgidefs.h"       HAVE_SGIDEFS_H  )
+    check_include_file( "stdafx.h"        HAVE_STDAFX_H   )
+    check_include_file( "elf.h"           HAVE_ELF_H      ) 
+    check_include_file( "libelf.h"        HAVE_LIBELF_H   ) 
+    check_include_file( "libelf/libelf.h" HAVE_LIBELF_LIBELF_H) 
+    check_include_file( "alloca.h"        HAVE_ALLOCA_H   )
+    check_include_file( "elfaccess.h"     HAVE_ELFACCESS_H)
+    check_include_file( "sys/elf_386.h"   HAVE_SYS_ELF_386_H  )
+    check_include_file( "sys/elf_amd64.h" HAVE_SYS_ELF_AMD64_H)
+    check_include_file( "sys/elf_sparc.h" HAVE_SYS_ELF_SPARC_H)
+    check_include_file( "sys/ia64/elf.h"  HAVE_SYS_IA64_ELF_H )
+    set(VERSION 0.3.1)
+    set(PACKAGE_VERSION "\"${VERSION}\"") 
+    configure_file(libdwarf/cmake/config.h.cmake config.h)
+    file(GLOB_RECURSE LIBDWARF_SOURCES "libdwarf/src/lib/libdwarf/*.c")
+    add_library(libdwarf STATIC ${LIBDWARF_SOURCES})
+    set_target_properties(libdwarf PROPERTIES OUTPUT_NAME "libdwarf")
+    if(HAVE_LIBELF_H OR HAVE_LIBELF_LIBELF_H)
+        target_link_libraries(libdwarf PUBLIC libelf)
+    endif()
+    target_include_directories(libdwarf SYSTEM PUBLIC "libdwarf/src/lib/libdwarf" ${CMAKE_BINARY_DIR}/contrib)
+    file(READ "addr2line/addr2line.c" ADDR2LINE_CONTENT)
+    string(REPLACE "static int" "int" ADDR2LINE_CONTENT "${ADDR2LINE_CONTENT}")
+    string(REPLACE "static void" "void" ADDR2LINE_CONTENT "${ADDR2LINE_CONTENT}")
+    string(REPLACE "main(" "main_addr2line(" ADDR2LINE_CONTENT "${ADDR2LINE_CONTENT}")
+    file(WRITE "addr2line/addr2line.c" "${ADDR2LINE_CONTENT}")
+    add_library(addr2line STATIC "addr2line/addr2line.c")
+    target_link_libraries(addr2line PUBLIC libdwarf dl z)
+    target_include_directories(addr2line PUBLIC "libdwarf/src/lib/libdwarf" )
+endif(${BUILD_ADDR2LINE})
 
 
 # ================================================================================================

example/src/tstream.c

@@ -83,7 +83,7 @@ int32_t create_stream() {
   /*pRes = tmq_create_stream(pConn, "stream1", "out1", sql);*/
   pRes = taos_query(
       pConn,
-      "create stream stream1 trigger window_close into outstb as select _wstartts, min(k), max(k), sum(k) as sum_of_k "
+      "create stream stream1 trigger at_once into outstb as select _wstartts, min(k), max(k), sum(k) as sum_of_k "
       "from tu1 interval(10m)");
   if (taos_errno(pRes) != 0) {
     printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes));

include/client/taos.h

@@ -221,15 +221,12 @@ DLL_EXPORT void        tmq_list_destroy(tmq_list_t *);
 DLL_EXPORT int32_t     tmq_list_get_size(const tmq_list_t *);
 DLL_EXPORT char      **tmq_list_to_c_array(const tmq_list_t *);
 
-#if 0
-DLL_EXPORT tmq_t *tmq_consumer_new(void *conn, tmq_conf_t *conf, char *errstr, int32_t errstrLen);
-#endif
-
 DLL_EXPORT tmq_t *tmq_consumer_new(tmq_conf_t *conf, char *errstr, int32_t errstrLen);
 
 DLL_EXPORT const char *tmq_err2str(tmq_resp_err_t);
 
 /* ------------------------TMQ CONSUMER INTERFACE------------------------ */
+
 DLL_EXPORT tmq_resp_err_t tmq_subscribe(tmq_t *tmq, const tmq_list_t *topic_list);
 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);
@@ -240,6 +237,7 @@ DLL_EXPORT tmq_resp_err_t tmq_commit(tmq_t *tmq, const tmq_topic_vgroup_list_t *
 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_seek(tmq_t *tmq, const tmq_topic_vgroup_t *offset);
 #endif
+
 /* ----------------------TMQ CONFIGURATION INTERFACE---------------------- */
 
 enum tmq_conf_res_t {
@@ -268,12 +266,9 @@ DLL_EXPORT char   *tmq_get_table_name(TAOS_RES *res);
 DLL_EXPORT int64_t     tmq_get_request_offset(tmq_message_t *message);
 DLL_EXPORT int64_t     tmq_get_response_offset(tmq_message_t *message);
 #endif
-/* --------------------TMPORARY INTERFACE FOR TESTING--------------------- */
-#if 0
-DLL_EXPORT TAOS_RES *tmq_create_topic(TAOS *taos, const char *name, const char *sql, int sqlLen);
-DLL_EXPORT TAOS_RES *tmq_create_stream(TAOS *taos, const char *streamName, const char *tbName, const char *sql);
-#endif
+
 /* ------------------------------ TMQ END -------------------------------- */
+
 #if 1  // Shuduo: temporary enable for app build
 typedef void (*TAOS_SUBSCRIBE_CALLBACK)(TAOS_SUB *tsub, TAOS_RES *res, void *param, int code);
 #endif

include/common/systable.h

@@ -45,7 +45,6 @@ extern "C" {
 
 #define TSDB_PERFORMANCE_SCHEMA_DB     "performance_schema"
 #define TSDB_PERFS_TABLE_SMAS          "smas"
-#define TSDB_PERFS_TABLE_SUBSCRIBES    "subscribes"
 #define TSDB_PERFS_TABLE_CONNECTIONS   "connections"
 #define TSDB_PERFS_TABLE_QUERIES       "queries"
 #define TSDB_PERFS_TABLE_TOPICS        "topics"

include/util/tskiplist2.h → include/common/tdata.h

@@ -12,59 +12,34 @@
  * 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_SKIPLIST2_H_
-#define _TD_UTIL_SKIPLIST2_H_
 
-#include "os.h"
+#ifndef _TD_TDATA_H_
+#define _TD_TDATA_H_
 
 #ifdef __cplusplus
 extern "C" {
 #endif
 
-#define SL_MAX_LEVEL 15
-
-typedef struct SSkipList2 SSkipList2;
-typedef struct SSLCursor  SSLCursor;
-typedef struct SSLCfg     SSLCfg;
-typedef struct SSLNode    SSLNode;
-
-typedef int32_t (*tslCmprFn)(const void *pKey1, int32_t nKey1, const void *pKey2, int32_t nKey2);
-
-// SSkipList2
-int32_t slOpen(const SSLCfg *pCfg, SSkipList2 **ppSl);
-int32_t slClose(SSkipList2 *pSl);
-int32_t slClear(SSkipList2 *pSl);
+#include "os.h"
 
-// SSLCursor
-int32_t slcOpen(SSkipList2 *pSl, SSLCursor *pSlc);
-int32_t slcClose(SSLCursor *pSlc);
-int32_t slcMoveTo(SSLCursor *pSlc, const void *pKey, int32_t nKey);
-int32_t slcMoveToNext(SSLCursor *pSlc);
-int32_t slcMoveToPrev(SSLCursor *pSlc);
-int32_t slcMoveToFirst(SSLCursor *pSlc);
-int32_t slcMoveToLast(SSLCursor *pSlc);
-int32_t slcPut(SSLCursor *pSlc, const void *pKey, int32_t nKey, const void *pData, int32_t nData);
-int32_t slcGet(SSLCursor *pSlc, const void **ppKey, int32_t *nKey, const void **ppData, int32_t *nData);
-int32_t slcDrop(SSLCursor *pSlc);
+typedef struct STaosData TDATA, tdata_t;
 
-// struct
-struct SSLCfg {
-  int8_t    maxLevel;
-  int32_t   nKey;
-  int32_t   nData;
-  tslCmprFn cmprFn;
-  void     *pPool;
-  void *(*xMalloc)(void *, int32_t size);
-  void (*xFree)(void *, void *);
-};
+typedef enum {
+  TAOS_META_STABLE_DATA = 0,  // super table meta
+  TAOS_META_TABLE_DATA,       // non-super table meta
+  TAOS_TS_ROW_DATA,           // row time-series data
+  TAOS_TS_COL_DATA,           // col time-series data
+  TAOS_DATA_MAX
+} ETaosDataT;
 
-struct SSLCursor {
-  SSkipList2 *pSl;
-  SSLNode   **forwards[SL_MAX_LEVEL];
+struct STaosData {
+  ETaosDataT type;
+  uint32_t   nPayload;
+  uint8_t   *pPayload;
 };
 
 #ifdef __cplusplus
 }
 #endif
 
-#endif /*_TD_UTIL_SKIPLIST2_H_*/
+#endif /*_TD_TDATA_H_*/
\ No newline at end of file

include/common/tdataformat.h

@@ -150,29 +150,6 @@ int32_t   tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, int8_t flags,
 STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder);
 
 // ----------------- Semantic timestamp key definition
-#ifdef TD_2_0
-
-typedef uint64_t TKEY;
-
-#define TKEY_INVALID       UINT64_MAX
-#define TKEY_NULL          TKEY_INVALID
-#define TKEY_NEGATIVE_FLAG (((TKEY)1) << 63)
-#define TKEY_DELETE_FLAG   (((TKEY)1) << 62)
-#define TKEY_VALUE_FILTER  (~(TKEY_NEGATIVE_FLAG | TKEY_DELETE_FLAG))
-
-#define TKEY_IS_NEGATIVE(tkey) (((tkey)&TKEY_NEGATIVE_FLAG) != 0)
-#define TKEY_IS_DELETED(tkey)  (((tkey)&TKEY_DELETE_FLAG) != 0)
-#define tdSetTKEYDeleted(tkey) ((tkey) | TKEY_DELETE_FLAG)
-#define tdGetTKEY(key)         (((TKEY)TABS(key)) | (TKEY_NEGATIVE_FLAG & (TKEY)(key)))
-#define tdGetKey(tkey)         (((TSKEY)((tkey)&TKEY_VALUE_FILTER)) * (TKEY_IS_NEGATIVE(tkey) ? -1 : 1))
-
-#define MIN_TS_KEY ((TSKEY)0x8000000000000001)
-#define MAX_TS_KEY ((TSKEY)0x3fffffffffffffff)
-
-#define TD_TO_TKEY(key) tdGetTKEY(((key) < MIN_TS_KEY) ? MIN_TS_KEY : (((key) > MAX_TS_KEY) ? MAX_TS_KEY : key))
-
-#else
-
 // typedef uint64_t TKEY;
 #define TKEY TSKEY
 
@@ -192,8 +169,6 @@ typedef uint64_t TKEY;
 
 #define TD_TO_TKEY(key) tdGetTKEY(((key) < MIN_TS_KEY) ? MIN_TS_KEY : (((key) > MAX_TS_KEY) ? MAX_TS_KEY : key))
 
-#endif
-
 static FORCE_INLINE TKEY keyToTkey(TSKEY key) {
   TSKEY lkey = key;
   if (key > MAX_TS_KEY) {
@@ -218,157 +193,6 @@ static FORCE_INLINE int32_t tkeyComparFn(const void *tkey1, const void *tkey2) {
   }
 }
 
-#if 0
-// ----------------- Data row structure
-
-/* A data row, the format is like below:
- * |<------------------------------------------------ len ---------------------------------->|
- * |<--     Head      -->|<---------   flen -------------->|                                 |
- * +---------------------+---------------------------------+---------------------------------+
- * | uint16_t |  int16_t |                                 |                                 |
- * +----------+----------+---------------------------------+---------------------------------+
- * |   len    | sversion |           First part            |             Second part         |
- * +----------+----------+---------------------------------+---------------------------------+
- *
- * NOTE: timestamp in this row structure is TKEY instead of TSKEY
- */
-typedef void *SDataRow;
-
-#define TD_DATA_ROW_HEAD_SIZE (sizeof(uint16_t) + sizeof(int16_t))
-
-#define dataRowLen(r)                (*(TDRowLenT *)(r))  // 0~65535
-#define dataRowEnd(r)                POINTER_SHIFT(r, dataRowLen(r))
-#define dataRowVersion(r)            (*(int16_t *)POINTER_SHIFT(r, sizeof(int16_t)))
-#define dataRowTuple(r)              POINTER_SHIFT(r, TD_DATA_ROW_HEAD_SIZE)
-#define dataRowTKey(r)               (*(TKEY *)(dataRowTuple(r)))
-#define dataRowKey(r)                tdGetKey(dataRowTKey(r))
-#define dataRowSetLen(r, l)          (dataRowLen(r) = (l))
-#define dataRowSetVersion(r, v)      (dataRowVersion(r) = (v))
-#define dataRowCpy(dst, r)           memcpy((dst), (r), dataRowLen(r))
-#define dataRowMaxBytesFromSchema(s) (schemaTLen(s) + TD_DATA_ROW_HEAD_SIZE)
-#define dataRowDeleted(r)            TKEY_IS_DELETED(dataRowTKey(r))
-
-SDataRow tdNewDataRowFromSchema(STSchema *pSchema);
-void     tdFreeDataRow(SDataRow row);
-void     tdInitDataRow(SDataRow row, STSchema *pSchema);
-SDataRow tdDataRowDup(SDataRow row);
-
-// offset here not include dataRow header length
-static FORCE_INLINE int32_t tdAppendDataColVal(SDataRow row, const void *value, bool isCopyVarData, int8_t type,
-                                           int32_t offset) {
-  assert(value != NULL);
-  int32_t toffset = offset + TD_DATA_ROW_HEAD_SIZE;
-
-  if (IS_VAR_DATA_TYPE(type)) {
-    *(VarDataOffsetT *)POINTER_SHIFT(row, toffset) = dataRowLen(row);
-    if (isCopyVarData) {
-      memcpy(POINTER_SHIFT(row, dataRowLen(row)), value, varDataTLen(value));
-    }
-    dataRowLen(row) += varDataTLen(value);
-  } else {
-    if (offset == 0) {
-      assert(type == TSDB_DATA_TYPE_TIMESTAMP);
-      TKEY tvalue = tdGetTKEY(*(TSKEY *)value);
-      memcpy(POINTER_SHIFT(row, toffset), (const void *)(&tvalue), TYPE_BYTES[type]);
-    } else {
-      memcpy(POINTER_SHIFT(row, toffset), value, TYPE_BYTES[type]);
-    }
-  }
-
-  return 0;
-}
-
-// offset here not include dataRow header length
-static FORCE_INLINE int32_t tdAppendColVal(SDataRow row, const void *value, int8_t type, int32_t offset) {
-  return tdAppendDataColVal(row, value, true, type, offset);
-}
-
-// NOTE: offset here including the header size
-static FORCE_INLINE void *tdGetRowDataOfCol(SDataRow row, int8_t type, int32_t offset) {
-  if (IS_VAR_DATA_TYPE(type)) {
-    return POINTER_SHIFT(row, *(VarDataOffsetT *)POINTER_SHIFT(row, offset));
-  } else {
-    return POINTER_SHIFT(row, offset);
-  }
-}
-
-static FORCE_INLINE void *tdGetPtrToCol(SDataRow row, STSchema *pSchema, int32_t idx) {
-  return POINTER_SHIFT(row, TD_DATA_ROW_HEAD_SIZE + pSchema->columns[idx].offset);
-}
-
-static FORCE_INLINE void *tdGetColOfRowBySchema(SDataRow row, STSchema *pSchema, int32_t idx) {
-  int16_t offset = TD_DATA_ROW_HEAD_SIZE + pSchema->columns[idx].offset;
-  int8_t  type = pSchema->columns[idx].type;
-
-  return tdGetRowDataOfCol(row, type, offset);
-}
-
-static FORCE_INLINE bool tdIsColOfRowNullBySchema(SDataRow row, STSchema *pSchema, int32_t idx) {
-  int16_t offset = TD_DATA_ROW_HEAD_SIZE + pSchema->columns[idx].offset;
-  int8_t  type = pSchema->columns[idx].type;
-
-  return isNull(tdGetRowDataOfCol(row, type, offset), type);
-}
-
-static FORCE_INLINE void tdSetColOfRowNullBySchema(SDataRow row, STSchema *pSchema, int32_t idx) {
-  int16_t offset = TD_DATA_ROW_HEAD_SIZE + pSchema->columns[idx].offset;
-  int8_t  type = pSchema->columns[idx].type;
-  int16_t bytes = pSchema->columns[idx].bytes;
-
-  setNull(tdGetRowDataOfCol(row, type, offset), type, bytes);
-}
-
-static FORCE_INLINE void tdCopyColOfRowBySchema(SDataRow dst, STSchema *pDstSchema, int32_t dstIdx, SDataRow src,
-                                                STSchema *pSrcSchema, int32_t srcIdx) {
-  int8_t type = pDstSchema->columns[dstIdx].type;
-  assert(type == pSrcSchema->columns[srcIdx].type);
-  void *pData = tdGetPtrToCol(dst, pDstSchema, dstIdx);
-  void *value = tdGetPtrToCol(src, pSrcSchema, srcIdx);
-
-  switch (type) {
-    case TSDB_DATA_TYPE_BINARY:
-    case TSDB_DATA_TYPE_NCHAR:
-      *(VarDataOffsetT *)pData = *(VarDataOffsetT *)value;
-      pData = POINTER_SHIFT(dst, *(VarDataOffsetT *)pData);
-      value = POINTER_SHIFT(src, *(VarDataOffsetT *)value);
-      memcpy(pData, value, varDataTLen(value));
-      break;
-    case TSDB_DATA_TYPE_NULL:
-    case TSDB_DATA_TYPE_BOOL:
-    case TSDB_DATA_TYPE_TINYINT:
-    case TSDB_DATA_TYPE_UTINYINT:
-      *(uint8_t *)pData = *(uint8_t *)value;
-      break;
-    case TSDB_DATA_TYPE_SMALLINT:
-    case TSDB_DATA_TYPE_USMALLINT:
-      *(uint16_t *)pData = *(uint16_t *)value;
-      break;
-    case TSDB_DATA_TYPE_INT:
-    case TSDB_DATA_TYPE_UINT:
-      *(uint32_t *)pData = *(uint32_t *)value;
-      break;
-    case TSDB_DATA_TYPE_BIGINT:
-    case TSDB_DATA_TYPE_UBIGINT:
-      *(uint64_t *)pData = *(uint64_t *)value;
-      break;
-    case TSDB_DATA_TYPE_FLOAT:
-      SET_FLOAT_PTR(pData, value);
-      break;
-    case TSDB_DATA_TYPE_DOUBLE:
-      SET_DOUBLE_PTR(pData, value);
-      break;
-    case TSDB_DATA_TYPE_TIMESTAMP:
-      if (pSrcSchema->columns[srcIdx].colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
-        *(TSKEY *)pData = tdGetKey(*(TKEY *)value);
-      } else {
-        *(TSKEY *)pData = *(TSKEY *)value;
-      }
-      break;
-    default:
-      memcpy(pData, value, pSrcSchema->columns[srcIdx].bytes);
-  }
-}
-#endif
 // ----------------- Data column structure
 // SDataCol arrangement: data => bitmap => dataOffset
 typedef struct SDataCol {
@@ -398,29 +222,6 @@ void   *dataColSetOffset(SDataCol *pCol, int32_t nEle);
 
 bool isNEleNull(SDataCol *pCol, int32_t nEle);
 
-#if 0
-// Get the data pointer from a column-wised data
-static FORCE_INLINE const void *tdGetColDataOfRow(SDataCol *pCol, int32_t row) {
-  if (isAllRowsNull(pCol)) {
-    return getNullValue(pCol->type);
-  }
-  if (IS_VAR_DATA_TYPE(pCol->type)) {
-    return POINTER_SHIFT(pCol->pData, pCol->dataOff[row]);
-  } else {
-    return POINTER_SHIFT(pCol->pData, TYPE_BYTES[pCol->type] * row);
-  }
-}
-
-static FORCE_INLINE int32_t dataColGetNEleLen(SDataCol *pDataCol, int32_t rows) {
-  assert(rows > 0);
-
-  if (IS_VAR_DATA_TYPE(pDataCol->type)) {
-    return pDataCol->dataOff[rows - 1] + varDataTLen(tdGetColDataOfRow(pDataCol, rows - 1));
-  } else {
-    return TYPE_BYTES[pDataCol->type] * rows;
-  }
-}
-#endif
 typedef struct {
   col_id_t  maxCols;    // max number of columns
   col_id_t  numOfCols;  // Total number of cols
@@ -479,7 +280,8 @@ void       tdResetDataCols(SDataCols *pCols);
 int32_t    tdInitDataCols(SDataCols *pCols, STSchema *pSchema);
 SDataCols *tdDupDataCols(SDataCols *pCols, bool keepData);
 SDataCols *tdFreeDataCols(SDataCols *pCols);
-int32_t tdMergeDataCols(SDataCols *target, SDataCols *source, int32_t rowsToMerge, int32_t *pOffset, bool update, TDRowVerT maxVer);
+int32_t    tdMergeDataCols(SDataCols *target, SDataCols *source, int32_t rowsToMerge, int32_t *pOffset, bool update,
+                           TDRowVerT maxVer);
 
 // ----------------- K-V data row structure
 /* |<-------------------------------------- len -------------------------------------------->|
@@ -542,54 +344,6 @@ static FORCE_INLINE void *tdGetKVRowIdxOfCol(SKVRow row, int16_t colId) {
   return taosbsearch(&colId, kvRowColIdx(row), kvRowNCols(row), sizeof(SColIdx), comparTagId, TD_EQ);
 }
 
-#if 0
-// offset here not include kvRow header length
-static FORCE_INLINE int32_t tdAppendKvColVal(SKVRow row, const void *value, bool isCopyValData, int16_t colId, int8_t type,
-                                         int32_t offset) {
-  assert(value != NULL);
-  int32_t  toffset = offset + TD_KV_ROW_HEAD_SIZE;
-  SColIdx *pColIdx = (SColIdx *)POINTER_SHIFT(row, toffset);
-  char *   ptr = (char *)POINTER_SHIFT(row, kvRowLen(row));
-
-  pColIdx->colId = colId;
-  pColIdx->offset = kvRowLen(row);  // offset of pColIdx including the TD_KV_ROW_HEAD_SIZE
-
-  if (IS_VAR_DATA_TYPE(type)) {
-    if (isCopyValData) {
-      memcpy(ptr, value, varDataTLen(value));
-    }
-    kvRowLen(row) += varDataTLen(value);
-  } else {
-    if (offset == 0) {
-      assert(type == TSDB_DATA_TYPE_TIMESTAMP);
-      TKEY tvalue = tdGetTKEY(*(TSKEY *)value);
-      memcpy(ptr, (void *)(&tvalue), TYPE_BYTES[type]);
-    } else {
-      memcpy(ptr, value, TYPE_BYTES[type]);
-    }
-    kvRowLen(row) += TYPE_BYTES[type];
-  }
-
-  return 0;
-}
-// NOTE: offset here including the header size
-static FORCE_INLINE void *tdGetKvRowDataOfCol(void *row, int32_t offset) { return POINTER_SHIFT(row, offset); }
-
-static FORCE_INLINE void *tdGetKVRowValOfColEx(SKVRow row, int16_t colId, int32_t *nIdx) {
-  while (*nIdx < kvRowNCols(row)) {
-    SColIdx *pColIdx = kvRowColIdxAt(row, *nIdx);
-    if (pColIdx->colId == colId) {
-      ++(*nIdx);
-      return tdGetKvRowDataOfCol(row, pColIdx->offset);
-    } else if (pColIdx->colId > colId) {
-      return NULL;
-    } else {
-      ++(*nIdx);
-    }
-  }
-  return NULL;
-}
-#endif
 // ----------------- K-V data row builder
 typedef struct {
   int16_t  tCols;
@@ -632,166 +386,6 @@ static FORCE_INLINE int32_t tdAddColToKVRow(SKVRowBuilder *pBuilder, col_id_t co
 
   return 0;
 }
-#if 0
-// ----------------- SMemRow appended with tuple row structure
-/*
- * |---------|------------------------------------------------- len ---------------------------------->|
- * |<--------     Head      ------>|<---------   flen -------------->|                                 |
- * |---------+---------------------+---------------------------------+---------------------------------+
- * | uint8_t | uint16_t |  int16_t |                                 |                                 |
- * |---------+----------+----------+---------------------------------+---------------------------------+
- * |  flag   |   len    | sversion |           First part            |             Second part         |
- * +---------+----------+----------+---------------------------------+---------------------------------+
- *
- * NOTE: timestamp in this row structure is TKEY instead of TSKEY
- */
-
-// ----------------- SMemRow appended with extended K-V data row structure
-/* |--------------------|------------------------------------------------  len ---------------------------------->|
- * |<-------------     Head      ------------>|<---------   flen -------------->|                                 |
- * |--------------------+----------+--------------------------------------------+---------------------------------+
- * | uint8_t | int16_t  | uint16_t |  int16_t |                                 |                                 |
- * |---------+----------+----------+----------+---------------------------------+---------------------------------+
- * |   flag  | sversion |   len    |   ncols  |           cols index            |             data part           |
- * |---------+----------+----------+----------+---------------------------------+---------------------------------+
- */
-
-typedef void *SMemRow;
-
-#define TD_MEM_ROW_TYPE_SIZE        sizeof(uint8_t)
-#define TD_MEM_ROW_KV_VER_SIZE      sizeof(int16_t)
-#define TD_MEM_ROW_KV_TYPE_VER_SIZE (TD_MEM_ROW_TYPE_SIZE + TD_MEM_ROW_KV_VER_SIZE)
-#define TD_MEM_ROW_DATA_HEAD_SIZE   (TD_MEM_ROW_TYPE_SIZE + TD_DATA_ROW_HEAD_SIZE)
-#define TD_MEM_ROW_KV_HEAD_SIZE     (TD_MEM_ROW_TYPE_SIZE + TD_MEM_ROW_KV_VER_SIZE + TD_KV_ROW_HEAD_SIZE)
-
-#define SMEM_ROW_DATA 0x0U   // SDataRow
-#define SMEM_ROW_KV   0x01U  // SKVRow
-
-#define KVRatioConvert (0.9f)
-
-#define memRowType(r) ((*(uint8_t *)(r)) & 0x01)
-
-#define memRowSetType(r, t)  ((*(uint8_t *)(r)) = (t))  // set the total byte in case of dirty memory
-#define isDataRowT(t)        (SMEM_ROW_DATA == (((uint8_t)(t)) & 0x01))
-#define isDataRow(r)         (SMEM_ROW_DATA == memRowType(r))
-#define isKvRowT(t)          (SMEM_ROW_KV == (((uint8_t)(t)) & 0x01))
-#define isKvRow(r)           (SMEM_ROW_KV == memRowType(r))
-#define isUtilizeKVRow(k, d) ((k) < ((d)*KVRatioConvert))
-
-#define memRowDataBody(r) POINTER_SHIFT(r, TD_MEM_ROW_TYPE_SIZE)  // section after flag
-#define memRowKvBody(r) \
-  POINTER_SHIFT(r, TD_MEM_ROW_KV_TYPE_VER_SIZE)  // section after flag + sversion as to reuse SKVRow
-
-#define memRowDataLen(r) (*(TDRowLenT *)memRowDataBody(r))  //  0~65535
-#define memRowKvLen(r)   (*(TDRowLenT *)memRowKvBody(r))    //  0~65535
-
-#define memRowDataTLen(r) \
-  ((TDRowLenT)(memRowDataLen(r) + TD_MEM_ROW_TYPE_SIZE))  // using uint32_t/int32_t to store the TLen
-
-#define memRowKvTLen(r) ((TDRowLenT)(memRowKvLen(r) + TD_MEM_ROW_KV_TYPE_VER_SIZE))
-
-#define memRowLen(r)  (isDataRow(r) ? memRowDataLen(r) : memRowKvLen(r))
-#define memRowTLen(r) (isDataRow(r) ? memRowDataTLen(r) : memRowKvTLen(r))  // using uint32_t/int32_t to store the TLen
-
-static FORCE_INLINE char *memRowEnd(SMemRow row) {
-  if (isDataRow(row)) {
-    return (char *)dataRowEnd(memRowDataBody(row));
-  } else {
-    return (char *)kvRowEnd(memRowKvBody(row));
-  }
-}
-
-#define memRowDataVersion(r)     dataRowVersion(memRowDataBody(r))
-#define memRowKvVersion(r)       (*(int16_t *)POINTER_SHIFT(r, TD_MEM_ROW_TYPE_SIZE))
-#define memRowVersion(r)         (isDataRow(r) ? memRowDataVersion(r) : memRowKvVersion(r))  // schema version
-#define memRowSetKvVersion(r, v) (memRowKvVersion(r) = (v))
-#define memRowTuple(r)           (isDataRow(r) ? dataRowTuple(memRowDataBody(r)) : kvRowValues(memRowKvBody(r)))
-
-#define memRowTKey(r) (isDataRow(r) ? dataRowTKey(memRowDataBody(r)) : kvRowTKey(memRowKvBody(r)))
-#define memRowKey(r)  (isDataRow(r) ? dataRowKey(memRowDataBody(r)) : kvRowKey(memRowKvBody(r)))
-#define memRowKeys(r) (isDataRow(r) ? dataRowTuple(memRowDataBody(r)) : kvRowKeys(memRowKvBody(r)))
-#define memRowSetTKey(r, k)                 \
-  do {                                      \
-    if (isDataRow(r)) {                     \
-      dataRowTKey(memRowDataBody(r)) = (k); \
-    } else {                                \
-      kvRowTKey(memRowKvBody(r)) = (k);     \
-    }                                       \
-  } while (0)
-
-#define memRowSetLen(r, l)          (isDataRow(r) ? memRowDataLen(r) = (l) : memRowKvLen(r) = (l))
-#define memRowSetVersion(r, v)      (isDataRow(r) ? dataRowSetVersion(memRowDataBody(r), v) : memRowSetKvVersion(r, v))
-#define memRowCpy(dst, r)           memcpy((dst), (r), memRowTLen(r))
-#define memRowMaxBytesFromSchema(s) (schemaTLen(s) + TD_MEM_ROW_DATA_HEAD_SIZE)
-#define memRowDeleted(r)            TKEY_IS_DELETED(memRowTKey(r))
-
-SMemRow tdMemRowDup(SMemRow row);
-void    tdAppendMemRowToDataCol(SMemRow row, STSchema *pSchema, SDataCols *pCols, bool forceSetNull);
-
-// NOTE: offset here including the header size
-static FORCE_INLINE void *tdGetMemRowDataOfCol(void *row, int16_t colId, int8_t colType, uint16_t offset) {
-  if (isDataRow(row)) {
-    return tdGetRowDataOfCol(memRowDataBody(row), colType, offset);
-  } else {
-    return tdGetKVRowValOfCol(memRowKvBody(row), colId);
-  }
-}
-
-/**
- * NOTE:
- *  1. Applicable to scan columns one by one
- *  2. offset here including the header size
- */
-static FORCE_INLINE void *tdGetMemRowDataOfColEx(void *row, int16_t colId, int8_t colType, int32_t offset,
-                                                 int32_t *kvNIdx) {
-  if (isDataRow(row)) {
-    return tdGetRowDataOfCol(memRowDataBody(row), colType, offset);
-  } else {
-    return tdGetKVRowValOfColEx(memRowKvBody(row), colId, kvNIdx);
-  }
-}
-
-static FORCE_INLINE int32_t tdAppendMemRowColVal(SMemRow row, const void *value, bool isCopyVarData, int16_t colId,
-                                             int8_t type, int32_t offset) {
-  if (isDataRow(row)) {
-    tdAppendDataColVal(memRowDataBody(row), value, isCopyVarData, type, offset);
-  } else {
-    tdAppendKvColVal(memRowKvBody(row), value, isCopyVarData, colId, type, offset);
-  }
-  return 0;
-}
-
-// make sure schema->flen appended for SDataRow
-static FORCE_INLINE int32_t tdGetColAppendLen(uint8_t rowType, const void *value, int8_t colType) {
-  int32_t len = 0;
-  if (IS_VAR_DATA_TYPE(colType)) {
-    len += varDataTLen(value);
-    if (rowType == SMEM_ROW_KV) {
-      len += sizeof(SColIdx);
-    }
-  } else {
-    if (rowType == SMEM_ROW_KV) {
-      len += TYPE_BYTES[colType];
-      len += sizeof(SColIdx);
-    }
-  }
-  return len;
-}
-
-typedef struct {
-  int16_t colId;
-  uint8_t colType;
-  char *  colVal;
-} SColInfo;
-
-static FORCE_INLINE void setSColInfo(SColInfo *colInfo, int16_t colId, uint8_t colType, char *colVal) {
-  colInfo->colId = colId;
-  colInfo->colType = colType;
-  colInfo->colVal = colVal;
-}
-
-SMemRow mergeTwoMemRows(void *buffer, SMemRow row1, SMemRow row2, STSchema *pSchema1, STSchema *pSchema2);
-#endif
 
 #ifdef __cplusplus
 }

include/common/tglobal.h

@@ -121,6 +121,9 @@ extern char     tsCompressor[];
 extern int32_t  tsDiskCfgNum;
 extern SDiskCfg tsDiskCfg[];
 
+// udf
+extern bool  tsStartUdfd;
+
 // internal
 extern int32_t tsTransPullupInterval;
 extern int32_t tsMqRebalanceInterval;

include/common/tmsg.h

@@ -99,7 +99,7 @@ typedef enum _mgmt_table {
   TSDB_MGMT_TABLE_VGROUP,
   TSDB_MGMT_TABLE_TOPICS,
   TSDB_MGMT_TABLE_CONSUMERS,
-  TSDB_MGMT_TABLE_SUBSCRIBES,
+  TSDB_MGMT_TABLE_SUBSCRIPTIONS,
   TSDB_MGMT_TABLE_TRANS,
   TSDB_MGMT_TABLE_SMAS,
   TSDB_MGMT_TABLE_CONFIGS,
@@ -131,12 +131,10 @@ typedef enum _mgmt_table {
 #define TSDB_ALTER_USER_SUPERUSER       0x2
 #define TSDB_ALTER_USER_ADD_READ_DB     0x3
 #define TSDB_ALTER_USER_REMOVE_READ_DB  0x4
-#define TSDB_ALTER_USER_CLEAR_READ_DB   0x5
-#define TSDB_ALTER_USER_ADD_WRITE_DB    0x6
-#define TSDB_ALTER_USER_REMOVE_WRITE_DB 0x7
-#define TSDB_ALTER_USER_CLEAR_WRITE_DB  0x8
-#define TSDB_ALTER_USER_ADD_ALL_DB      0x9
-#define TSDB_ALTER_USER_REMOVE_ALL_DB   0xA
+#define TSDB_ALTER_USER_ADD_WRITE_DB    0x5
+#define TSDB_ALTER_USER_REMOVE_WRITE_DB 0x6
+#define TSDB_ALTER_USER_ADD_ALL_DB      0x7
+#define TSDB_ALTER_USER_REMOVE_ALL_DB   0x8
 
 #define TSDB_ALTER_USER_PRIVILEGES 0x2
 
@@ -330,6 +328,9 @@ typedef struct {
   int8_t  alterType;
   int32_t numOfFields;
   SArray* pFields;
+  int32_t ttl;
+  int32_t commentLen;
+  char*   comment;
 } SMAlterStbReq;
 
 int32_t tSerializeSMAlterStbReq(void* buf, int32_t bufLen, SMAlterStbReq* pReq);
@@ -1457,7 +1458,7 @@ typedef struct {
 static FORCE_INLINE SMqRebInfo* tNewSMqRebSubscribe(const char* key) {
   SMqRebInfo* pRebInfo = (SMqRebInfo*)taosMemoryCalloc(1, sizeof(SMqRebInfo));
   if (pRebInfo == NULL) {
-    goto _err;
+    return NULL;
   }
   strcpy(pRebInfo->key, key);
   pRebInfo->lostConsumers = taosArrayInit(0, sizeof(int64_t));
@@ -1492,7 +1493,7 @@ typedef struct {
 } SMVSubscribeRsp;
 
 typedef struct {
-  char   name[TSDB_TABLE_FNAME_LEN];
+  char   name[TSDB_TOPIC_FNAME_LEN];
   int8_t igNotExists;
 } SMDropTopicReq;
 

include/common/tmsgcb.h

@@ -57,6 +57,7 @@ typedef struct {
   RegisterBrokenLinkArgFp registerBrokenLinkArgFp;
   ReleaseHandleFp         releaseHandleFp;
   ReportStartup           reportStartupFp;
+  void*                   clientRpc;
 } SMsgCb;
 
 void    tmsgSetDefaultMsgCb(const SMsgCb* pMsgCb);

include/common/tname.h

@@ -16,8 +16,8 @@
 #ifndef _TD_COMMON_NAME_H_
 #define _TD_COMMON_NAME_H_
 
-#include "tdef.h"
 #include "tarray.h"
+#include "tdef.h"
 
 #ifdef __cplusplus
 extern "C" {
@@ -65,19 +65,19 @@ bool tNameDBNameEqual(SName* left, SName* right);
 
 typedef struct {
   // input
-  SArray         *tags;         // element is SSmlKV
-  const char     *sTableName;   // super table name
-  uint8_t        sTableNameLen; // the length of super table name
+  SArray*     tags;           // element is SSmlKv
+  const char* sTableName;     // super table name
+  uint8_t     sTableNameLen;  // the length of super table name
 
   // output
-  char           *childTableName;   // must have size of TSDB_TABLE_NAME_LEN;
-  uint64_t       uid;               // child table uid, may be useful
+  char*    childTableName;  // must have size of TSDB_TABLE_NAME_LEN;
+  uint64_t uid;             // child table uid, may be useful
 } RandTableName;
 
-void buildChildTableName(RandTableName *rName);
+void buildChildTableName(RandTableName* rName);
 
 #ifdef __cplusplus
 }
 #endif
 
-#endif  /*_TD_COMMON_NAME_H_*/
+#endif /*_TD_COMMON_NAME_H_*/

include/libs/executor/executor.h

@@ -22,6 +22,7 @@ extern "C" {
 
 #include "query.h"
 #include "tcommon.h"
+#include "tmsgcb.h"
 
 typedef void* qTaskInfo_t;
 typedef void* DataSinkHandle;
@@ -29,11 +30,12 @@ struct SRpcMsg;
 struct SSubplan;
 
 typedef struct SReadHandle {
-  void* reader;
-  void* meta;
-  void* config;
-  void* vnode;
-  void* mnd;
+  void*   reader;
+  void*   meta;
+  void*   config;
+  void*   vnode;
+  void*   mnd;
+  SMsgCb* pMsgCb;
 } SReadHandle;
 
 #define STREAM_DATA_TYPE_SUBMIT_BLOCK 0x1

include/libs/function/function.h

@@ -336,6 +336,17 @@ int32_t udfcOpen();
  */
 int32_t udfcClose();
 
+/**
+ * start udfd that serves udf function invocation under dnode startDnodeId
+ * @param startDnodeId
+ * @return
+ */
+int32_t udfStartUdfd(int32_t startDnodeId);
+/**
+ * stop udfd
+ * @return
+ */
+int32_t udfStopUdfd();
 #ifdef __cplusplus
 }
 #endif

include/libs/function/functionMgt.h

@@ -31,13 +31,13 @@ typedef enum EFunctionType {
   FUNCTION_TYPE_ELAPSED,
   FUNCTION_TYPE_IRATE,
   FUNCTION_TYPE_LAST_ROW,
-  FUNCTION_TYPE_LEASTSQUARES,
   FUNCTION_TYPE_MAX,
   FUNCTION_TYPE_MIN,
   FUNCTION_TYPE_MODE,
   FUNCTION_TYPE_PERCENTILE,
   FUNCTION_TYPE_SPREAD,
   FUNCTION_TYPE_STDDEV,
+  FUNCTION_TYPE_LEASTSQUARES,
   FUNCTION_TYPE_SUM,
   FUNCTION_TYPE_TWA,
   FUNCTION_TYPE_HISTOGRAM,
@@ -55,6 +55,8 @@ typedef enum EFunctionType {
   FUNCTION_TYPE_TAIL,
   FUNCTION_TYPE_TOP,
   FUNCTION_TYPE_UNIQUE,
+  FUNCTION_TYPE_STATE_COUNT,
+  FUNCTION_TYPE_STATE_DURATION,
 
   // math function
   FUNCTION_TYPE_ABS = 1000,
@@ -154,6 +156,7 @@ bool fmIsWindowClauseFunc(int32_t funcId);
 bool fmIsSpecialDataRequiredFunc(int32_t funcId);
 bool fmIsDynamicScanOptimizedFunc(int32_t funcId);
 bool fmIsMultiResFunc(int32_t funcId);
+bool fmIsRepeatScanFunc(int32_t funcId);
 bool fmIsUserDefinedFunc(int32_t funcId);
 
 typedef enum EFuncDataRequired {
@@ -170,7 +173,7 @@ int32_t fmGetScalarFuncExecFuncs(int32_t funcId, SScalarFuncExecFuncs* pFpSet);
 int32_t fmGetUdafExecFuncs(int32_t funcId, SFuncExecFuncs* pFpSet);
 int32_t fmSetInvertFunc(int32_t funcId, SFuncExecFuncs* pFpSet);
 int32_t fmSetNormalFunc(int32_t funcId, SFuncExecFuncs* pFpSet);
-bool fmIsInvertible(int32_t funcId);
+bool    fmIsInvertible(int32_t funcId);
 
 #ifdef __cplusplus
 }

include/libs/nodes/nodes.h

@@ -179,6 +179,7 @@ typedef enum ENodeType {
   QUERY_NODE_KILL_CONNECTION_STMT,
   QUERY_NODE_KILL_QUERY_STMT,
   QUERY_NODE_KILL_TRANSACTION_STMT,
+  QUERY_NODE_QUERY,
 
   // logic plan node
   QUERY_NODE_LOGIC_PLAN_SCAN,
@@ -206,6 +207,7 @@ typedef enum ENodeType {
   QUERY_NODE_PHYSICAL_PLAN_EXCHANGE,
   QUERY_NODE_PHYSICAL_PLAN_SORT,
   QUERY_NODE_PHYSICAL_PLAN_INTERVAL,
+  QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL,
   QUERY_NODE_PHYSICAL_PLAN_FILL,
   QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW,
   QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW,

include/libs/nodes/plannodes.h

@@ -274,6 +274,8 @@ typedef struct SIntervalPhysiNode {
   int8_t           slidingUnit;
 } SIntervalPhysiNode;
 
+typedef SIntervalPhysiNode SStreamIntervalPhysiNode;
+
 typedef struct SFillPhysiNode {
   SPhysiNode  node;
   EFillMode   mode;

include/libs/nodes/querynodes.h

@@ -233,6 +233,7 @@ typedef struct SSelectStmt {
   uint8_t     precision;
   bool        isEmptyResult;
   bool        hasAggFuncs;
+  bool        hasRepeatScanFuncs;
   bool        isTimeOrderQuery;
 } SSelectStmt;
 
@@ -294,6 +295,37 @@ typedef struct SExplainStmt {
   SNode*           pQuery;
 } SExplainStmt;
 
+typedef struct SCmdMsgInfo {
+  int16_t msgType;
+  SEpSet  epSet;
+  void*   pMsg;
+  int32_t msgLen;
+  void*   pExtension;  // todo remove it soon
+} SCmdMsgInfo;
+
+typedef enum EQueryExecMode {
+  QUERY_EXEC_MODE_LOCAL = 1,
+  QUERY_EXEC_MODE_RPC,
+  QUERY_EXEC_MODE_SCHEDULE,
+  QUERY_EXEC_MODE_EMPTY_RESULT
+} EQueryExecMode;
+
+typedef struct SQuery {
+  ENodeType      type;
+  EQueryExecMode execMode;
+  bool           haveResultSet;
+  SNode*         pRoot;
+  int32_t        numOfResCols;
+  SSchema*       pResSchema;
+  int8_t         precision;
+  SCmdMsgInfo*   pCmdMsg;
+  int32_t        msgType;
+  SArray*        pDbList;
+  SArray*        pTableList;
+  bool           showRewrite;
+  int32_t        placeholderNum;
+} SQuery;
+
 void nodesWalkSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeWalker walker, void* pContext);
 void nodesRewriteSelectStmt(SSelectStmt* pSelect, ESqlClause clause, FNodeRewriter rewriter, void* pContext);
 

include/libs/parser/parser.h

@@ -48,36 +48,6 @@ typedef struct SParseContext {
   bool             isSuperUser;
 } SParseContext;
 
-typedef struct SCmdMsgInfo {
-  int16_t msgType;
-  SEpSet  epSet;
-  void*   pMsg;
-  int32_t msgLen;
-  void*   pExtension;  // todo remove it soon
-} SCmdMsgInfo;
-
-typedef enum EQueryExecMode {
-  QUERY_EXEC_MODE_LOCAL = 1,
-  QUERY_EXEC_MODE_RPC,
-  QUERY_EXEC_MODE_SCHEDULE,
-  QUERY_EXEC_MODE_EMPTY_RESULT
-} EQueryExecMode;
-
-typedef struct SQuery {
-  EQueryExecMode execMode;
-  bool           haveResultSet;
-  SNode*         pRoot;
-  int32_t        numOfResCols;
-  SSchema*       pResSchema;
-  int8_t         precision;
-  SCmdMsgInfo*   pCmdMsg;
-  int32_t        msgType;
-  SArray*        pDbList;
-  SArray*        pTableList;
-  bool           showRewrite;
-  int32_t        placeholderNum;
-} SQuery;
-
 int32_t qParseQuerySql(SParseContext* pCxt, SQuery** pQuery);
 bool    isInsertSql(const char* pStr, size_t length);
 
@@ -103,9 +73,10 @@ void    destroyBoundColumnInfo(void* pBoundInfo);
 int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char* dbName, char* msgBuf,
                      int32_t msgBufLen);
 
-void*   smlInitHandle(SQuery *pQuery);
-void    smlDestroyHandle(void *pHandle);
-int32_t smlBindData(void *handle, SArray *tags, SArray *colsFormat, SArray *colsSchema, SArray *cols, bool format, STableMeta *pTableMeta, char *tableName, char *msgBuf, int16_t msgBufLen);
+void*   smlInitHandle(SQuery* pQuery);
+void    smlDestroyHandle(void* pHandle);
+int32_t smlBindData(void* handle, SArray* tags, SArray* colsFormat, SArray* colsSchema, SArray* cols, bool format,
+                    STableMeta* pTableMeta, char* tableName, char* msgBuf, int16_t msgBufLen);
 int32_t smlBuildOutput(void* handle, SHashObj* pVgHash);
 
 #ifdef __cplusplus

include/libs/qcom/query.h

@@ -24,6 +24,7 @@ extern "C" {
 #include "thash.h"
 #include "tlog.h"
 #include "tmsg.h"
+#include "tmsgcb.h"
 
 typedef enum {
   JOB_TASK_STATUS_NULL = 0,
@@ -149,7 +150,8 @@ int32_t cleanupTaskQueue();
  */
 int32_t taosAsyncExec(__async_exec_fn_t execFn, void* execParam, int32_t* code);
 
-int32_t asyncSendMsgToServerExt(void* pTransporter, SEpSet* epSet, int64_t* pTransporterId, const SMsgSendInfo* pInfo, bool persistHandle, void *ctx);
+int32_t asyncSendMsgToServerExt(void* pTransporter, SEpSet* epSet, int64_t* pTransporterId, const SMsgSendInfo* pInfo,
+                                bool persistHandle, void* ctx);
 
 /**
  * Asynchronously send message to server, after the response received, the callback will be incured.
@@ -182,7 +184,7 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
 #define SET_META_TYPE_TABLE(t)      (t) = META_TYPE_TABLE
 #define SET_META_TYPE_BOTH_TABLE(t) (t) = META_TYPE_BOTH_TABLE
 
-#define NEED_CLIENT_RM_TBLMETA_ERROR(_code) ((_code) == TSDB_CODE_TDB_INVALID_TABLE_ID || (_code) == TSDB_CODE_VND_TB_NOT_EXIST)
+#define NEED_CLIENT_RM_TBLMETA_ERROR(_code) ((_code) == TSDB_CODE_PAR_TABLE_NOT_EXIST || (_code) == TSDB_CODE_VND_TB_NOT_EXIST)
 #define NEED_CLIENT_REFRESH_VG_ERROR(_code) ((_code) == TSDB_CODE_VND_HASH_MISMATCH || (_code) == TSDB_CODE_VND_INVALID_VGROUP_ID)
 #define NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code) ((_code) == TSDB_CODE_TDB_TABLE_RECREATED)
 #define NEED_CLIENT_HANDLE_ERROR(_code) (NEED_CLIENT_RM_TBLMETA_ERROR(_code) || NEED_CLIENT_REFRESH_VG_ERROR(_code) || NEED_CLIENT_REFRESH_TBLMETA_ERROR(_code))

include/libs/stream/tstream.h

@@ -16,6 +16,7 @@
 #include "tdatablock.h"
 #include "tmsg.h"
 #include "tmsgcb.h"
+#include "tqueue.h"
 #include "trpc.h"
 
 #ifdef __cplusplus
@@ -154,6 +155,10 @@ struct SStreamTask {
     STaskDispatcherShuffle shuffleDispatcher;
   };
 
+  // msg buffer
+  int32_t     memUsed;
+  STaosQueue* inputQ;
+
   // application storage
   void* ahandle;
 };
@@ -194,6 +199,8 @@ typedef struct {
   SArray* res;  // SArray<SSDataBlock>
 } SStreamSinkReq;
 
+int32_t streamEnqueueData(SStreamTask* pTask, const void* input, int32_t inputType);
+
 int32_t streamExecTask(SStreamTask* pTask, SMsgCb* pMsgCb, const void* input, int32_t inputType, int32_t workId);
 
 #ifdef __cplusplus

include/os/osFile.h

@@ -80,6 +80,7 @@ int64_t taosWriteFile(TdFilePtr pFile, const void *buf, int64_t count);
 void    taosFprintfFile(TdFilePtr pFile, const char *format, ...);
 
 int64_t taosGetLineFile(TdFilePtr pFile, char ** __restrict ptrBuf);
+int64_t taosGetsFile(TdFilePtr pFile, int32_t maxSize, char *__restrict buf);
 
 int32_t taosEOFFile(TdFilePtr pFile);
  

include/os/osMemory.h

@@ -35,6 +35,7 @@ void *taosMemoryRealloc(void *ptr, int32_t size);
 void *taosMemoryStrDup(void *ptr);
 void taosMemoryFree(void *ptr);
 int32_t taosMemorySize(void *ptr);
+void taosPrintBackTrace();
 
 #define taosMemoryFreeClear(ptr)   \
   do {                             \

include/util/taoserror.h

@@ -88,6 +88,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_CFG_NOT_FOUND                 TAOS_DEF_ERROR_CODE(0, 0x0114)
 #define TSDB_CODE_REPEAT_INIT                   TAOS_DEF_ERROR_CODE(0, 0x0115)
 #define TSDB_CODE_DUP_KEY                       TAOS_DEF_ERROR_CODE(0, 0x0116)
+#define TSDB_CODE_NEED_RETRY                    TAOS_DEF_ERROR_CODE(0, 0x0117)
 
 #define TSDB_CODE_REF_NO_MEMORY                 TAOS_DEF_ERROR_CODE(0, 0x0140)
 #define TSDB_CODE_REF_FULL                      TAOS_DEF_ERROR_CODE(0, 0x0141)
@@ -632,6 +633,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN    TAOS_DEF_ERROR_CODE(0, 0x2642)
 #define TSDB_CODE_PAR_INVALID_TAGS_NUM          TAOS_DEF_ERROR_CODE(0, 0x2643)
 #define TSDB_CODE_PAR_PERMISSION_DENIED         TAOS_DEF_ERROR_CODE(0, 0x2644)
+#define TSDB_CODE_PAR_INVALID_STREAM_QUERY      TAOS_DEF_ERROR_CODE(0, 0x2645)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)

include/util/tdef.h

@@ -86,10 +86,9 @@ extern const int32_t TYPE_BYTES[15];
 #define TS_PATH_DELIMITER "."
 #define TS_ESCAPE_CHAR    '`'
 
-
-#define TSDB_TIME_PRECISION_MILLI 0
-#define TSDB_TIME_PRECISION_MICRO 1
-#define TSDB_TIME_PRECISION_NANO  2
+#define TSDB_TIME_PRECISION_MILLI   0
+#define TSDB_TIME_PRECISION_MICRO   1
+#define TSDB_TIME_PRECISION_NANO    2
 #define TSDB_TIME_PRECISION_HOURS   3
 #define TSDB_TIME_PRECISION_MINUTES 4
 #define TSDB_TIME_PRECISION_SECONDS 5
@@ -249,7 +248,6 @@ typedef enum ELogicConditionType {
 #define TSDB_SHOW_SQL_LEN       512
 #define TSDB_SLOW_QUERY_SQL_LEN 512
 #define TSDB_SHOW_SUBQUERY_LEN  1000
-#define TSDB_SHOW_LIST_LEN      1000
 
 #define TSDB_TRANS_STAGE_LEN 12
 #define TSDB_TRANS_TYPE_LEN  16
@@ -376,9 +374,9 @@ typedef enum ELogicConditionType {
  * 1. ordinary sub query for select * from super_table
  * 2. all sqlobj generated by createSubqueryObj with this flag
  */
-#define TSDB_QUERY_TYPE_INSERT           0x100u  // insert type
-#define TSDB_QUERY_TYPE_FILE_INSERT      0x400u   // insert data from file
-#define TSDB_QUERY_TYPE_STMT_INSERT      0x800u   // stmt insert type
+#define TSDB_QUERY_TYPE_INSERT      0x100u  // insert type
+#define TSDB_QUERY_TYPE_FILE_INSERT 0x400u  // insert data from file
+#define TSDB_QUERY_TYPE_STMT_INSERT 0x800u  // stmt insert type
 
 #define TSDB_QUERY_HAS_TYPE(x, _type)   (((x) & (_type)) != 0)
 #define TSDB_QUERY_SET_TYPE(x, _type)   ((x) |= (_type))

source/client/inc/clientInt.h

@@ -254,7 +254,7 @@ extern int (*handleRequestRspFp[TDMT_MAX])(void*, const SDataBuf* pMsg, int32_t
 int           genericRspCallback(void* param, const SDataBuf* pMsg, int32_t code);
 SMsgSendInfo* buildMsgInfoImpl(SRequestObj* pReqObj);
 
-void*    createTscObj(const char* user, const char* auth, const char* db, SAppInstInfo* pAppInfo);
+void*    createTscObj(const char* user, const char* auth, const char* db, int32_t connType, SAppInstInfo* pAppInfo);
 void     destroyTscObj(void* pObj);
 STscObj* acquireTscObj(int64_t rid);
 int32_t  releaseTscObj(int64_t rid);
@@ -310,6 +310,7 @@ void hbMgrInitMqHbRspHandle();
 SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, int32_t code, bool keepQuery);
 int32_t      getQueryPlan(SRequestObj* pRequest, SQuery* pQuery, SArray** pNodeList);
 int32_t      scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList);
+int32_t refreshMeta(STscObj* pTscObj, SRequestObj* pRequest);
 
 #ifdef __cplusplus
 }

source/client/inc/clientStmt.h

@@ -60,6 +60,7 @@ typedef struct SStmtBindInfo {
   int32_t      sBindRowNum;
   int32_t      sBindLastIdx;
   int8_t       tbType;
+  bool         tagsCached;
   void*        boundTags;  
   char*        tbName;
   SName        sname;

source/client/src/clientEnv.c

@@ -131,7 +131,7 @@ void destroyTscObj(void *pObj) {
   taosMemoryFreeClear(pTscObj);
 }
 
-void *createTscObj(const char *user, const char *auth, const char *db, SAppInstInfo *pAppInfo) {
+void *createTscObj(const char *user, const char *auth, const char *db, int32_t connType, SAppInstInfo *pAppInfo) {
   STscObj *pObj = (STscObj *)taosMemoryCalloc(1, sizeof(STscObj));
   if (NULL == pObj) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
@@ -145,6 +145,7 @@ void *createTscObj(const char *user, const char *auth, const char *db, SAppInstI
     return NULL;
   }
 
+  pObj->connType = connType;
   pObj->pAppInfo = pAppInfo;
   tstrncpy(pObj->user, user, sizeof(pObj->user));
   memcpy(pObj->pass, auth, TSDB_PASSWORD_LEN);

source/client/src/clientImpl.c

@@ -25,7 +25,7 @@
 #include "tref.h"
 
 static int32_t       initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet);
-static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest, int8_t connType);
+static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest);
 static void          destroySendMsgInfo(SMsgSendInfo* pMsgBody);
 
 static bool stringLengthCheck(const char* str, size_t maxsize) {
@@ -491,7 +491,7 @@ int initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSe
 
 STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __taos_async_fn_t fp, void* param,
                          SAppInstInfo* pAppInfo, int connType) {
-  STscObj* pTscObj = createTscObj(user, auth, db, pAppInfo);
+  STscObj* pTscObj = createTscObj(user, auth, db, connType, pAppInfo);
   if (NULL == pTscObj) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
     return pTscObj;
@@ -504,7 +504,7 @@ STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __t
     return NULL;
   }
 
-  SMsgSendInfo* body = buildConnectMsg(pRequest, connType);
+  SMsgSendInfo* body = buildConnectMsg(pRequest);
 
   int64_t transporterId = 0;
   asyncSendMsgToServer(pTscObj->pAppInfo->pTransporter, &pTscObj->pAppInfo->mgmtEp.epSet, &transporterId, body);
@@ -527,7 +527,7 @@ STscObj* taosConnectImpl(const char* user, const char* auth, const char* db, __t
   return pTscObj;
 }
 
-static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest, int8_t connType) {
+static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest) {
   SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
   if (pMsgSendInfo == NULL) {
     terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
@@ -550,9 +550,10 @@ static SMsgSendInfo* buildConnectMsg(SRequestObj* pRequest, int8_t connType) {
   }
   taosMemoryFreeClear(db);
 
-  connectReq.connType = connType;
-  connectReq.pid = htonl(appInfo.pid);
+  connectReq.connType  = pObj->connType;
+  connectReq.pid       = htonl(appInfo.pid);
   connectReq.startTime = htobe64(appInfo.startTime);
+
   tstrncpy(connectReq.app, appInfo.appName, sizeof(connectReq.app));
   tstrncpy(connectReq.user, pObj->user, sizeof(connectReq.user));
   tstrncpy(connectReq.passwd, pObj->pass, sizeof(connectReq.passwd));

source/client/src/clientMain.c

@@ -563,7 +563,11 @@ const char *taos_get_server_info(TAOS *taos) {
 }
 
 void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param) {
-  // TODO
+  if (taos == NULL || sql == NULL) {
+    // todo directly call fp
+  }
+
+  taos_query_l(taos, sql, (int32_t) strlen(sql));
 }
 
 void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {

source/client/src/clientSml.c

@@ -401,7 +401,7 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) {
 
     code = catalogGetSTableMeta(info->pCatalog, info->taos->pAppInfo->pTransporter, &ep, &pName, &pTableMeta);
 
-    if (code == TSDB_CODE_TDB_INVALID_TABLE_ID || code == TSDB_CODE_MND_INVALID_STB) {
+    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_MND_INVALID_STB) {
       SSchemaAction schemaAction = {.action = SCHEMA_ACTION_CREATE_STABLE};
       memcpy(schemaAction.createSTable.sTableName, superTable, superTableLen);
       schemaAction.createSTable.tags = sTableData->tags;

source/client/src/clientStmt.c

@@ -29,6 +29,11 @@ int32_t stmtSwitchStatus(STscStmt* pStmt, STMT_STATUS newStatus) {
       if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND_COL)) {
         code = TSDB_CODE_TSC_STMT_API_ERROR;
       }
+/*      
+      if ((pStmt->sql.type == STMT_TYPE_MULTI_INSERT) && ()) {
+        code = TSDB_CODE_TSC_STMT_API_ERROR;
+      }
+*/
       break;
     case STMT_BIND_COL:
       if (STMT_STATUS_EQ(INIT) || STMT_STATUS_EQ(BIND)) {
@@ -123,6 +128,7 @@ int32_t stmtSetBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags) {
   pStmt->bInfo.tbSuid = pTableMeta->suid;
   pStmt->bInfo.tbType = pTableMeta->tableType;
   pStmt->bInfo.boundTags = tags;
+  pStmt->bInfo.tagsCached = false;
 
   return TSDB_CODE_SUCCESS;
 }
@@ -207,8 +213,6 @@ int32_t stmtParseSql(STscStmt* pStmt) {
       STMT_ERR_RET(TSDB_CODE_TSC_STMT_CLAUSE_ERROR);
   }
 
-  STMT_ERR_RET(stmtCacheBlock(pStmt));
-
   return TSDB_CODE_SUCCESS;
 }
 
@@ -219,8 +223,10 @@ int32_t stmtCleanBindInfo(STscStmt* pStmt) {
   pStmt->bInfo.needParse = true;
 
   taosMemoryFreeClear(pStmt->bInfo.tbName);
-  destroyBoundColumnInfo(pStmt->bInfo.boundTags);
-  taosMemoryFreeClear(pStmt->bInfo.boundTags);
+  if (!pStmt->bInfo.tagsCached) {
+    destroyBoundColumnInfo(pStmt->bInfo.boundTags);
+    taosMemoryFreeClear(pStmt->bInfo.boundTags);
+  }
 
   return TSDB_CODE_SUCCESS;
 }
@@ -275,6 +281,7 @@ int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
 
     qDestroyStmtDataBlock(pCache->pDataBlock);
     destroyBoundColumnInfo(pCache->boundTags);
+    taosMemoryFreeClear(pCache->boundTags);
     
     pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
   }
@@ -302,7 +309,15 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
 
   STableMeta *pTableMeta = NULL;
   SEpSet ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
-  STMT_ERR_RET(catalogGetTableMeta(pStmt->pCatalog, pStmt->taos->pAppInfo->pTransporter, &ep, &pStmt->bInfo.sname, &pTableMeta));
+  int32_t code = catalogGetTableMeta(pStmt->pCatalog, pStmt->taos->pAppInfo->pTransporter, &ep, &pStmt->bInfo.sname, &pTableMeta);
+  if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
+    STMT_ERR_RET(stmtCleanBindInfo(pStmt));
+    
+    return TSDB_CODE_SUCCESS;
+  }
+
+  STMT_ERR_RET(code);
+  
   uint64_t uid = pTableMeta->uid;
   uint64_t suid = pTableMeta->suid;
   int8_t tableType = pTableMeta->tableType;
@@ -328,6 +343,7 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
     pStmt->bInfo.tbSuid = suid;
     pStmt->bInfo.tbType = tableType;
     pStmt->bInfo.boundTags = pCache->boundTags;
+    pStmt->bInfo.tagsCached = true;
 
     return TSDB_CODE_SUCCESS;
   }
@@ -340,6 +356,7 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
     pStmt->bInfo.tbSuid = suid;
     pStmt->bInfo.tbType = tableType;
     pStmt->bInfo.boundTags = pCache->boundTags;
+    pStmt->bInfo.tagsCached = true;
 
     STableDataBlocks* pNewBlock = NULL;
     STMT_ERR_RET(qRebuildStmtDataBlock(&pNewBlock, pCache->pDataBlock));
@@ -448,10 +465,12 @@ int stmtSetTbTags(TAOS_STMT *stmt, TAOS_MULTI_BIND *tags) {
 
   STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_SETTAGS));
 
-  if (pStmt->bInfo.needParse) {
-    STMT_ERR_RET(stmtParseSql(pStmt));
+  if (!pStmt->bInfo.needParse) {
+    return TSDB_CODE_SUCCESS;
   }
 
+  STMT_ERR_RET(stmtParseSql(pStmt));
+
   STableDataBlocks **pDataBlock = (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, (const char*)&pStmt->bInfo.tbUid, sizeof(pStmt->bInfo.tbUid));
   if (NULL == pDataBlock) {
     tscError("table uid %" PRIx64 "not found in exec blockHash", pStmt->bInfo.tbUid);
@@ -501,8 +520,6 @@ int32_t stmtFetchColFields(STscStmt* pStmt, int32_t *fieldNum, TAOS_FIELD** fiel
 int stmtBindBatch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind, int32_t colIdx) {
   STscStmt* pStmt = (STscStmt*)stmt;
 
-  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
-
   if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 && STMT_TYPE_MULTI_INSERT != pStmt->sql.type) {
     pStmt->bInfo.needParse = false;
   }
@@ -520,6 +537,8 @@ int stmtBindBatch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind, int32_t colIdx) {
     STMT_ERR_RET(stmtParseSql(pStmt));
   }
 
+  STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_BIND));
+
   if (STMT_TYPE_QUERY == pStmt->sql.type) {
     if (NULL == pStmt->sql.pQueryPlan) {
       STMT_ERR_RET(getQueryPlan(pStmt->exec.pRequest, pStmt->sql.pQuery, &pStmt->sql.nodeList));
@@ -586,6 +605,16 @@ int stmtExec(TAOS_STMT *stmt) {
     STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->exec.pVgHash, pStmt->exec.pBlockHash));
     launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, TSDB_CODE_SUCCESS, true);
   }
+
+  if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
+    code = refreshMeta(pStmt->exec.pRequest->pTscObj, pStmt->exec.pRequest);
+    if (code) {
+      pStmt->exec.pRequest->code = code;
+    } else {
+      STMT_ERR_RET(stmtResetStmt(pStmt));
+      STMT_ERR_RET(TSDB_CODE_NEED_RETRY);
+    }
+  }
   
   STMT_ERR_JRET(pStmt->exec.pRequest->code);
 
@@ -613,13 +642,11 @@ int stmtClose(TAOS_STMT *stmt) {
 const char *stmtErrstr(TAOS_STMT *stmt) {
   STscStmt* pStmt = (STscStmt*)stmt;
 
-  if (stmt == NULL) {
+  if (stmt == NULL || NULL == pStmt->exec.pRequest) {
     return (char*) tstrerror(terrno);
   }
 
-  if (pStmt->exec.pRequest) {
-    pStmt->exec.pRequest->code = terrno;
-  }
+  pStmt->exec.pRequest->code = terrno;
 
   return taos_errstr(pStmt->exec.pRequest);
 }

source/client/src/tmq.c

@@ -187,7 +187,7 @@ typedef struct {
 
 tmq_conf_t* tmq_conf_new() {
   tmq_conf_t* conf = taosMemoryCalloc(1, sizeof(tmq_conf_t));
-  conf->autoCommit = false;
+  conf->autoCommit = true;
   conf->autoCommitInterval = 5000;
   conf->resetOffset = TMQ_CONF__RESET_OFFSET__EARLIEAST;
   return conf;
@@ -395,7 +395,7 @@ tmq_resp_err_t tmq_subscription(tmq_t* tmq, tmq_list_t** topics) {
   }
   for (int i = 0; i < taosArrayGetSize(tmq->clientTopics); i++) {
     SMqClientTopic* topic = taosArrayGet(tmq->clientTopics, i);
-    tmq_list_append(*topics, topic->topicName);
+    tmq_list_append(*topics, strchr(topic->topicName, '.') + 1);
   }
   return TMQ_RESP_ERR__SUCCESS;
 }
@@ -1307,7 +1307,18 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t wait_time) {
 }
 
 tmq_resp_err_t tmq_consumer_close(tmq_t* tmq) {
-  // TODO
+  if (tmq->status == TMQ_CONSUMER_STATUS__READY) {
+    tmq_list_t*    lst = tmq_list_new();
+    tmq_resp_err_t rsp = tmq_subscribe(tmq, lst);
+    tmq_list_destroy(lst);
+    if (rsp == TMQ_RESP_ERR__SUCCESS) {
+      // TODO: free resources
+      return TMQ_RESP_ERR__SUCCESS;
+    } else {
+      return TMQ_RESP_ERR__FAIL;
+    }
+  }
+  // TODO: free resources
   return TMQ_RESP_ERR__SUCCESS;
 }
 

source/common/src/systable.c

@@ -14,9 +14,9 @@
  */
 
 #include "systable.h"
+#include "taos.h"
 #include "tdef.h"
 #include "types.h"
-#include "taos.h"
 
 #define SYSTABLE_SCH_TABLE_NAME_LEN ((TSDB_TABLE_NAME_LEN - 1) + VARSTR_HEADER_SIZE)
 #define SYSTABLE_SCH_DB_NAME_LEN    ((TSDB_DB_NAME_LEN - 1) + VARSTR_HEADER_SIZE)
@@ -262,10 +262,10 @@ static const SSysDbTableSchema topicSchema[] = {
 
 static const SSysDbTableSchema consumerSchema[] = {
     {.name = "consumer_id", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
-    {.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+    {.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 = "status", .bytes = 20 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
-    {.name = "topics", .bytes = TSDB_SHOW_LIST_LEN + 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},
     {.name = "end_point", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "up_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
@@ -274,8 +274,8 @@ static const SSysDbTableSchema consumerSchema[] = {
 };
 
 static const SSysDbTableSchema subscriptionSchema[] = {
-    {.name = "topic_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
-    {.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "topic_name", .bytes = TSDB_TOPIC_FNAME_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "consumer_group", .bytes = TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "consumer_id", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
 };

source/common/src/tdata.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/src/tdatablock.c

@@ -493,7 +493,7 @@ SSDataBlock* blockDataExtractBlock(SSDataBlock* pBlock, int32_t startIndex, int3
     for (int32_t j = startIndex; j < (startIndex + rowCount); ++j) {
       bool isNull = false;
       if (pBlock->pBlockAgg == NULL) {
-        isNull = colDataIsNull_s(pColData, pBlock->info.rows);
+        isNull = colDataIsNull_s(pColData, j);
       } else {
         isNull = colDataIsNull(pColData, pBlock->info.rows, j, pBlock->pBlockAgg[i]);
       }

source/common/src/tdataformat.c

@@ -129,50 +129,6 @@ void *tdDecodeSchema(void *buf, STSchema **pRSchema) {
   return buf;
 }
 
-#if 0
-int32_t tEncodeSTColumn(SCoder *pEncoder, const STColumn *pCol) {
-  if (tEncodeI16(pEncoder, pCol->colId) < 0) return -1;
-  if (tEncodeI8(pEncoder, pCol->type) < 0) return -1;
-  if (tEncodeI8(pEncoder, pCol->sma) < 0) return -1;
-  if (tEncodeI32(pEncoder, pCol->bytes) < 0) return -1;
-  if (tEncodeI32(pEncoder, pCol->offset) < 0) return -1;
-  return pEncoder->pos;
-}
-
-int32_t tDecodeSTColumn(SCoder *pDecoder, STColumn *pCol) {
-  if (tDecodeI16(pDecoder, &pCol->colId) < 0) return -1;
-  if (tDecodeI8(pDecoder, &pCol->type) < 0) return -1;
-  if (tDecodeI8(pDecoder, &pCol->sma) < 0) return -1;
-  if (tDecodeI32(pDecoder, &pCol->bytes) < 0) return -1;
-  if (tDecodeI32(pDecoder, &pCol->offset) < 0) return -1;
-  return 0;
-}
-
-int32_t tEncodeSchema(SCoder *pEncoder, const STSchema *pSchema) {
-  if (tEncodeI32(pEncoder, pSchema->numOfCols) < 0) return -1;
-  if (tEncodeI16(pEncoder, pSchema->version) < 0) return -1;
-  if (tEncodeU16(pEncoder, pSchema->flen) < 0) return -1;
-  if (tEncodeI32(pEncoder, pSchema->vlen) < 0) return -1;
-  if (tEncodeI32(pEncoder, pSchema->tlen) < 0) return -1;
-
-  for (int32_t i = 0; i < schemaNCols(pSchema); i++) {
-    const STColumn *pCol = schemaColAt(pSchema, i);
-    if (tEncodeSTColumn(pEncoder, pCol) < 0) return -1;
-  }
-  return 0;
-}
-
-int32_t tDecodeSchema(SCoder *pDecoder, STSchema *pSchema) {
-  if (tDecodeI32(pDecoder, &pSchema->numOfCols) < 0) return -1;
-  if (tDecodeI16(pDecoder, &pSchema->version) < 0) return -1;
-  if (tDecodeU16(pDecoder, &pSchema->flen) < 0) return -1;
-  if (tDecodeI32(pDecoder, &pSchema->vlen) < 0) return -1;
-  if (tDecodeI32(pDecoder, &pSchema->tlen) < 0) return -1;
-
-  return 0;
-}
-#endif
-
 int tdInitTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version) {
   if (pBuilder == NULL) return -1;
 
@@ -260,49 +216,6 @@ STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder) {
   return pSchema;
 }
 
-#if 0
-/**
- * Initialize a data row
- */
-void tdInitDataRow(SDataRow row, STSchema *pSchema) {
-  dataRowSetLen(row, TD_DATA_ROW_HEAD_SIZE + schemaFLen(pSchema));
-  dataRowSetVersion(row, schemaVersion(pSchema));
-}
-
-SDataRow tdNewDataRowFromSchema(STSchema *pSchema) {
-  int32_t size = dataRowMaxBytesFromSchema(pSchema);
-
-  SDataRow row = taosMemoryMalloc(size);
-  if (row == NULL) return NULL;
-
-  tdInitDataRow(row, pSchema);
-  return row;
-}
-
-/**
- * Free the SDataRow object
- */
-void tdFreeDataRow(SDataRow row) {
-  if (row) taosMemoryFree(row);
-}
-
-SDataRow tdDataRowDup(SDataRow row) {
-  SDataRow trow = taosMemoryMalloc(dataRowLen(row));
-  if (trow == NULL) return NULL;
-
-  dataRowCpy(trow, row);
-  return trow;
-}
-
-SMemRow tdMemRowDup(SMemRow row) {
-  SMemRow trow = taosMemoryMalloc(memRowTLen(row));
-  if (trow == NULL) return NULL;
-
-  memRowCpy(trow, row);
-  return trow;
-}
-#endif
-
 void dataColInit(SDataCol *pDataCol, STColumn *pCol, int maxPoints) {
   pDataCol->type = colType(pCol);
   pDataCol->colId = colColId(pCol);
@@ -312,39 +225,6 @@ void dataColInit(SDataCol *pDataCol, STColumn *pCol, int maxPoints) {
   pDataCol->len = 0;
 }
 
-#if 0
-// value from timestamp should be TKEY here instead of TSKEY
-int dataColAppendVal(SDataCol *pCol, const void *value, int numOfRows, int maxPoints) {
-  ASSERT(pCol != NULL && value != NULL);
-
-  if (isAllRowsNull(pCol)) {
-    if (isNull(value, pCol->type)) {
-      // all null value yet, just return
-      return 0;
-    }
-
-    if (tdAllocMemForCol(pCol, maxPoints) < 0) return -1;
-    if (numOfRows > 0) {
-      // Find the first not null value, fill all previouse values as NULL
-      dataColSetNEleNull(pCol, numOfRows);
-    }
-  }
-
-  if (IS_VAR_DATA_TYPE(pCol->type)) {
-    // set offset
-    pCol->dataOff[numOfRows] = pCol->len;
-    // Copy data
-    memcpy(POINTER_SHIFT(pCol->pData, pCol->len), value, varDataTLen(value));
-    // Update the length
-    pCol->len += varDataTLen(value);
-  } else {
-    ASSERT(pCol->len == TYPE_BYTES[pCol->type] * numOfRows);
-    memcpy(POINTER_SHIFT(pCol->pData, pCol->len), value, pCol->bytes);
-    pCol->len += pCol->bytes;
-  }
-  return 0;
-}
-#endif
 static FORCE_INLINE const void *tdGetColDataOfRowUnsafe(SDataCol *pCol, int row) {
   if (IS_VAR_DATA_TYPE(pCol->type)) {
     return POINTER_SHIFT(pCol->pData, pCol->dataOff[row]);
@@ -361,31 +241,6 @@ bool isNEleNull(SDataCol *pCol, int nEle) {
   return true;
 }
 
-#if 0
-static FORCE_INLINE void dataColSetNullAt(SDataCol *pCol, int index) {
-  if (IS_VAR_DATA_TYPE(pCol->type)) {
-    pCol->dataOff[index] = pCol->len;
-    char *ptr = POINTER_SHIFT(pCol->pData, pCol->len);
-    setVardataNull(ptr, pCol->type);
-    pCol->len += varDataTLen(ptr);
-  } else {
-    setNull(POINTER_SHIFT(pCol->pData, TYPE_BYTES[pCol->type] * index), pCol->type, pCol->bytes);
-    pCol->len += TYPE_BYTES[pCol->type];
-  }
-}
-
-static void dataColSetNEleNull(SDataCol *pCol, int nEle, int8_t bitmapMode) {
-  if (IS_VAR_DATA_TYPE(pCol->type)) {
-    pCol->len = 0;
-    for (int i = 0; i < nEle; ++i) {
-      dataColSetNullAt(pCol, i);
-    }
-  } else {
-    setNullN(pCol->pData, pCol->type, pCol->bytes, nEle);
-    pCol->len = TYPE_BYTES[pCol->type] * nEle;
-  }
-}
-#endif
 void *dataColSetOffset(SDataCol *pCol, int nEle) {
   ASSERT(((pCol->type == TSDB_DATA_TYPE_BINARY) || (pCol->type == TSDB_DATA_TYPE_NCHAR)));
 
@@ -483,42 +338,6 @@ SDataCols *tdFreeDataCols(SDataCols *pCols) {
   return NULL;
 }
 
-#if 0
-SDataCols *tdDupDataCols(SDataCols *pDataCols, bool keepData) {
-  SDataCols *pRet = tdNewDataCols(pDataCols->maxCols, pDataCols->maxPoints);
-  if (pRet == NULL) return NULL;
-
-  pRet->numOfCols = pDataCols->numOfCols;
-  pRet->sversion = pDataCols->sversion;
-  if (keepData) pRet->numOfRows = pDataCols->numOfRows;
-
-  for (int i = 0; i < pDataCols->numOfCols; i++) {
-    pRet->cols[i].type = pDataCols->cols[i].type;
-    pRet->cols[i].bitmap = pDataCols->cols[i].bitmap;
-    pRet->cols[i].colId = pDataCols->cols[i].colId;
-    pRet->cols[i].bytes = pDataCols->cols[i].bytes;
-    pRet->cols[i].offset = pDataCols->cols[i].offset;
-
-    if (keepData) {
-      if (pDataCols->cols[i].len > 0) {
-        if (tdAllocMemForCol(&pRet->cols[i], pRet->maxPoints) < 0) {
-          tdFreeDataCols(pRet);
-          return NULL;
-        }
-        pRet->cols[i].len = pDataCols->cols[i].len;
-        memcpy(pRet->cols[i].pData, pDataCols->cols[i].pData, pDataCols->cols[i].len);
-        if (IS_VAR_DATA_TYPE(pRet->cols[i].type)) {
-          int dataOffSize = sizeof(VarDataOffsetT) * pDataCols->maxPoints;
-          memcpy(pRet->cols[i].dataOff, pDataCols->cols[i].dataOff, dataOffSize);
-        }
-      }
-    }
-  }
-
-  return pRet;
-}
-#endif
-
 void tdResetDataCols(SDataCols *pCols) {
   if (pCols != NULL) {
     pCols->numOfRows = 0;
@@ -528,180 +347,6 @@ void tdResetDataCols(SDataCols *pCols) {
     }
   }
 }
-#if 0
-static void tdAppendDataRowToDataCol(SDataRow row, STSchema *pSchema, SDataCols *pCols, bool forceSetNull) {
-  ASSERT(pCols->numOfRows == 0 || dataColsKeyLast(pCols) < dataRowKey(row));
-
-  int rcol = 0;
-  int dcol = 0;
-
-  while (dcol < pCols->numOfCols) {
-    bool setCol = 0;
-    SDataCol *pDataCol = &(pCols->cols[dcol]);
-    if (rcol >= schemaNCols(pSchema)) {
-      dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
-      dcol++;
-      continue;
-    }
-
-    STColumn *pRowCol = schemaColAt(pSchema, rcol);
-    if (pRowCol->colId == pDataCol->colId) {
-      void *value = tdGetRowDataOfCol(row, pRowCol->type, pRowCol->offset + TD_DATA_ROW_HEAD_SIZE);
-      if(!isNull(value, pDataCol->type)) setCol = 1;
-      dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints);
-      dcol++;
-      rcol++;
-    } else if (pRowCol->colId < pDataCol->colId) {
-      rcol++;
-    } else {
-      if(forceSetNull || setCol) {
-        dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
-      }
-      dcol++;
-    }
-  }
-  pCols->numOfRows++;
-}
-
-static void tdAppendKVRowToDataCol(SKVRow row, STSchema *pSchema, SDataCols *pCols, bool forceSetNull) {
-  ASSERT(pCols->numOfRows == 0 || dataColsKeyLast(pCols) < kvRowKey(row));
-
-  int rcol = 0;
-  int dcol = 0;
-
-  int nRowCols = kvRowNCols(row);
-
-  while (dcol < pCols->numOfCols) {
-    bool setCol = 0;
-    SDataCol *pDataCol = &(pCols->cols[dcol]);
-    if (rcol >= nRowCols || rcol >= schemaNCols(pSchema)) {
-      dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
-      ++dcol;
-      continue;
-    }
-
-    SColIdx *colIdx = kvRowColIdxAt(row, rcol);
-
-    if (colIdx->colId == pDataCol->colId) {
-      void *value = tdGetKvRowDataOfCol(row, colIdx->offset);
-      if(!isNull(value, pDataCol->type)) setCol = 1;
-      dataColAppendVal(pDataCol, value, pCols->numOfRows, pCols->maxPoints);
-      ++dcol;
-      ++rcol;
-    } else if (colIdx->colId < pDataCol->colId) {
-      ++rcol;
-    } else {
-      if(forceSetNull || setCol) {
-        dataColAppendVal(pDataCol, getNullValue(pDataCol->type), pCols->numOfRows, pCols->maxPoints);
-      }
-      ++dcol;
-    }
-  }
-  pCols->numOfRows++;
-}
-
-void tdAppendMemRowToDataCol(SMemRow row, STSchema *pSchema, SDataCols *pCols, bool forceSetNull) {
-  if (isDataRow(row)) {
-    tdAppendDataRowToDataCol(memRowDataBody(row), pSchema, pCols, forceSetNull);
-  } else if (isKvRow(row)) {
-    tdAppendKVRowToDataCol(memRowKvBody(row), pSchema, pCols, forceSetNull);
-  } else {
-    ASSERT(0);
-  }
-}
-
-int tdMergeDataCols(SDataCols *target, SDataCols *source, int rowsToMerge, int *pOffset, bool forceSetNull) {
-  ASSERT(rowsToMerge > 0 && rowsToMerge <= source->numOfRows);
-  ASSERT(target->numOfCols == source->numOfCols);
-  int offset = 0;
-
-  if (pOffset == NULL) {
-    pOffset = &offset;
-  }
-
-  SDataCols *pTarget = NULL;
-
-  if ((target->numOfRows == 0) || (dataColsKeyLast(target) < dataColsKeyAtRow(source, *pOffset))) {  // No overlap
-    ASSERT(target->numOfRows + rowsToMerge <= target->maxPoints);
-    for (int i = 0; i < rowsToMerge; i++) {
-      for (int j = 0; j < source->numOfCols; j++) {
-        if (source->cols[j].len > 0 || target->cols[j].len > 0) {
-          dataColAppendVal(target->cols + j, tdGetColDataOfRow(source->cols + j, i + (*pOffset)), target->numOfRows,
-                           target->maxPoints);
-        }
-      }
-      target->numOfRows++;
-    }
-    (*pOffset) += rowsToMerge;
-  } else {
-    pTarget = tdDupDataCols(target, true);
-    if (pTarget == NULL) goto _err;
-
-    int iter1 = 0;
-    tdMergeTwoDataCols(target, pTarget, &iter1, pTarget->numOfRows, source, pOffset, source->numOfRows,
-                       pTarget->numOfRows + rowsToMerge, forceSetNull);
-  }
-
-  tdFreeDataCols(pTarget);
-  return 0;
-
-_err:
-  tdFreeDataCols(pTarget);
-  return -1;
-}
-
-// src2 data has more priority than src1
-static void tdMergeTwoDataCols(SDataCols *target, SDataCols *src1, int *iter1, int limit1, SDataCols *src2, int *iter2,
-                               int limit2, int tRows, bool forceSetNull) {
-  tdResetDataCols(target);
-  ASSERT(limit1 <= src1->numOfRows && limit2 <= src2->numOfRows);
-
-  while (target->numOfRows < tRows) {
-    if (*iter1 >= limit1 && *iter2 >= limit2) break;
-
-    TSKEY key1 = (*iter1 >= limit1) ? INT64_MAX : dataColsKeyAt(src1, *iter1);
-    TKEY  tkey1 = (*iter1 >= limit1) ? TKEY_NULL : dataColsTKeyAt(src1, *iter1);
-    TSKEY key2 = (*iter2 >= limit2) ? INT64_MAX : dataColsKeyAt(src2, *iter2);
-    TKEY  tkey2 = (*iter2 >= limit2) ? TKEY_NULL : dataColsTKeyAt(src2, *iter2);
-
-    ASSERT(tkey1 == TKEY_NULL || (!TKEY_IS_DELETED(tkey1)));
-
-    if (key1 < key2) {
-      for (int i = 0; i < src1->numOfCols; i++) {
-        ASSERT(target->cols[i].type == src1->cols[i].type);
-        if (src1->cols[i].len > 0 || target->cols[i].len > 0) {
-          dataColAppendVal(&(target->cols[i]), tdGetColDataOfRow(src1->cols + i, *iter1), target->numOfRows,
-                           target->maxPoints);
-        }
-      }
-
-      target->numOfRows++;
-      (*iter1)++;
-    } else if (key1 >= key2) {
-      if ((key1 > key2) || (key1 == key2 && !TKEY_IS_DELETED(tkey2))) {
-        for (int i = 0; i < src2->numOfCols; i++) {
-          ASSERT(target->cols[i].type == src2->cols[i].type);
-          if (src2->cols[i].len > 0 && !isNull(src2->cols[i].pData, src2->cols[i].type)) {
-            dataColAppendVal(&(target->cols[i]), tdGetColDataOfRow(src2->cols + i, *iter2), target->numOfRows,
-                             target->maxPoints);
-          } else if(!forceSetNull && key1 == key2 && src1->cols[i].len > 0) {
-            dataColAppendVal(&(target->cols[i]), tdGetColDataOfRow(src1->cols + i, *iter1), target->numOfRows,
-                             target->maxPoints);
-          } else if(target->cols[i].len > 0) {
-            dataColSetNullAt(&target->cols[i], target->numOfRows);
-          }
-        }
-        target->numOfRows++;
-      }
-
-      (*iter2)++;
-      if (key1 == key2) (*iter1)++;
-    }
-
-    ASSERT(target->numOfRows <= target->maxPoints);
-  }
-}
-#endif
 
 SKVRow tdKVRowDup(SKVRow row) {
   SKVRow trow = taosMemoryMalloc(kvRowLen(row));
@@ -859,98 +504,3 @@ SKVRow tdGetKVRowFromBuilder(SKVRowBuilder *pBuilder) {
 
   return row;
 }
-#if 0
-SMemRow mergeTwoMemRows(void *buffer, SMemRow row1, SMemRow row2, STSchema *pSchema1, STSchema *pSchema2) {
-#if 0
-  ASSERT(memRowKey(row1) == memRowKey(row2));
-  ASSERT(schemaVersion(pSchema1) == memRowVersion(row1));
-  ASSERT(schemaVersion(pSchema2) == memRowVersion(row2));
-  ASSERT(schemaVersion(pSchema1) >= schemaVersion(pSchema2));
-#endif
-
-  SArray *stashRow = taosArrayInit(pSchema1->numOfCols, sizeof(SColInfo));
-  if (stashRow == NULL) {
-    return NULL;
-  }
-
-  SMemRow  pRow = buffer;
-  SDataRow dataRow = memRowDataBody(pRow);
-  memRowSetType(pRow, SMEM_ROW_DATA);
-  dataRowSetVersion(dataRow, schemaVersion(pSchema1));  // use latest schema version
-  dataRowSetLen(dataRow, (TDRowLenT)(TD_DATA_ROW_HEAD_SIZE + pSchema1->flen));
-
-  TDRowLenT dataLen = 0, kvLen = TD_MEM_ROW_KV_HEAD_SIZE;
-
-  int32_t  i = 0;  // row1
-  int32_t  j = 0;  // row2
-  int32_t  nCols1 = schemaNCols(pSchema1);
-  int32_t  nCols2 = schemaNCols(pSchema2);
-  SColInfo colInfo = {0};
-  int32_t  kvIdx1 = 0, kvIdx2 = 0;
-
-  while (i < nCols1) {
-    STColumn *pCol = schemaColAt(pSchema1, i);
-    void *    val1 = tdGetMemRowDataOfColEx(row1, pCol->colId, pCol->type, TD_DATA_ROW_HEAD_SIZE + pCol->offset, &kvIdx1);
-    // if val1 != NULL, use val1;
-    if (val1 != NULL && !isNull(val1, pCol->type)) {
-      tdAppendColVal(dataRow, val1, pCol->type, pCol->offset);
-      kvLen += tdGetColAppendLen(SMEM_ROW_KV, val1, pCol->type);
-      setSColInfo(&colInfo, pCol->colId, pCol->type, val1);
-      taosArrayPush(stashRow, &colInfo);
-      ++i;  // next col
-      continue;
-    }
-
-    void *val2 = NULL;
-    while (j < nCols2) {
-      STColumn *tCol = schemaColAt(pSchema2, j);
-      if (tCol->colId < pCol->colId) {
-        ++j;
-        continue;
-      }
-      if (tCol->colId == pCol->colId) {
-        val2 = tdGetMemRowDataOfColEx(row2, tCol->colId, tCol->type, TD_DATA_ROW_HEAD_SIZE + tCol->offset, &kvIdx2);
-      } else if (tCol->colId > pCol->colId) {
-        // set NULL
-      }
-      break;
-    }  // end of while(j<nCols2)
-    if (val2 == NULL) {
-      val2 = (void *)getNullValue(pCol->type);
-    }
-    tdAppendColVal(dataRow, val2, pCol->type, pCol->offset);
-    if (!isNull(val2, pCol->type)) {
-      kvLen += tdGetColAppendLen(SMEM_ROW_KV, val2, pCol->type);
-      setSColInfo(&colInfo, pCol->colId, pCol->type, val2);
-      taosArrayPush(stashRow, &colInfo);
-    }
-
-    ++i;  // next col
-  }
-
-  dataLen = memRowTLen(pRow);
-
-  if (kvLen < dataLen) {
-    // scan stashRow and generate SKVRow
-    memset(buffer, 0, sizeof(dataLen));
-    SMemRow tRow = buffer;
-    memRowSetType(tRow, SMEM_ROW_KV);
-    SKVRow kvRow = (SKVRow)memRowKvBody(tRow);
-    int16_t nKvNCols = (int16_t) taosArrayGetSize(stashRow);
-    kvRowSetLen(kvRow, (TDRowLenT)(TD_KV_ROW_HEAD_SIZE + sizeof(SColIdx) * nKvNCols));
-    kvRowSetNCols(kvRow, nKvNCols);
-    memRowSetKvVersion(tRow, pSchema1->version);
-
-    int32_t toffset = 0;
-    int16_t k;
-    for (k = 0; k < nKvNCols; ++k) {
-      SColInfo *pColInfo = taosArrayGet(stashRow, k);
-      tdAppendKvColVal(kvRow, pColInfo->colVal, true, pColInfo->colId, pColInfo->colType, toffset);
-      toffset += sizeof(SColIdx);
-    }
-    ASSERT(kvLen == memRowTLen(tRow));
-  }
-  taosArrayDestroy(stashRow);
-  return buffer;
-}
-#endif

source/common/src/tglobal.c

@@ -169,6 +169,9 @@ uint32_t tsMaxRange = 500;                      // max range
 uint32_t tsCurRange = 100;                      // range
 char     tsCompressor[32] = "ZSTD_COMPRESSOR";  // ZSTD_COMPRESSOR or GZIP_COMPRESSOR
 
+// udf
+bool     tsStartUdfd = true;
+
 // internal
 int32_t tsTransPullupInterval = 6;
 int32_t tsMqRebalanceInterval = 2;
@@ -348,7 +351,7 @@ static int32_t taosAddSystemCfg(SConfig *pCfg) {
 }
 
 static int32_t taosAddServerCfg(SConfig *pCfg) {
-  if (cfgAddInt32(pCfg, "supportVnodes", 256, 0, 65536, 0) != 0) return -1;
+  if (cfgAddInt32(pCfg, "supportVnodes", 256, 0, 4096, 0) != 0) return -1;
   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, "maxNumOfDistinctRes", tsMaxNumOfDistinctResults, 10 * 10000, 10000 * 10000, 0) != 0) return -1;
@@ -441,6 +444,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "transPullupInterval", tsTransPullupInterval, 1, 10000, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "mqRebalanceInterval", tsMqRebalanceInterval, 1, 10000, 1) != 0) return -1;
 
+  if (cfgAddBool(pCfg, "startUdfd", tsStartUdfd, 0) != 0) return -1;
   return 0;
 }
 
@@ -581,6 +585,8 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
   tsTransPullupInterval = cfgGetItem(pCfg, "transPullupInterval")->i32;
   tsMqRebalanceInterval = cfgGetItem(pCfg, "mqRebalanceInterval")->i32;
 
+  tsStartUdfd = cfgGetItem(pCfg, "startUdfd")->bval;
+
   if (tsQueryBufferSize >= 0) {
     tsQueryBufferSizeBytes = tsQueryBufferSize * 1048576UL;
   }

source/common/src/tmsg.c

@@ -607,6 +607,11 @@ int32_t tSerializeSMAlterStbReq(void *buf, int32_t bufLen, SMAlterStbReq *pReq)
     if (tEncodeI32(&encoder, pField->bytes) < 0) return -1;
     if (tEncodeCStr(&encoder, pField->name) < 0) return -1;
   }
+  if (tEncodeI32(&encoder, pReq->ttl) < 0) return -1;
+  if (tEncodeI32(&encoder, pReq->commentLen) < 0) return -1;
+  if (pReq->commentLen > 0) {
+    if (tEncodeCStr(&encoder, pReq->comment) < 0) return -1;
+  }
   tEndEncode(&encoder);
 
   int32_t tlen = encoder.pos;
@@ -639,6 +644,14 @@ int32_t tDeserializeSMAlterStbReq(void *buf, int32_t bufLen, SMAlterStbReq *pReq
     }
   }
 
+  if (tDecodeI32(&decoder, &pReq->ttl) < 0) return -1;
+  if (tDecodeI32(&decoder, &pReq->commentLen) < 0) return -1;
+  if (pReq->commentLen > 0) {
+    pReq->comment = taosMemoryMalloc(pReq->commentLen);
+    if (pReq->comment == NULL) return -1;
+    if (tDecodeCStrTo(&decoder, pReq->comment) < 0) return -1;
+  }
+
   tEndDecode(&decoder);
   tDecoderClear(&decoder);
   return 0;
@@ -1233,21 +1246,21 @@ int32_t tDeserializeSGetUserAuthRspImpl(SDecoder *pDecoder, SGetUserAuthRsp *pRs
   for (int32_t i = 0; i < numOfCreatedDbs; ++i) {
     char db[TSDB_DB_FNAME_LEN] = {0};
     if (tDecodeCStrTo(pDecoder, db) < 0) return -1;
-    int32_t len = strlen(db) + 1;
+    int32_t len = strlen(db);
     taosHashPut(pRsp->createdDbs, db, len, db, len);
   }
 
   for (int32_t i = 0; i < numOfReadDbs; ++i) {
     char db[TSDB_DB_FNAME_LEN] = {0};
     if (tDecodeCStrTo(pDecoder, db) < 0) return -1;
-    int32_t len = strlen(db) + 1;
+    int32_t len = strlen(db);
     taosHashPut(pRsp->readDbs, db, len, db, len);
   }
 
   for (int32_t i = 0; i < numOfWriteDbs; ++i) {
     char db[TSDB_DB_FNAME_LEN] = {0};
     if (tDecodeCStrTo(pDecoder, db) < 0) return -1;
-    int32_t len = strlen(db) + 1;
+    int32_t len = strlen(db);
     taosHashPut(pRsp->writeDbs, db, len, db, len);
   }
 

source/common/src/tname.c

@@ -18,11 +18,9 @@
 #include "tcommon.h"
 #include "tstrbuild.h"
 
-#define VALID_NAME_TYPE(x)  ((x) == TSDB_DB_NAME_T || (x) == TSDB_TABLE_NAME_T)
+#define VALID_NAME_TYPE(x) ((x) == TSDB_DB_NAME_T || (x) == TSDB_TABLE_NAME_T)
 
-bool tscValidateTableNameLength(size_t len) {
-  return len < TSDB_TABLE_NAME_LEN;
-}
+bool tscValidateTableNameLength(size_t len) { return len < TSDB_TABLE_NAME_LEN; }
 
 #if 0
 // TODO refactor
@@ -95,12 +93,12 @@ int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t slidingTime, in
       * but in case of DST, the start time of one day need to be dynamically decided.
       */
       // todo refactor to extract function that is available for Linux/Windows/Mac platform
-  #if defined(WINDOWS) && _MSC_VER >= 1900
+#if defined(WINDOWS) && _MSC_VER >= 1900
       // see https://docs.microsoft.com/en-us/cpp/c-runtime-library/daylight-dstbias-timezone-and-tzname?view=vs-2019
       int64_t timezone = _timezone;
       int32_t daylight = _daylight;
       char**  tzname = _tzname;
-  #endif
+#endif
 
       int64_t t = (precision == TSDB_TIME_PRECISION_MILLI) ? MILLISECOND_PER_SECOND : MILLISECOND_PER_SECOND * 1000L;
       start += timezone * t;
@@ -142,10 +140,10 @@ int32_t tNameExtractFullName(const SName* name, char* dst) {
 int32_t tNameLen(const SName* name) {
   assert(name != NULL);
 
-  char tmp[12] = {0};
+  char    tmp[12] = {0};
   int32_t len = sprintf(tmp, "%d", name->acctId);
-  int32_t len1 = (int32_t) strlen(name->dbname);
-  int32_t len2 = (int32_t) strlen(name->tname);
+  int32_t len1 = (int32_t)strlen(name->dbname);
+  int32_t len2 = (int32_t)strlen(name->tname);
 
   if (name->type == TSDB_DB_NAME_T) {
     assert(len2 == 0);
@@ -200,9 +198,7 @@ const char* tNameGetTableName(const SName* name) {
   return &name->tname[0];
 }
 
-void tNameAssign(SName* dst, const SName* src) {
-  memcpy(dst, src, sizeof(SName));
-}
+void tNameAssign(SName* dst, const SName* src) { memcpy(dst, src, sizeof(SName)); }
 
 int32_t tNameSetDbName(SName* dst, int32_t acct, const char* dbName, size_t nameLen) {
   assert(dst != NULL && dbName != NULL && nameLen > 0);
@@ -244,7 +240,6 @@ bool tNameDBNameEqual(SName* left, SName* right) {
   return (0 == strcmp(left->dbname, right->dbname));
 }
 
-
 int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
   assert(dst != NULL && str != NULL && strlen(str) > 0);
 
@@ -260,14 +255,14 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
 
   if ((type & T_NAME_DB) == T_NAME_DB) {
     dst->type = TSDB_DB_NAME_T;
-    char* start = (char*)((p == NULL)? str:(p+1));
+    char* start = (char*)((p == NULL) ? str : (p + 1));
 
     int32_t len = 0;
     p = strstr(start, TS_PATH_DELIMITER);
     if (p == NULL) {
-      len = (int32_t) strlen(start);
+      len = (int32_t)strlen(start);
     } else {
-      len = (int32_t) (p - start);
+      len = (int32_t)(p - start);
     }
 
     // too long account id or too long db name
@@ -275,21 +270,21 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
       return -1;
     }
 
-    memcpy (dst->dbname, start, len);
+    memcpy(dst->dbname, start, len);
     dst->dbname[len] = 0;
   }
 
   if ((type & T_NAME_TABLE) == T_NAME_TABLE) {
     dst->type = TSDB_TABLE_NAME_T;
-    char* start = (char*) ((p == NULL)? str: (p+1));
+    char* start = (char*)((p == NULL) ? str : (p + 1));
 
     // too long account id or too long db name
-    int32_t len = (int32_t) strlen(start);
+    int32_t len = (int32_t)strlen(start);
     if ((len >= tListLen(dst->tname)) || (len <= 0)) {
       return -1;
     }
 
-    memcpy (dst->tname, start, len);
+    memcpy(dst->tname, start, len);
     dst->tname[len] = 0;
   }
 
@@ -305,14 +300,14 @@ static int compareKv(const void* p1, const void* p2) {
   if (res != 0) {
     return res;
   } else {
-    return kvLen1-kvLen2;
+    return kvLen1 - kvLen2;
   }
 }
 
 /*
  * use stable name and tags to grearate child table name
  */
-void buildChildTableName(RandTableName *rName) {
+void buildChildTableName(RandTableName* rName) {
   int32_t size = taosArrayGetSize(rName->tags);
   ASSERT(size > 0);
   taosArraySort(rName->tags, compareKv);
@@ -320,19 +315,19 @@ void buildChildTableName(RandTableName *rName) {
   SStringBuilder sb = {0};
   taosStringBuilderAppendStringLen(&sb, rName->sTableName, rName->sTableNameLen);
   for (int j = 0; j < size; ++j) {
-    SSmlKv *tagKv = taosArrayGetP(rName->tags, j);
+    SSmlKv* tagKv = taosArrayGetP(rName->tags, j);
     taosStringBuilderAppendStringLen(&sb, tagKv->key, tagKv->keyLen);
     taosStringBuilderAppendStringLen(&sb, tagKv->value, tagKv->valueLen);
   }
-  size_t len = 0;
-  char* keyJoined = taosStringBuilderGetResult(&sb, &len);
+  size_t    len = 0;
+  char*     keyJoined = taosStringBuilderGetResult(&sb, &len);
   T_MD5_CTX context;
   tMD5Init(&context);
-  tMD5Update(&context, (uint8_t *)keyJoined, (uint32_t)len);
+  tMD5Update(&context, (uint8_t*)keyJoined, (uint32_t)len);
   tMD5Final(&context);
   uint64_t digest1 = *(uint64_t*)(context.digest);
   uint64_t digest2 = *(uint64_t*)(context.digest + 8);
-  snprintf(rName->childTableName, TSDB_TABLE_NAME_LEN, "t_%016"PRIx64"%016"PRIx64, digest1, digest2);
+  snprintf(rName->childTableName, TSDB_TABLE_NAME_LEN, "t_%016" PRIx64 "%016" PRIx64, digest1, digest2);
   taosStringBuilderDestroy(&sb);
   rName->uid = digest1;
 }

source/dnode/mgmt/implement/inc/dmImp.h

@@ -26,8 +26,10 @@ int32_t dmOpenNode(SMgmtWrapper *pWrapper);
 void    dmCloseNode(SMgmtWrapper *pWrapper);
 
 // dmTransport.c
-int32_t  dmInitTrans(SDnode *pDnode);
-void     dmCleanupTrans(SDnode *pDnode);
+int32_t  dmInitServer(SDnode *pDnode);
+void     dmCleanupServer(SDnode *pDnode);
+int32_t  dmInitClient(SDnode *pDnode);
+void     dmCleanupClient(SDnode *pDnode);
 SProcCfg dmGenProcCfg(SMgmtWrapper *pWrapper);
 SMsgCb   dmGetMsgcb(SMgmtWrapper *pWrapper);
 int32_t  dmInitMsgHandle(SDnode *pDnode);

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

@@ -213,10 +213,12 @@ static int32_t dmOpenNodes(SDnode *pDnode) {
     }
 
     pWrapper->procType = DND_PROC_CHILD;
+    if (dmInitClient(pDnode) != 0) {
+      return -1;
+    }
 
-    SMsgCb msgCb = pDnode->data.msgCb;
-    msgCb.pWrapper = pWrapper;
-    tmsgSetDefaultMsgCb(&msgCb);
+    pDnode->data.msgCb = dmGetMsgcb(pWrapper);
+    tmsgSetDefaultMsgCb(&pDnode->data.msgCb);
 
     if (dmOpenNode(pWrapper) != 0) {
       dError("node:%s, failed to open since %s", pWrapper->name, terrstr());
@@ -234,6 +236,15 @@ static int32_t dmOpenNodes(SDnode *pDnode) {
         pWrapper->procType = DND_PROC_SINGLE;
       }
 
+      if (n == DNODE) {
+        if (dmInitClient(pDnode) != 0) {
+          return -1;
+        }
+
+        pDnode->data.msgCb = dmGetMsgcb(pWrapper);
+        tmsgSetDefaultMsgCb(&pDnode->data.msgCb);
+      }
+
       if (dmOpenNode(pWrapper) != 0) {
         dError("node:%s, failed to open since %s", pWrapper->name, terrstr());
         return -1;
@@ -281,21 +292,21 @@ static void dmProcessProcHandle(void *handle) {
 }
 
 static void dmWatchNodes(SDnode *pDnode) {
+  if (pDnode->ptype != DND_PROC_PARENT) return;
+  if (pDnode->ntype == NODE_END) return;
+
   taosThreadMutexLock(&pDnode->mutex);
-  if (pDnode->ptype == DND_PROC_PARENT) {
-    for (EDndNodeType n = DNODE + 1; n < NODE_END; ++n) {
-      SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
-      if (!pWrapper->required) continue;
-      if (pWrapper->procType != DND_PROC_PARENT) continue;
-      if (pDnode->ntype == NODE_END) continue;
+  for (EDndNodeType n = DNODE + 1; n < NODE_END; ++n) {
+    SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
+    if (!pWrapper->required) continue;
+    if (pWrapper->procType != DND_PROC_PARENT) continue;
 
-      if (pWrapper->procId <= 0 || !taosProcExist(pWrapper->procId)) {
-        dWarn("node:%s, process:%d is killed and needs to be restarted", pWrapper->name, pWrapper->procId);
-        if (pWrapper->procObj) {
-          taosProcCloseHandles(pWrapper->procObj, dmProcessProcHandle);
-        }
-        dmNewNodeProc(pWrapper, n);
+    if (pWrapper->procId <= 0 || !taosProcExist(pWrapper->procId)) {
+      dWarn("node:%s, process:%d is killed and needs to be restarted", pWrapper->name, pWrapper->procId);
+      if (pWrapper->procObj) {
+        taosProcCloseHandles(pWrapper->procObj, dmProcessProcHandle);
       }
+      dmNewNodeProc(pWrapper, n);
     }
   }
   taosThreadMutexUnlock(&pDnode->mutex);

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

@@ -217,145 +217,6 @@ static void dmStopMgmt(SMgmtWrapper *pWrapper) {
   dmStopStatusThread(pWrapper->pDnode);
 }
 
-static int32_t dmSpawnUdfd(SDnode *pDnode);
-
-void dmUdfdExit(uv_process_t *process, int64_t exitStatus, int termSignal) {
-  dInfo("udfd process exited with status %" PRId64 ", signal %d", exitStatus, termSignal);
-  SDnode *pDnode = process->data;
-  if (exitStatus == 0 && termSignal == 0 || atomic_load_32(&pDnode->udfdData.stopCalled)) {
-    dInfo("udfd process exit due to SIGINT or dnode-mgmt called stop");
-  } else {
-    dInfo("udfd process restart");
-    dmSpawnUdfd(pDnode);
-  }
-}
-
-static int32_t dmSpawnUdfd(SDnode *pDnode) {
-  dInfo("dnode start spawning udfd");
-  uv_process_options_t options = {0};
-
-  char path[PATH_MAX] = {0};
-  if (tsProcPath == NULL) {
-    path[0] = '.';
-  } else {
-    strncpy(path, tsProcPath, strlen(tsProcPath));
-    taosDirName(path);
-  }
-#ifdef WINDOWS
-  strcat(path, "udfd.exe");
-#else
-  strcat(path, "/udfd");
-#endif
-  char* argsUdfd[] = {path, "-c", configDir, NULL};
-  options.args = argsUdfd;
-  options.file = path;
-
-  options.exit_cb = dmUdfdExit;
-
-  SUdfdData *pData = &pDnode->udfdData;
-  uv_pipe_init(&pData->loop, &pData->ctrlPipe, 1);
-
-  uv_stdio_container_t child_stdio[3];
-  child_stdio[0].flags = UV_CREATE_PIPE | UV_READABLE_PIPE;
-  child_stdio[0].data.stream = (uv_stream_t*) &pData->ctrlPipe;
-  child_stdio[1].flags = UV_IGNORE;
-  child_stdio[2].flags = UV_INHERIT_FD;
-  child_stdio[2].data.fd = 2;
-  options.stdio_count = 3;
-  options.stdio = child_stdio;
-
-  options.flags = UV_PROCESS_DETACHED;
-
-  char dnodeIdEnvItem[32] = {0};
-  char thrdPoolSizeEnvItem[32] = {0};
-  snprintf(dnodeIdEnvItem, 32, "%s=%d", "DNODE_ID", pDnode->data.dnodeId);
-  float numCpuCores = 4;
-  taosGetCpuCores(&numCpuCores);
-  snprintf(thrdPoolSizeEnvItem,32,  "%s=%d", "UV_THREADPOOL_SIZE", (int)numCpuCores*2);
-  char* envUdfd[] = {dnodeIdEnvItem, thrdPoolSizeEnvItem, NULL};
-  options.env = envUdfd;
-
-  int err = uv_spawn(&pData->loop, &pData->process, &options);
-  pData->process.data = (void*)pDnode;
-
-  if (err != 0) {
-    dError("can not spawn udfd. path: %s, error: %s", path, uv_strerror(err));
-  }
-  return err;
-}
-
-static void dmUdfdCloseWalkCb(uv_handle_t* handle, void* arg) {
-  if (!uv_is_closing(handle)) {
-    uv_close(handle, NULL);
-  }
-}
-
-static void dmUdfdStopAsyncCb(uv_async_t *async) {
-  SDnode *pDnode = async->data;
-  SUdfdData *pData = &pDnode->udfdData;
-  uv_stop(&pData->loop);
-}
-
-static void dmWatchUdfd(void *args) {
-  SDnode *pDnode = args;
-  SUdfdData *pData = &pDnode->udfdData;
-  uv_loop_init(&pData->loop);
-  uv_async_init(&pData->loop, &pData->stopAsync, dmUdfdStopAsyncCb);
-  pData->stopAsync.data = pDnode;
-  int32_t err = dmSpawnUdfd(pDnode);
-  atomic_store_32(&pData->spawnErr, err);
-  uv_barrier_wait(&pData->barrier);
-  uv_run(&pData->loop, UV_RUN_DEFAULT);
-  uv_loop_close(&pData->loop);
-
-  uv_walk(&pData->loop, dmUdfdCloseWalkCb, NULL);
-  uv_run(&pData->loop, UV_RUN_DEFAULT);
-  uv_loop_close(&pData->loop);
-  return;
-}
-
-static int32_t dmStartUdfd(SDnode *pDnode) {
-  char dnodeId[8] = {0};
-  snprintf(dnodeId, sizeof(dnodeId), "%d", pDnode->data.dnodeId);
-  uv_os_setenv("DNODE_ID", dnodeId);
-  SUdfdData *pData = &pDnode->udfdData;
-  if (pData->startCalled) {
-    dInfo("dnode-mgmt start udfd already called");
-    return 0;
-  }
-  pData->startCalled = true;
-  uv_barrier_init(&pData->barrier, 2);
-  uv_thread_create(&pData->thread, dmWatchUdfd, pDnode);
-  uv_barrier_wait(&pData->barrier);
-  int32_t err = atomic_load_32(&pData->spawnErr);
-  if (err != 0) {
-    uv_barrier_destroy(&pData->barrier);
-    uv_async_send(&pData->stopAsync);
-    uv_thread_join(&pData->thread);
-    pData->needCleanUp = false;
-    dInfo("dnode-mgmt udfd cleaned up after spawn err");
-  } else {
-    pData->needCleanUp = true;
-  }
-  return err;
-}
-
-static int32_t dmStopUdfd(SDnode *pDnode) {
-  dInfo("dnode-mgmt to stop udfd. need cleanup: %d, spawn err: %d",
-        pDnode->udfdData.needCleanUp, pDnode->udfdData.spawnErr);
-  SUdfdData *pData = &pDnode->udfdData;
-  if (!pData->needCleanUp || atomic_load_32(&pData->stopCalled)) {
-    return 0;
-  }
-  atomic_store_32(&pData->stopCalled, 1);
-  pData->needCleanUp = false;
-  uv_barrier_destroy(&pData->barrier);
-  uv_async_send(&pData->stopAsync);
-  uv_thread_join(&pData->thread);
-  dInfo("dnode-mgmt udfd cleaned up");
-  return 0;
-}
-
 static int32_t dmInitMgmt(SMgmtWrapper *pWrapper) {
   dInfo("dnode-mgmt start to init");
   SDnode *pDnode = pWrapper->pDnode;
@@ -381,13 +242,13 @@ static int32_t dmInitMgmt(SMgmtWrapper *pWrapper) {
     return -1;
   }
 
-  if (dmInitTrans(pDnode) != 0) {
+  if (dmInitServer(pDnode) != 0) {
     dError("failed to init transport since %s", terrstr());
     return -1;
   }
   dmReportStartup(pDnode, "dnode-transport", "initialized");
 
-  if (dmStartUdfd(pDnode) != 0) {
+  if (udfStartUdfd(pDnode->data.dnodeId) != 0) {
     dError("failed to start udfd");
   }
 
@@ -398,7 +259,9 @@ static int32_t dmInitMgmt(SMgmtWrapper *pWrapper) {
 static void dmCleanupMgmt(SMgmtWrapper *pWrapper) {
   dInfo("dnode-mgmt start to clean up");
   SDnode *pDnode = pWrapper->pDnode;
-  dmStopUdfd(pDnode);
+
+  udfStopUdfd();
+
   dmStopWorker(pDnode);
 
   taosWLockLatch(&pDnode->data.latch);
@@ -412,7 +275,8 @@ static void dmCleanupMgmt(SMgmtWrapper *pWrapper) {
   }
   taosWUnLockLatch(&pDnode->data.latch);
 
-  dmCleanupTrans(pDnode);
+  dmCleanupClient(pDnode);
+  dmCleanupServer(pDnode);
   dInfo("dnode-mgmt is cleaned up");
 }
 

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

@@ -124,9 +124,6 @@ SDnode *dmCreate(const SDnodeOpt *pOption) {
     goto _OVER;
   }
 
-  pDnode->data.msgCb = dmGetMsgcb(&pDnode->wrappers[DNODE]);
-  tmsgSetDefaultMsgCb(&pDnode->data.msgCb);
-
   dInfo("dnode is created, data:%p", pDnode);
   code = 0;
 

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

@@ -16,6 +16,8 @@
 #define _DEFAULT_SOURCE
 #include "dmImp.h"
 
+#include "qworker.h"
+
 #define INTERNAL_USER   "_dnd"
 #define INTERNAL_CKEY   "_key"
 #define INTERNAL_SECRET "_pwd"
@@ -85,17 +87,14 @@ static void dmProcessRpcMsg(SMgmtWrapper *pWrapper, SRpcMsg *pRpc, SEpSet *pEpSe
   if ((pMsg = taosAllocateQitem(sizeof(SNodeMsg))) == NULL) goto _OVER;
   if (dmBuildMsg(pMsg, pRpc) != 0) goto _OVER;
 
-  if (pWrapper->procType == DND_PROC_SINGLE) {
+  if (pWrapper->procType != DND_PROC_PARENT) {
     dTrace("msg:%p, created, type:%s handle:%p user:%s", pMsg, TMSG_INFO(msgType), pRpc->handle, pMsg->user);
     code = (*msgFp)(pWrapper, pMsg);
-  } else if (pWrapper->procType == DND_PROC_PARENT) {
+  } else {
     dTrace("msg:%p, created and put into child queue, type:%s handle:%p code:0x%04x user:%s contLen:%d", pMsg,
            TMSG_INFO(msgType), pRpc->handle, pMsg->rpcMsg.code & 0XFFFF, pMsg->user, pRpc->contLen);
     code = taosProcPutToChildQ(pWrapper->procObj, pMsg, sizeof(SNodeMsg), pRpc->pCont, pRpc->contLen,
                                (isReq && (pMsg->rpcMsg.code == 0)) ? pRpc->handle : NULL, pRpc->refId, PROC_FUNC_REQ);
-  } else {
-    dTrace("msg:%p, should not processed in child process, handle:%p user:%s", pMsg, pRpc->handle, pMsg->user);
-    ASSERT(1);
   }
 
 _OVER:
@@ -108,7 +107,7 @@ _OVER:
   } else {
     dError("msg:%p, type:%s handle:%p failed to process since 0x%04x:%s", pMsg, TMSG_INFO(msgType), pRpc->handle,
            code & 0XFFFF, terrstr());
-    if (msgType & 1U) {
+    if (isReq) {
       if (terrno != 0) code = terrno;
       if (code == TSDB_CODE_NODE_NOT_DEPLOYED || code == TSDB_CODE_NODE_OFFLINE) {
         if (msgType > TDMT_MND_MSG && msgType < TDMT_VND_MSG) {
@@ -130,22 +129,27 @@ _OVER:
 }
 
 static void dmProcessMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
-  SDnodeTrans * pTrans = &pDnode->trans;
+  SDnodeTrans  *pTrans = &pDnode->trans;
   tmsg_t        msgType = pMsg->msgType;
   bool          isReq = msgType & 1u;
-  SMsgHandle *  pHandle = &pTrans->msgHandles[TMSG_INDEX(msgType)];
+  SMsgHandle   *pHandle = &pTrans->msgHandles[TMSG_INDEX(msgType)];
   SMgmtWrapper *pWrapper = pHandle->pNdWrapper;
 
-  if (msgType == TDMT_DND_SERVER_STATUS) {
-    dTrace("server status req will be processed, handle:%p, app:%p", pMsg->handle, pMsg->ahandle);
-    dmProcessServerStatusReq(pDnode, pMsg);
-    return;
-  }
-
-  if (msgType == TDMT_DND_NET_TEST) {
-    dTrace("net test req will be processed, handle:%p, app:%p", pMsg->handle, pMsg->ahandle);
-    dmProcessServerStatusReq(pDnode, pMsg);
-    return;
+  switch (msgType) {
+    case TDMT_DND_SERVER_STATUS:
+      dTrace("server status req will be processed, handle:%p, app:%p", pMsg->handle, pMsg->ahandle);
+      dmProcessServerStatusReq(pDnode, pMsg);
+      return;
+    case TDMT_DND_NET_TEST:
+      dTrace("net test req will be processed, handle:%p, app:%p", pMsg->handle, pMsg->ahandle);
+      dmProcessNetTestReq(pDnode, pMsg);
+      return;
+    case TDMT_MND_SYSTABLE_RETRIEVE_RSP:
+    case TDMT_VND_FETCH_RSP:
+      dTrace("retrieve rsp is received");
+      qWorkerProcessFetchRsp(NULL, NULL, pMsg);
+      pMsg->pCont = NULL;  // already freed in qworker
+      return;
   }
 
   if (pDnode->status != DND_STAT_RUNNING) {
@@ -233,16 +237,6 @@ int32_t dmInitMsgHandle(SDnode *pDnode) {
   return 0;
 }
 
-static inline int32_t dmSendRpcReq(SDnode *pDnode, const SEpSet *pEpSet, SRpcMsg *pReq) {
-  if (pDnode->trans.clientRpc == NULL) {
-    terrno = TSDB_CODE_NODE_OFFLINE;
-    return -1;
-  }
-
-  rpcSendRequest(pDnode->trans.clientRpc, pEpSet, pReq, NULL);
-  return 0;
-}
-
 static void dmSendRpcRedirectRsp(SDnode *pDnode, const SRpcMsg *pReq) {
   SEpSet epSet = {0};
   dmGetMnodeEpSet(pDnode, &epSet);
@@ -288,28 +282,20 @@ void dmSendToMnodeRecv(SDnode *pDnode, SRpcMsg *pReq, SRpcMsg *pRsp) {
 }
 
 static inline int32_t dmSendReq(SMgmtWrapper *pWrapper, const SEpSet *pEpSet, SRpcMsg *pReq) {
-  if (pWrapper->pDnode->status != DND_STAT_RUNNING) {
+  SDnode *pDnode = pWrapper->pDnode;
+  if (pDnode->status != DND_STAT_RUNNING) {
     terrno = TSDB_CODE_NODE_OFFLINE;
     dError("failed to send rpc msg since %s, handle:%p", terrstr(), pReq->handle);
     return -1;
   }
 
-  if (pWrapper->procType != DND_PROC_CHILD) {
-    return dmSendRpcReq(pWrapper->pDnode, pEpSet, pReq);
-  } else {
-    char *pHead = taosMemoryMalloc(sizeof(SRpcMsg) + sizeof(SEpSet));
-    if (pHead == NULL) {
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
-      return -1;
-    }
-
-    memcpy(pHead, pReq, sizeof(SRpcMsg));
-    memcpy(pHead + sizeof(SRpcMsg), pEpSet, sizeof(SEpSet));
-    taosProcPutToParentQ(pWrapper->procObj, pHead, sizeof(SRpcMsg) + sizeof(SEpSet), pReq->pCont, pReq->contLen,
-                         PROC_FUNC_REQ);
-    taosMemoryFree(pHead);
-    return 0;
+  if (pDnode->trans.clientRpc == NULL) {
+    terrno = TSDB_CODE_NODE_OFFLINE;
+    return -1;
   }
+
+  rpcSendRequest(pDnode->trans.clientRpc, pEpSet, pReq, NULL);
+  return 0;
 }
 
 static inline void dmSendRsp(SMgmtWrapper *pWrapper, const SRpcMsg *pRsp) {
@@ -396,9 +382,10 @@ static void dmConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t
   pMsg->pCont = pCont;
 
   if (ftype == PROC_FUNC_REQ) {
+    ASSERT(1);
     dTrace("msg:%p, get from parent queue, send req:%s handle:%p code:0x%04x, app:%p", pMsg, TMSG_INFO(pMsg->msgType),
            pMsg->handle, code, pMsg->ahandle);
-    dmSendRpcReq(pWrapper->pDnode, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg);
+    dmSendReq(pWrapper, (SEpSet *)((char *)pMsg + sizeof(SRpcMsg)), pMsg);
   } else if (ftype == PROC_FUNC_RSP) {
     dTrace("msg:%p, get from parent queue, rsp handle:%p code:0x%04x, app:%p", pMsg, pMsg->handle, code, pMsg->ahandle);
     pMsg->refId = taosProcRemoveHandle(pWrapper->procObj, pMsg->handle);
@@ -421,23 +408,25 @@ static void dmConsumeParentQueue(SMgmtWrapper *pWrapper, SRpcMsg *pMsg, int16_t
 }
 
 SProcCfg dmGenProcCfg(SMgmtWrapper *pWrapper) {
-  SProcCfg cfg = {.childConsumeFp = (ProcConsumeFp)dmConsumeChildQueue,
-                  .childMallocHeadFp = (ProcMallocFp)taosAllocateQitem,
-                  .childFreeHeadFp = (ProcFreeFp)taosFreeQitem,
-                  .childMallocBodyFp = (ProcMallocFp)rpcMallocCont,
-                  .childFreeBodyFp = (ProcFreeFp)rpcFreeCont,
-                  .parentConsumeFp = (ProcConsumeFp)dmConsumeParentQueue,
-                  .parentMallocHeadFp = (ProcMallocFp)taosMemoryMalloc,
-                  .parentFreeHeadFp = (ProcFreeFp)taosMemoryFree,
-                  .parentMallocBodyFp = (ProcMallocFp)rpcMallocCont,
-                  .parentFreeBodyFp = (ProcFreeFp)rpcFreeCont,
-                  .shm = pWrapper->procShm,
-                  .parent = pWrapper,
-                  .name = pWrapper->name};
+  SProcCfg cfg = {
+      .childConsumeFp = (ProcConsumeFp)dmConsumeChildQueue,
+      .childMallocHeadFp = (ProcMallocFp)taosAllocateQitem,
+      .childFreeHeadFp = (ProcFreeFp)taosFreeQitem,
+      .childMallocBodyFp = (ProcMallocFp)rpcMallocCont,
+      .childFreeBodyFp = (ProcFreeFp)rpcFreeCont,
+      .parentConsumeFp = (ProcConsumeFp)dmConsumeParentQueue,
+      .parentMallocHeadFp = (ProcMallocFp)taosMemoryMalloc,
+      .parentFreeHeadFp = (ProcFreeFp)taosMemoryFree,
+      .parentMallocBodyFp = (ProcMallocFp)rpcMallocCont,
+      .parentFreeBodyFp = (ProcFreeFp)rpcFreeCont,
+      .shm = pWrapper->procShm,
+      .parent = pWrapper,
+      .name = pWrapper->name,
+  };
   return cfg;
 }
 
-bool rpcRfp(int32_t code) {
+static bool rpcRfp(int32_t code) {
   if (code == TSDB_CODE_RPC_REDIRECT) {
     return true;
   } else {
@@ -445,7 +434,7 @@ bool rpcRfp(int32_t code) {
   }
 }
 
-static int32_t dmInitClient(SDnode *pDnode) {
+int32_t dmInitClient(SDnode *pDnode) {
   SDnodeTrans *pTrans = &pDnode->trans;
 
   SRpcInit rpcInit = {0};
@@ -471,11 +460,15 @@ static int32_t dmInitClient(SDnode *pDnode) {
     return -1;
   }
 
+  pDnode->data.msgCb = dmGetMsgcb(&pDnode->wrappers[DNODE]);
+  tmsgSetDefaultMsgCb(&pDnode->data.msgCb);
+
   dDebug("dnode rpc client is initialized");
+
   return 0;
 }
 
-static void dmCleanupClient(SDnode *pDnode) {
+void dmCleanupClient(SDnode *pDnode) {
   SDnodeTrans *pTrans = &pDnode->trans;
   if (pTrans->clientRpc) {
     rpcClose(pTrans->clientRpc);
@@ -517,7 +510,7 @@ static inline int32_t dmRetrieveUserAuthInfo(SDnode *pDnode, char *user, char *s
   SAuthReq authReq = {0};
   tstrncpy(authReq.user, user, TSDB_USER_LEN);
   int32_t contLen = tSerializeSAuthReq(NULL, 0, &authReq);
-  void *  pReq = rpcMallocCont(contLen);
+  void   *pReq = rpcMallocCont(contLen);
   tSerializeSAuthReq(pReq, contLen, &authReq);
 
   SRpcMsg rpcMsg = {.pCont = pReq, .contLen = contLen, .msgType = TDMT_MND_AUTH, .ahandle = (void *)9528};
@@ -543,7 +536,7 @@ static inline int32_t dmRetrieveUserAuthInfo(SDnode *pDnode, char *user, char *s
   return rpcRsp.code;
 }
 
-static int32_t dmInitServer(SDnode *pDnode) {
+int32_t dmInitServer(SDnode *pDnode) {
   SDnodeTrans *pTrans = &pDnode->trans;
 
   SRpcInit rpcInit = {0};
@@ -569,7 +562,7 @@ static int32_t dmInitServer(SDnode *pDnode) {
   return 0;
 }
 
-static void dmCleanupServer(SDnode *pDnode) {
+void dmCleanupServer(SDnode *pDnode) {
   SDnodeTrans *pTrans = &pDnode->trans;
   if (pTrans->serverRpc) {
     rpcClose(pTrans->serverRpc);
@@ -578,17 +571,6 @@ static void dmCleanupServer(SDnode *pDnode) {
   }
 }
 
-int32_t dmInitTrans(SDnode *pDnode) {
-  if (dmInitServer(pDnode) != 0) return -1;
-  if (dmInitClient(pDnode) != 0) return -1;
-  return 0;
-}
-
-void dmCleanupTrans(SDnode *pDnode) {
-  dmCleanupServer(pDnode);
-  dmCleanupClient(pDnode);
-}
-
 SMsgCb dmGetMsgcb(SMgmtWrapper *pWrapper) {
   SMsgCb msgCb = {
       .sendReqFp = dmSendReq,
@@ -598,6 +580,7 @@ SMsgCb dmGetMsgcb(SMgmtWrapper *pWrapper) {
       .releaseHandleFp = dmReleaseHandle,
       .reportStartupFp = dmReportStartupByWrapper,
       .pWrapper = pWrapper,
+      .clientRpc = pWrapper->pDnode->trans.clientRpc,
   };
   return msgCb;
 }

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

@@ -156,6 +156,8 @@ typedef struct SUdfdData {
   uv_pipe_t     ctrlPipe;
   uv_async_t    stopAsync;
   int32_t        stopCalled;
+
+  int32_t         dnodeId;
 } SUdfdData;
 
 typedef struct SDnode {

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

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

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

@@ -171,16 +171,17 @@ static void dmGetServerStatus(SDnode *pDnode, SServerStatusRsp *pStatus) {
   }
 }
 
-void dmProcessNettestReq(SDnode *pDnode, SRpcMsg *pRpc) {
+void dmProcessNetTestReq(SDnode *pDnode, SRpcMsg *pReq) {
   dDebug("net test req is received");
-  SRpcMsg rsp = {.handle = pRpc->handle, .refId = pRpc->refId, .ahandle = pRpc->ahandle, .code = 0};
-  rsp.pCont = rpcMallocCont(pRpc->contLen);
+  SRpcMsg rsp = {.handle = pReq->handle, .refId = pReq->refId, .ahandle = pReq->ahandle, .code = 0};
+  rsp.pCont = rpcMallocCont(pReq->contLen);
   if (rsp.pCont == NULL) {
     rsp.code = TSDB_CODE_OUT_OF_MEMORY;
   } else {
-    rsp.contLen = pRpc->contLen;
+    rsp.contLen = pReq->contLen;
   }
   rpcSendResponse(&rsp);
+  rpcFreeCont(pReq->pCont);
 }
 
 void dmProcessServerStatusReq(SDnode *pDnode, SRpcMsg *pReq) {
@@ -208,6 +209,7 @@ void dmProcessServerStatusReq(SDnode *pDnode, SRpcMsg *pReq) {
 
 _OVER:
   rpcSendResponse(&rspMsg);
+  rpcFreeCont(pReq->pCont);
 }
 
 void dmGetMonitorSysInfo(SMonSysInfo *pInfo) {

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

@@ -289,7 +289,6 @@ void vmInitMsgHandle(SMgmtWrapper *pWrapper) {
   dmSetMsgHandle(pWrapper, TDMT_VND_QUERY, vmProcessQueryMsg, DEFAULT_HANDLE);
   dmSetMsgHandle(pWrapper, TDMT_VND_QUERY_CONTINUE, vmProcessQueryMsg, DEFAULT_HANDLE);
   dmSetMsgHandle(pWrapper, TDMT_VND_FETCH, vmProcessFetchMsg, DEFAULT_HANDLE);
-  dmSetMsgHandle(pWrapper, TDMT_VND_FETCH_RSP, vmProcessFetchMsg, DEFAULT_HANDLE);
   dmSetMsgHandle(pWrapper, TDMT_VND_ALTER_TABLE, vmProcessWriteMsg, DEFAULT_HANDLE);
   dmSetMsgHandle(pWrapper, TDMT_VND_UPDATE_TAG_VAL, vmProcessWriteMsg, DEFAULT_HANDLE);
   dmSetMsgHandle(pWrapper, TDMT_VND_TABLE_META, vmProcessFetchMsg, DEFAULT_HANDLE);