Merge branch '3.0' into test/jcy

contrib/CMakeLists.txt

@@ -14,24 +14,9 @@ if(${BUILD_PTHREAD})
     cat("${TD_SUPPORT_DIR}/pthread_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
 endif()
 
-# iconv
-if(${BUILD_WITH_ICONV})
-    cat("${TD_SUPPORT_DIR}/iconv_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
-endif()
-
-# msvc regex
-if(${BUILD_MSVCREGEX})
-    cat("${TD_SUPPORT_DIR}/msvcregex_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
-endif()
-
-# wcwidth
-if(${BUILD_WCWIDTH})
-    cat("${TD_SUPPORT_DIR}/wcwidth_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
-endif()
-
-# wingetopt
-if(${BUILD_WINGETOPT})
-    cat("${TD_SUPPORT_DIR}/wingetopt_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+# gnu regex
+if(${BUILD_GNUREGEX})
+    cat("${TD_SUPPORT_DIR}/gnuregex_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
 endif()
 
 # googletest
@@ -114,27 +99,8 @@ if(${BUILD_TEST})
     target_include_directories(
         gtest
         PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/cpp-stub/src>
+        PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/cpp-stub/src_linux>
     )
-    if(${TD_WINDOWS})
-        target_include_directories(
-            gtest
-            PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/cpp-stub/src_win>
-        )
-    endif(${TD_WINDOWS})
-    if(${TD_LINUX})
-        target_include_directories(
-            gtest
-            PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/cpp-stub/src_linux>
-        )
-    endif(${TD_LINUX})
-    if(${TD_DARWIN})
-        target_include_directories(
-            gtest
-            PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/cpp-stub/src_darwin>
-        )
-    endif(${TD_DARWIN})
-    
-    
 endif(${BUILD_TEST})
 
 # cJson
@@ -216,53 +182,6 @@ if(${BUILD_WITH_NURAFT})
     add_subdirectory(nuraft)
 endif(${BUILD_WITH_NURAFT})
 
-# pthread
-if(${BUILD_PTHREAD})
-    set(CMAKE_BUILD_TYPE release)
-    add_definitions(-DPTW32_STATIC_LIB) 
-    add_subdirectory(pthread)
-    set_target_properties(libpthreadVC3 PROPERTIES OUTPUT_NAME pthread)
-    add_library(pthread STATIC IMPORTED GLOBAL)
-    SET_PROPERTY(TARGET pthread PROPERTY IMPORTED_LOCATION ${LIBRARY_OUTPUT_PATH}/pthread.lib)
-endif()
-
-# iconv
-if(${BUILD_WITH_ICONV})
-    add_subdirectory(iconv)
-endif(${BUILD_WITH_ICONV})
-
-# wingetopt
-if(${BUILD_WINGETOPT})
-    add_subdirectory(wingetopt)
-endif(${BUILD_WINGETOPT})
-
-# msvcregex
-if(${BUILD_MSVCREGEX})
-    add_library(msvcregex STATIC "")
-    target_sources(msvcregex
-        PRIVATE "msvcregex/regex.c"
-    )
-    target_include_directories(msvcregex
-        PRIVATE "msvcregex"
-    )
-    target_link_libraries(msvcregex 
-        INTERFACE Shell32
-    )
-    SET_TARGET_PROPERTIES(msvcregex PROPERTIES OUTPUT_NAME msvcregex)
-endif(${BUILD_MSVCREGEX})
-
-# msvcregex
-if(${BUILD_WCWIDTH})
-    add_library(wcwidth STATIC "")
-    target_sources(wcwidth
-        PRIVATE "wcwidth/wcwidth.c"
-    )
-    target_include_directories(wcwidth
-        PRIVATE "wcwidth"
-    )
-    SET_TARGET_PROPERTIES(wcwidth PROPERTIES OUTPUT_NAME wcwidth)
-endif(${BUILD_WCWIDTH})
-
 # CRAFT
 if(${BUILD_WITH_CRAFT})
     add_library(craft STATIC IMPORTED GLOBAL)
@@ -319,12 +238,8 @@ if(${BUILD_WITH_SQLITE})
     target_link_libraries(sqlite 
         INTERFACE m 
         INTERFACE pthread 
+        INTERFACE dl
     )
-    if(NOT TD_WINDOWS)
-        target_link_libraries(sqlite 
-            INTERFACE dl
-        )
-    endif(NOT TD_WINDOWS)
 endif(${BUILD_WITH_SQLITE})
 
 # pthread

contrib/test/craft/CMakeLists.txt

 add_executable(simulate_vnode "simulate_vnode.c")
-target_link_libraries(simulate_vnode PUBLIC craft lz4 uv_a)
-if(${BUILD_WINGETOPT})
+target_link_libraries(simulate_vnode PUBLIC craft lz4 uv_a)
\ No newline at end of file
-    target_link_libraries(simulate_vnode PUBLIC wingetopt)
-    target_include_directories(
-        simulate_vnode
-        PUBLIC "${TD_SOURCE_DIR}/contrib/wingetopt/src"
-    )
-endif()
\ No newline at end of file

contrib/test/tdev/src/main.c

@@ -6,39 +6,43 @@
 #define POINTER_SHIFT(ptr, s) ((void *)(((char *)ptr) + (s)))
 #define POINTER_DISTANCE(pa, pb) ((char *)(pb) - (char *)(pa))
 
-static inline void tPutA(void **buf, uint64_t val) {
-  memcpy(buf, &val, sizeof(val));
-  *buf = POINTER_SHIFT(buf, sizeof(val));
-}
+#define tPutA(buf, val)              \
+  ({                                 \
+    memcpy(buf, &val, sizeof(val));  \
+    POINTER_SHIFT(buf, sizeof(val)); \
+  })
 
-static inline void tPutB(void **buf, uint64_t val) {
-  ((uint8_t *)buf)[7] = ((val) >> 56) & 0xff;
-  ((uint8_t *)buf)[6] = ((val) >> 48) & 0xff;
-  ((uint8_t *)buf)[5] = ((val) >> 40) & 0xff;
-  ((uint8_t *)buf)[4] = ((val) >> 32) & 0xff;
-  ((uint8_t *)buf)[3] = ((val) >> 24) & 0xff;
-  ((uint8_t *)buf)[2] = ((val) >> 16) & 0xff;
-  ((uint8_t *)buf)[1] = ((val) >> 8) & 0xff;
-  ((uint8_t *)buf)[0] = (val)&0xff;
-  *buf = POINTER_SHIFT(buf, sizeof(val));
-}
+#define tPutB(buf, val)                         \
+  ({                                            \
+    ((uint8_t *)buf)[7] = ((val) >> 56) & 0xff; \
+    ((uint8_t *)buf)[6] = ((val) >> 48) & 0xff; \
+    ((uint8_t *)buf)[5] = ((val) >> 40) & 0xff; \
+    ((uint8_t *)buf)[4] = ((val) >> 32) & 0xff; \
+    ((uint8_t *)buf)[3] = ((val) >> 24) & 0xff; \
+    ((uint8_t *)buf)[2] = ((val) >> 16) & 0xff; \
+    ((uint8_t *)buf)[1] = ((val) >> 8) & 0xff;  \
+    ((uint8_t *)buf)[0] = (val)&0xff;           \
+    POINTER_SHIFT(buf, sizeof(val));            \
+  })
 
-static inline void tPutC(void **buf, uint64_t val) {
-  if (buf) {
-    ((uint64_t *)buf)[0] = (val);
-    POINTER_SHIFT(buf, sizeof(val));
-  }
-  *buf = NULL;
-}
+#define tPutC(buf, val)                \
+  ({                                   \
+    if (buf) {                         \
+      ((uint64_t *)buf)[0] = (val);    \
+      POINTER_SHIFT(buf, sizeof(val)); \
+    }                                  \
+    NULL;                                 \
+  })
 
-static inline void tPutD(void **buf, uint64_t val) {
-  uint64_t tmp = val;
-  for (size_t i = 0; i < sizeof(val); i++) {
-    ((uint8_t *)buf)[i] = tmp & 0xff;
-    tmp >>= 8;
-  }
-  *buf = POINTER_SHIFT(buf, sizeof(val));
-}
+#define tPutD(buf, val)                        \
+  ({                                           \
+    uint64_t tmp = val;                        \
+    for (size_t i = 0; i < sizeof(val); i++) { \
+      ((uint8_t *)buf)[i] = tmp & 0xff;        \
+      tmp >>= 8;                               \
+    }                                          \
+    POINTER_SHIFT(buf, sizeof(val));           \
+  })
 
 static inline void tPutE(void **buf, uint64_t val) {
   if (buf) {
@@ -57,7 +61,7 @@ static void func(T t) {
   switch (t) {
     case A:
       for (size_t i = 0; i < 10 * 1024l * 1024l * 1024l; i++) {
-        tPutA(pBuf, val);
+        pBuf = tPutA(pBuf, val);
         if (POINTER_DISTANCE(buf, pBuf) == 1024) {
           pBuf = buf;
         }
@@ -65,7 +69,7 @@ static void func(T t) {
       break;
     case B:
       for (size_t i = 0; i < 10 * 1024l * 1024l * 1024l; i++) {
-        tPutB(pBuf, val);
+        pBuf = tPutB(pBuf, val);
         if (POINTER_DISTANCE(buf, pBuf) == 1024) {
           pBuf = buf;
         }
@@ -73,7 +77,7 @@ static void func(T t) {
       break;
     case C:
       for (size_t i = 0; i < 10 * 1024l * 1024l * 1024l; i++) {
-        tPutC(pBuf, val);
+        pBuf = tPutC(pBuf, val);
         if (POINTER_DISTANCE(buf, pBuf) == 1024) {
           pBuf = buf;
         }
@@ -81,7 +85,7 @@ static void func(T t) {
       break;
     case D:
       for (size_t i = 0; i < 10 * 1024l * 1024l * 1024l; i++) {
-        tPutD(pBuf, val);
+        pBuf = tPutD(pBuf, val);
         if (POINTER_DISTANCE(buf, pBuf) == 1024) {
           pBuf = buf;
         }

example/src/tmq.c

@@ -17,6 +17,7 @@
 #include <stdio.h>
 #include <string.h>
 #include <time.h>
+#include <unistd.h>
 #include "taos.h"
 
 static int  running = 1;
@@ -47,6 +48,7 @@ int32_t init_env() {
     return -1;
   }
   taos_free_result(pRes);
+  sleep(1);
 
   pRes = taos_query(pConn, "use abc1");
   if (taos_errno(pRes) != 0) {
@@ -58,6 +60,7 @@ int32_t init_env() {
   pRes =
       taos_query(pConn, "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(10)) tags(t1 int)");
   if (taos_errno(pRes) != 0) {
+    assert(0);
     printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes));
     return -1;
   }
@@ -265,10 +268,11 @@ void perf_loop(tmq_t* tmq, tmq_list_t* topics) {
 }
 
 int main(int argc, char* argv[]) {
-  int code;
   if (argc > 1) {
     printf("env init\n");
-    code = init_env();
+    if (init_env() < 0) {
+      return -1;
+    }
     create_topic();
   }
   tmq_t*      tmq = build_consumer();

include/common/tdatablock.h

@@ -238,10 +238,16 @@ static FORCE_INLINE int32_t blockCompressColData(SColumnInfoData* pColRes, int32
 
 static FORCE_INLINE void blockCompressEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_t numOfCols,
                                              int8_t needCompress) {
-  int32_t* colSizes = (int32_t*)data;
+  int32_t* actualLen = (int32_t*) data;
+  data += sizeof(int32_t);
+
+  uint64_t* groupId = (uint64_t*) data;
+  data += sizeof(uint64_t);
 
+  int32_t* colSizes = (int32_t*)data;
   data += numOfCols * sizeof(int32_t);
-  *dataLen = (numOfCols * sizeof(int32_t));
+
+  *dataLen = (numOfCols * sizeof(int32_t) + sizeof(uint64_t) + sizeof(int32_t));
 
   int32_t numOfRows = pBlock->info.rows;
   for (int32_t col = 0; col < numOfCols; ++col) {
@@ -273,6 +279,9 @@ static FORCE_INLINE void blockCompressEncode(const SSDataBlock* pBlock, char* da
 
     colSizes[col] = htonl(colSizes[col]);
   }
+
+  *actualLen = *dataLen;
+  *groupId = pBlock->info.groupId;
 }
 
 #ifdef __cplusplus

include/common/tmsg.h

@@ -1314,7 +1314,6 @@ typedef struct {
 } SMqConsumerLostMsg;
 
 typedef struct {
-  int32_t topicNum;
   int64_t consumerId;
   char    cgroup[TSDB_CGROUP_LEN];
   SArray* topicNames;  // SArray<char*>
@@ -1322,22 +1321,27 @@ typedef struct {
 
 static FORCE_INLINE int32_t tSerializeSCMSubscribeReq(void** buf, const SCMSubscribeReq* pReq) {
   int32_t tlen = 0;
-  tlen += taosEncodeFixedI32(buf, pReq->topicNum);
   tlen += taosEncodeFixedI64(buf, pReq->consumerId);
   tlen += taosEncodeString(buf, pReq->cgroup);
 
-  for (int32_t i = 0; i < pReq->topicNum; i++) {
+  int32_t topicNum = taosArrayGetSize(pReq->topicNames);
+  tlen += taosEncodeFixedI32(buf, topicNum);
+
+  for (int32_t i = 0; i < topicNum; i++) {
     tlen += taosEncodeString(buf, (char*)taosArrayGetP(pReq->topicNames, i));
   }
   return tlen;
 }
 
 static FORCE_INLINE void* tDeserializeSCMSubscribeReq(void* buf, SCMSubscribeReq* pReq) {
-  buf = taosDecodeFixedI32(buf, &pReq->topicNum);
   buf = taosDecodeFixedI64(buf, &pReq->consumerId);
   buf = taosDecodeStringTo(buf, pReq->cgroup);
-  pReq->topicNames = taosArrayInit(pReq->topicNum, sizeof(void*));
-  for (int32_t i = 0; i < pReq->topicNum; i++) {
+
+  int32_t topicNum;
+  buf = taosDecodeFixedI32(buf, &topicNum);
+
+  pReq->topicNames = taosArrayInit(topicNum, sizeof(void*));
+  for (int32_t i = 0; i < topicNum; i++) {
     char* name;
     buf = taosDecodeString(buf, &name);
     taosArrayPush(pReq->topicNames, &name);
@@ -1969,7 +1973,6 @@ typedef struct {
   int8_t  withTbName;
   int8_t  withSchema;
   int8_t  withTag;
-  int8_t  withTagSchema;
   char*   qmsg;
 } SMqRebVgReq;
 
@@ -1984,7 +1987,6 @@ static FORCE_INLINE int32_t tEncodeSMqRebVgReq(void** buf, const SMqRebVgReq* pR
   tlen += taosEncodeFixedI8(buf, pReq->withTbName);
   tlen += taosEncodeFixedI8(buf, pReq->withSchema);
   tlen += taosEncodeFixedI8(buf, pReq->withTag);
-  tlen += taosEncodeFixedI8(buf, pReq->withTagSchema);
   if (pReq->subType == TOPIC_SUB_TYPE__TABLE) {
     tlen += taosEncodeString(buf, pReq->qmsg);
   }
@@ -2001,7 +2003,6 @@ static FORCE_INLINE void* tDecodeSMqRebVgReq(const void* buf, SMqRebVgReq* pReq)
   buf = taosDecodeFixedI8(buf, &pReq->withTbName);
   buf = taosDecodeFixedI8(buf, &pReq->withSchema);
   buf = taosDecodeFixedI8(buf, &pReq->withTag);
-  buf = taosDecodeFixedI8(buf, &pReq->withTagSchema);
   if (pReq->subType == TOPIC_SUB_TYPE__TABLE) {
     buf = taosDecodeString(buf, &pReq->qmsg);
   }
@@ -2590,7 +2591,6 @@ typedef struct {
   int8_t     withTbName;
   int8_t     withSchema;
   int8_t     withTag;
-  int8_t     withTagSchema;
   SArray*    blockDataLen;    // SArray<int32_t>
   SArray*    blockData;       // SArray<SRetrieveTableRsp*>
   SArray*    blockTbName;     // SArray<char*>
@@ -2609,7 +2609,6 @@ static FORCE_INLINE int32_t tEncodeSMqDataBlkRsp(void** buf, const SMqDataBlkRsp
     tlen += taosEncodeFixedI8(buf, pRsp->withTbName);
     tlen += taosEncodeFixedI8(buf, pRsp->withSchema);
     tlen += taosEncodeFixedI8(buf, pRsp->withTag);
-    tlen += taosEncodeFixedI8(buf, pRsp->withTagSchema);
 
     for (int32_t i = 0; i < pRsp->blockNum; i++) {
       int32_t bLen = *(int32_t*)taosArrayGet(pRsp->blockDataLen, i);
@@ -2632,7 +2631,6 @@ static FORCE_INLINE void* tDecodeSMqDataBlkRsp(const void* buf, SMqDataBlkRsp* p
     buf = taosDecodeFixedI8(buf, &pRsp->withTbName);
     buf = taosDecodeFixedI8(buf, &pRsp->withSchema);
     buf = taosDecodeFixedI8(buf, &pRsp->withTag);
-    buf = taosDecodeFixedI8(buf, &pRsp->withTagSchema);
 
     for (int32_t i = 0; i < pRsp->blockNum; i++) {
       int32_t bLen = 0;

include/libs/function/function.h

@@ -37,7 +37,7 @@ typedef struct SFuncExecEnv {
 typedef bool (*FExecGetEnv)(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 typedef bool (*FExecInit)(struct SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo);
 typedef int32_t (*FExecProcess)(struct SqlFunctionCtx *pCtx);
-typedef int32_t (*FExecFinalize)(struct SqlFunctionCtx *pCtx, SSDataBlock* pBlock, int32_t slotId);
+typedef int32_t (*FExecFinalize)(struct SqlFunctionCtx *pCtx, SSDataBlock* pBlock);
 typedef int32_t (*FScalarExecProcess)(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
 
 typedef struct SScalarFuncExecFuncs {
@@ -141,8 +141,7 @@ struct SResultRowEntryInfo;
 
 //for selectivity query, the corresponding tag value is assigned if the data is qualified
 typedef struct SSubsidiaryResInfo {
-  int16_t                 bufLen;      // keep the tags data for top/bottom query result
-  int16_t                 numOfCols;
+  int16_t num;
   struct SqlFunctionCtx **pCtx;
 } SSubsidiaryResInfo;
 
@@ -187,8 +186,8 @@ typedef struct SqlFunctionCtx {
   uint8_t          currentStage;  // record current running step, default: 0
   bool             isAggSet;
   int64_t          startTs;       // timestamp range of current query when function is executed on a specific data block, TODO remove it
-  /////////////////////////////////////////////////////////////////
   bool             stableQuery;
+  /////////////////////////////////////////////////////////////////
   int16_t          functionId;    // function id
   char *           pOutput;       // final result output buffer, point to sdata->data
   int32_t          numOfParams;
@@ -198,11 +197,15 @@ typedef struct SqlFunctionCtx {
   int32_t          offset;
   SVariant         tag;
   struct  SResultRowEntryInfo *resultInfo;
-  SSubsidiaryResInfo     subsidiaryRes;
-  SPoint1          start;
-  SPoint1          end;
-  SFuncExecFuncs   fpSet;
-  SScalarFuncExecFuncs sfp;
+  SSubsidiaryResInfo     subsidiaries;
+  SPoint1                start;
+  SPoint1                end;
+  SFuncExecFuncs         fpSet;
+  SScalarFuncExecFuncs   sfp;
+  SExprInfo             *pExpr;
+  struct SDiskbasedBuf  *pBuf;
+  struct SSDataBlock    *pSrcBlock;
+  int32_t                curBufPage;
 } SqlFunctionCtx;
 
 enum {
@@ -319,6 +322,20 @@ struct SUdfInfo;
 void qAddUdfInfo(uint64_t id, struct SUdfInfo* pUdfInfo);
 void qRemoveUdfInfo(uint64_t id, struct SUdfInfo* pUdfInfo);
 
+/**
+ * create udfd proxy, called once in process that call setupUdf/callUdfxxx/teardownUdf
+ * @return error code
+ */
+int32_t udfcOpen();
+
+/**
+ * destroy udfd proxy
+ * @return error code
+ */
+int32_t udfcClose();
+
+typedef void *UdfcFuncHandle;
+
 #ifdef __cplusplus
 }
 #endif

include/libs/wal/wal.h

@@ -192,7 +192,13 @@ int32_t walEndSnapshot(SWal *);
 SWalReadHandle *walOpenReadHandle(SWal *);
 void            walCloseReadHandle(SWalReadHandle *);
 int32_t         walReadWithHandle(SWalReadHandle *pRead, int64_t ver);
-int32_t         walReadWithHandle_s(SWalReadHandle *pRead, int64_t ver, SWalReadHead **ppHead);
+
+// only for tq usage
+// int32_t walReadWithHandle_s(SWalReadHandle *pRead, int64_t ver, SWalReadHead **ppHead);
+void    walSetReaderCapacity(SWalReadHandle *pRead, int32_t capacity);
+int32_t walFetchHead(SWalReadHandle *pRead, int64_t ver, SWalHead *pHead);
+int32_t walFetchBody(SWalReadHandle *pRead, SWalHead **ppHead);
+int32_t walSkipFetchBody(SWalReadHandle *pRead, const SWalHead *pHead);
 
 // deprecated
 #if 0

include/util/taoserror.h

@@ -252,7 +252,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_MND_COLUMN_NOT_EXIST          TAOS_DEF_ERROR_CODE(0, 0x03AE)
 
 // mnode-infoSchema
-#define TSDB_CODE_MND_INVALID_INFOS_TBL         TAOS_DEF_ERROR_CODE(0, 0x03B0)
+#define TSDB_CODE_MND_INVALID_SYS_TABLENAME         TAOS_DEF_ERROR_CODE(0, 0x03B0)
 
 // mnode-func
 #define TSDB_CODE_MND_FUNC_ALREADY_EXIST        TAOS_DEF_ERROR_CODE(0, 0x03C0)

source/client/src/clientImpl.c

@@ -105,9 +105,9 @@ TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass,
     epSet.epSet.eps[0].port = port;
   }
 
-  char*          key = getClusterKey(user, secretEncrypt, ip, port);
-  SAppInstInfo** pInst = NULL;
+  char* key = getClusterKey(user, secretEncrypt, ip, port);
 
+  SAppInstInfo** pInst = NULL;
   taosThreadMutexLock(&appInfo.mutex);
 
   pInst = taosHashGet(appInfo.pInstMap, key, strlen(key));
@@ -840,10 +840,21 @@ int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32
     return code;
   }
 
-  int32_t* colLength = (int32_t*)pResultInfo->pData;
-  char*    pStart = ((char*)pResultInfo->pData) + sizeof(int32_t) * numOfCols;
+  char* p = (char*) pResultInfo->pData;
+
+  int32_t dataLen = *(int32_t*) p;
+  p += sizeof(int32_t);
+
+  uint64_t groupId = *(uint64_t*) p;
+  p += sizeof(uint64_t);
+
+  int32_t* colLength = (int32_t*)p;
+  p += sizeof(int32_t) * numOfCols;
+
+  char* pStart = p;
   for (int32_t i = 0; i < numOfCols; ++i) {
     colLength[i] = htonl(colLength[i]);
+    ASSERT(colLength[i] < dataLen);
 
     if (IS_VAR_DATA_TYPE(pResultInfo->fields[i].type)) {
       pResultInfo->pCol[i].offset = (int32_t*)pStart;

source/client/src/clientMsgHandler.c

@@ -13,13 +13,13 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#include "os.h"
-#include "tdef.h"
-#include "tname.h"
+#include "catalog.h"
 #include "clientInt.h"
 #include "clientLog.h"
-#include "catalog.h"
+#include "os.h"
 #include "query.h"
+#include "tdef.h"
+#include "tname.h"
 
 int32_t (*handleRequestRspFp[TDMT_MAX])(void*, const SDataBuf* pMsg, int32_t code);
 
@@ -50,7 +50,13 @@ int32_t processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
 
   SConnectRsp connectRsp = {0};
   tDeserializeSConnectRsp(pMsg->pData, pMsg->len, &connectRsp);
-  assert(connectRsp.epSet.numOfEps > 0);
+  /*assert(connectRsp.epSet.numOfEps > 0);*/
+  if (connectRsp.epSet.numOfEps == 0) {
+    taosMemoryFree(pMsg->pData);
+    setErrno(pRequest, TSDB_CODE_MND_APP_ERROR);
+    tsem_post(&pRequest->body.rspSem);
+    return code;
+  }
 
   if (!isEpsetEqual(&pTscObj->pAppInfo->mgmtEp.epSet, &connectRsp.epSet)) {
     updateEpSet_s(&pTscObj->pAppInfo->mgmtEp, &connectRsp.epSet);
@@ -82,18 +88,20 @@ int32_t processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
   return 0;
 }
 
-SMsgSendInfo* buildMsgInfoImpl(SRequestObj *pRequest) {
+SMsgSendInfo* buildMsgInfoImpl(SRequestObj* pRequest) {
   SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
 
   pMsgSendInfo->requestObjRefId = pRequest->self;
-  pMsgSendInfo->requestId       = pRequest->requestId;
-  pMsgSendInfo->param           = pRequest;
-  pMsgSendInfo->msgType         = pRequest->type;
+  pMsgSendInfo->requestId = pRequest->requestId;
+  pMsgSendInfo->param = pRequest;
+  pMsgSendInfo->msgType = pRequest->type;
 
   assert(pRequest != NULL);
   pMsgSendInfo->msgInfo = pRequest->body.requestMsg;
 
-  pMsgSendInfo->fp = (handleRequestRspFp[TMSG_INDEX(pRequest->type)] == NULL)? genericRspCallback:handleRequestRspFp[TMSG_INDEX(pRequest->type)];
+  pMsgSendInfo->fp = (handleRequestRspFp[TMSG_INDEX(pRequest->type)] == NULL)
+                         ? genericRspCallback
+                         : handleRequestRspFp[TMSG_INDEX(pRequest->type)];
   return pMsgSendInfo;
 }
 
@@ -114,7 +122,7 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
   if (TSDB_CODE_MND_DB_NOT_EXIST == code) {
     SUseDbRsp usedbRsp = {0};
     tDeserializeSUseDbRsp(pMsg->pData, pMsg->len, &usedbRsp);
-    struct SCatalog *pCatalog = NULL;
+    struct SCatalog* pCatalog = NULL;
 
     if (usedbRsp.vgVersion >= 0) {
       int32_t code1 = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);

source/client/src/tmq.c

@@ -382,12 +382,9 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
   pTmq->pTscObj = taos_connect_internal(conf->ip, user, pass, NULL, conf->db, conf->port, CONN_TYPE__TMQ);
   if (pTmq->pTscObj == NULL) return NULL;
 
-  /*pTmq->inWaiting = 0;*/
   pTmq->status = 0;
   pTmq->pollCnt = 0;
   pTmq->epoch = 0;
-  /*pTmq->waitingRequest = 0;*/
-  /*pTmq->readyRequest = 0;*/
   pTmq->epStatus = 0;
   pTmq->epSkipCnt = 0;
   // set conf
@@ -509,7 +506,6 @@ tmq_resp_err_t tmq_subscribe(tmq_t* tmq, tmq_list_t* topic_list) {
   tmq->clientTopics = taosArrayInit(sz, sizeof(SMqClientTopic));
 
   SCMSubscribeReq req;
-  req.topicNum = sz;
   req.consumerId = tmq->consumerId;
   strcpy(req.cgroup, tmq->groupId);
   req.topicNames = taosArrayInit(sz, sizeof(void*));
@@ -519,12 +515,16 @@ tmq_resp_err_t tmq_subscribe(tmq_t* tmq, tmq_list_t* topic_list) {
     char* topicName = taosArrayGetP(container, i);
 
     SName name = {0};
+#if 0
     char* dbName = getDbOfConnection(tmq->pTscObj);
     if (dbName == NULL) {
       return TMQ_RESP_ERR__FAIL;
     }
-    tNameSetDbName(&name, tmq->pTscObj->acctId, dbName, strlen(dbName));
+#endif
+    tNameSetDbName(&name, tmq->pTscObj->acctId, topicName, strlen(topicName));
+#if 0
     tNameFromString(&name, topicName, T_NAME_TABLE);
+#endif
 
     char* topicFname = taosMemoryCalloc(1, TSDB_TOPIC_FNAME_LEN);
     if (topicFname == NULL) {
@@ -542,7 +542,9 @@ tmq_resp_err_t tmq_subscribe(tmq_t* tmq, tmq_list_t* topic_list) {
     topic.vgs = taosArrayInit(0, sizeof(SMqClientVg));
     taosArrayPush(tmq->clientTopics, &topic);
     taosArrayPush(req.topicNames, &topicFname);
+#if 0
     taosMemoryFree(dbName);
+#endif
   }
 
   int   tlen = tSerializeSCMSubscribeReq(NULL, &req);

source/common/src/tdatablock.c

@@ -644,12 +644,12 @@ size_t blockDataGetRowSize(SSDataBlock* pBlock) {
 
 /**
  * @refitem blockDataToBuf for the meta size
- *
  * @param pBlock
  * @return
  */
 size_t blockDataGetSerialMetaSize(const SSDataBlock* pBlock) {
-  return sizeof(int32_t) + pBlock->info.numOfCols * sizeof(int32_t);
+  // | total rows/total length | block group id | each column length |
+  return sizeof(int32_t) + sizeof(uint64_t) + pBlock->info.numOfCols * sizeof(int32_t);
 }
 
 double blockDataGetSerialRowSize(const SSDataBlock* pBlock) {
@@ -1219,7 +1219,6 @@ void colDataDestroy(SColumnInfoData* pColData) {
   taosMemoryFree(pColData->pData);
 }
 
-
 static void doShiftBitmap(char* nullBitmap, size_t n, size_t total) {
   int32_t len = BitmapLen(total);
 

source/common/src/tmsg.c

@@ -126,7 +126,6 @@ int32_t tDecodeSQueryNodeAddr(SCoder *pDecoder, SQueryNodeAddr *pAddr) {
   return 0;
 }
 
-
 int32_t taosEncodeSEpSet(void **buf, const SEpSet *pEp) {
   int32_t tlen = 0;
   tlen += taosEncodeFixedI8(buf, pEp->inUse);
@@ -2745,11 +2744,11 @@ int32_t tSerializeSCMCreateTopicReq(void *buf, int32_t bufLen, const SCMCreateTo
   if (tEncodeI8(&encoder, pReq->withTbName) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->withSchema) < 0) return -1;
   if (tEncodeI8(&encoder, pReq->withTag) < 0) return -1;
+  if (tEncodeCStr(&encoder, pReq->subscribeDbName) < 0) return -1;
   if (tEncodeI32(&encoder, sqlLen) < 0) return -1;
   if (tEncodeI32(&encoder, astLen) < 0) return -1;
   if (sqlLen > 0 && tEncodeCStr(&encoder, pReq->sql) < 0) return -1;
   if (astLen > 0 && tEncodeCStr(&encoder, pReq->ast) < 0) return -1;
-  if (0 == astLen && tEncodeCStr(&encoder, pReq->subscribeDbName) < 0) return -1;
 
   tEndEncode(&encoder);
 
@@ -2771,6 +2770,7 @@ int32_t tDeserializeSCMCreateTopicReq(void *buf, int32_t bufLen, SCMCreateTopicR
   if (tDecodeI8(&decoder, &pReq->withTbName) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->withSchema) < 0) return -1;
   if (tDecodeI8(&decoder, &pReq->withTag) < 0) return -1;
+  if (tDecodeCStrTo(&decoder, pReq->subscribeDbName) < 0) return -1;
   if (tDecodeI32(&decoder, &sqlLen) < 0) return -1;
   if (tDecodeI32(&decoder, &astLen) < 0) return -1;
 
@@ -2785,7 +2785,6 @@ int32_t tDeserializeSCMCreateTopicReq(void *buf, int32_t bufLen, SCMCreateTopicR
     if (pReq->ast == NULL) return -1;
     if (tDecodeCStrTo(&decoder, pReq->ast) < 0) return -1;
   } else {
-    if (tDecodeCStrTo(&decoder, pReq->subscribeDbName) < 0) return -1;
   }
 
   tEndDecode(&decoder);

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

@@ -15,6 +15,7 @@
 
 #define _DEFAULT_SOURCE
 #include "vmInt.h"
+#include "libs/function/function.h"
 
 SVnodeObj *vmAcquireVnode(SVnodesMgmt *pMgmt, int32_t vgId) {
   SVnodeObj *pVnode = NULL;
@@ -275,7 +276,7 @@ static void vmCleanup(SMgmtWrapper *pWrapper) {
   pWrapper->pMgmt = NULL;
 
   // syncCleanUp();
-
+  udfcClose();
   dInfo("vnode-mgmt is cleaned up");
 }
 
@@ -339,6 +340,10 @@ static int32_t vmInit(SMgmtWrapper *pWrapper) {
   }
   dmReportStartup(pDnode, "vnode-vnodes", "initialized");
 
+  if (udfcOpen() != 0) {
+    dError("failed to open udfc in dnode");
+  }
+
   code = 0;
 
 _OVER:

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

@@ -450,7 +450,6 @@ typedef struct {
   int8_t         withTbName;
   int8_t         withSchema;
   int8_t         withTag;
-  int8_t         withTagSchema;
   SRWLatch       lock;
   int32_t        sqlLen;
   int32_t        astLen;
@@ -517,7 +516,6 @@ typedef struct {
   int8_t    withTbName;
   int8_t    withSchema;
   int8_t    withTag;
-  int8_t    withTagSchema;
   SHashObj* consumerHash;  // consumerId -> SMqConsumerEpInSub
   // TODO put -1 into unassignVgs
   // SArray*   unassignedVgs;

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

@@ -237,7 +237,6 @@ SMqSubscribeObj *tCloneSubscribeObj(const SMqSubscribeObj *pSub) {
   pSubNew->withTbName = pSub->withTbName;
   pSubNew->withSchema = pSub->withSchema;
   pSubNew->withTag = pSub->withTag;
-  pSubNew->withTagSchema = pSub->withTagSchema;
 
   pSubNew->vgNum = pSub->vgNum;
   pSubNew->consumerHash = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
@@ -270,7 +269,6 @@ int32_t tEncodeSubscribeObj(void **buf, const SMqSubscribeObj *pSub) {
   tlen += taosEncodeFixedI8(buf, pSub->withTbName);
   tlen += taosEncodeFixedI8(buf, pSub->withSchema);
   tlen += taosEncodeFixedI8(buf, pSub->withTag);
-  tlen += taosEncodeFixedI8(buf, pSub->withTagSchema);
 
   void   *pIter = NULL;
   int32_t sz = taosHashGetSize(pSub->consumerHash);
@@ -297,7 +295,6 @@ void *tDecodeSubscribeObj(const void *buf, SMqSubscribeObj *pSub) {
   buf = taosDecodeFixedI8(buf, &pSub->withTbName);
   buf = taosDecodeFixedI8(buf, &pSub->withSchema);
   buf = taosDecodeFixedI8(buf, &pSub->withTag);
-  buf = taosDecodeFixedI8(buf, &pSub->withTagSchema);
 
   int32_t sz;
   buf = taosDecodeFixedI32(buf, &sz);

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

@@ -325,7 +325,7 @@ static int32_t mndInsInitMeta(SHashObj *hash) {
       return -1;
     }
 
-    if (taosHashPut(hash, meta.tbName, strlen(meta.tbName) + 1, &meta, sizeof(meta))) {
+    if (taosHashPut(hash, meta.tbName, strlen(meta.tbName), &meta, sizeof(meta))) {
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       return -1;
     }
@@ -340,10 +340,10 @@ int32_t mndBuildInsTableSchema(SMnode *pMnode, const char *dbFName, const char *
     return -1;
   }
 
-  STableMetaRsp *pMeta = taosHashGet(pMnode->infosMeta, tbName, strlen(tbName) + 1);
+  STableMetaRsp *pMeta = taosHashGet(pMnode->infosMeta, tbName, strlen(tbName));
   if (NULL == pMeta) {
     mError("invalid information schema table name:%s", tbName);
-    terrno = TSDB_CODE_MND_INVALID_INFOS_TBL;
+    terrno = TSDB_CODE_MND_INVALID_SYS_TABLENAME;
     return -1;
   }
 

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

@@ -128,7 +128,7 @@ int32_t mndBuildPerfsTableSchema(SMnode *pMnode, const char *dbFName, const char
   STableMetaRsp *meta = (STableMetaRsp *)taosHashGet(pMnode->perfsMeta, tbName, strlen(tbName));
   if (NULL == meta) {
     mError("invalid performance schema table name:%s", tbName);
-    terrno = TSDB_CODE_MND_INVALID_INFOS_TBL;
+    terrno = TSDB_CODE_MND_INVALID_SYS_TABLENAME;
     return -1;
   }
 

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

@@ -100,6 +100,8 @@ static int32_t convertToRetrieveType(char* name, int32_t len) {
     type = TSDB_MGMT_TABLE_QUERIES;
   }  else if (strncasecmp(name, TSDB_INS_TABLE_VNODES, len) == 0) {
     type = TSDB_MGMT_TABLE_VNODES;
+  } else if (strncasecmp(name, TSDB_PERFS_TABLE_TOPICS, len) == 0) {
+    type = TSDB_MGMT_TABLE_TOPICS;
   } else {
 //    ASSERT(0);
   }
@@ -187,11 +189,14 @@ static int32_t mndProcessRetrieveSysTableReq(SNodeMsg *pReq) {
   }
 
   if (retrieveReq.showId == 0) {
-    STableMetaRsp *pMeta = (STableMetaRsp *)taosHashGet(pMnode->infosMeta, retrieveReq.tb, strlen(retrieveReq.tb) + 1);
+    STableMetaRsp *pMeta = (STableMetaRsp *)taosHashGet(pMnode->infosMeta, retrieveReq.tb, strlen(retrieveReq.tb));
     if (pMeta == NULL) {
-      terrno = TSDB_CODE_MND_INVALID_INFOS_TBL;
-      mError("failed to process show-retrieve req:%p since %s", pShow, terrstr());
-      return -1;
+      pMeta = (STableMetaRsp *)taosHashGet(pMnode->perfsMeta, retrieveReq.tb, strlen(retrieveReq.tb));
+      if (pMeta == NULL) {
+        terrno = TSDB_CODE_MND_INVALID_SYS_TABLENAME;
+        mError("failed to process show-retrieve req:%p since %s", pShow, terrstr());
+        return -1;
+      }
     }
 
     pShow = mndCreateShowObj(pMnode, &retrieveReq);

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

@@ -35,11 +35,6 @@
 
 #define MND_SUBSCRIBE_REBALANCE_CNT 3
 
-enum {
-  MQ_SUBSCRIBE_STATUS__ACTIVE = 1,
-  MQ_SUBSCRIBE_STATUS__DELETED,
-};
-
 static SSdbRaw *mndSubActionEncode(SMqSubscribeObj *);
 static SSdbRow *mndSubActionDecode(SSdbRaw *pRaw);
 static int32_t  mndSubActionInsert(SSdb *pSdb, SMqSubscribeObj *);
@@ -89,7 +84,6 @@ static SMqSubscribeObj *mndCreateSub(SMnode *pMnode, const SMqTopicObj *pTopic,
   pSub->withTbName = pTopic->withTbName;
   pSub->withSchema = pTopic->withSchema;
   pSub->withTag = pTopic->withTag;
-  pSub->withTagSchema = pTopic->withTagSchema;
 
   ASSERT(taosHashGetSize(pSub->consumerHash) == 1);
 
@@ -115,7 +109,6 @@ static int32_t mndBuildSubChangeReq(void **pBuf, int32_t *pLen, const SMqSubscri
   req.withTbName = pSub->withTbName;
   req.withSchema = pSub->withSchema;
   req.withTag = pSub->withTag;
-  req.withTagSchema = pSub->withTagSchema;
   strncpy(req.subKey, pSub->key, TSDB_SUBSCRIBE_KEY_LEN);
 
   int32_t tlen = sizeof(SMsgHead) + tEncodeSMqRebVgReq(NULL, &req);
@@ -514,9 +507,11 @@ static int32_t mndProcessRebalanceReq(SNodeMsg *pMsg) {
 
     // TODO replace assert with error check
     ASSERT(mndDoRebalance(pMnode, &rebInput, &rebOutput) == 0);
+
     // if add more consumer to balanced subscribe,
     // possibly no vg is changed
     /*ASSERT(taosArrayGetSize(rebOutput.rebVgs) != 0);*/
+
     ASSERT(mndPersistRebResult(pMnode, pMsg, &rebOutput) == 0);
 
     if (rebInput.pTopic) {
@@ -673,177 +668,7 @@ void mndReleaseSubscribe(SMnode *pMnode, SMqSubscribeObj *pSub) {
   sdbRelease(pSdb, pSub);
 }
 
-#if 0
-static int32_t mndProcessSubscribeReq(SNodeMsg *pMsg) {
-  SMnode         *pMnode = pMsg->pNode;
-  char           *msgStr = pMsg->rpcMsg.pCont;
-  SCMSubscribeReq subscribe;
-  tDeserializeSCMSubscribeReq(msgStr, &subscribe);
-  int64_t consumerId = subscribe.consumerId;
-  char   *cgroup = subscribe.consumerGroup;
-
-  SArray *newSub = subscribe.topicNames;
-  int32_t newTopicNum = subscribe.topicNum;
-
-  taosArraySortString(newSub, taosArrayCompareString);
-
-  SArray *oldSub = NULL;
-  int32_t oldTopicNum = 0;
-  bool    createConsumer = false;
-  // create consumer if not exist
-  SMqConsumerObj *pConsumer = mndAcquireConsumer(pMnode, consumerId);
-  if (pConsumer == NULL) {
-    // create consumer
-    pConsumer = mndCreateConsumer(consumerId, cgroup);
-    createConsumer = true;
-  } else {
-    pConsumer->epoch++;
-    oldSub = pConsumer->currentTopics;
-  }
-  pConsumer->currentTopics = newSub;
-
-  if (oldSub != NULL) {
-    oldTopicNum = taosArrayGetSize(oldSub);
-  }
-
-  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_SUBSCRIBE, &pMsg->rpcMsg);
-  if (pTrans == NULL) {
-    // TODO: free memory
-    return -1;
-  }
-
-  int32_t i = 0, j = 0;
-  while (i < newTopicNum || j < oldTopicNum) {
-    char *newTopicName = NULL;
-    char *oldTopicName = NULL;
-    if (i >= newTopicNum) {
-      // encode unset topic msg to all vnodes related to that topic
-      oldTopicName = taosArrayGetP(oldSub, j);
-      j++;
-    } else if (j >= oldTopicNum) {
-      newTopicName = taosArrayGetP(newSub, i);
-      i++;
-    } else {
-      newTopicName = taosArrayGetP(newSub, i);
-      oldTopicName = taosArrayGetP(oldSub, j);
-
-      int32_t comp = compareLenPrefixedStr(newTopicName, oldTopicName);
-      if (comp == 0) {
-        // do nothing
-        oldTopicName = newTopicName = NULL;
-        i++;
-        j++;
-        continue;
-      } else if (comp < 0) {
-        oldTopicName = NULL;
-        i++;
-      } else {
-        newTopicName = NULL;
-        j++;
-      }
-    }
-
-    if (oldTopicName != NULL) {
-      ASSERT(newTopicName == NULL);
-
-      // cancel subscribe of old topic
-      SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, cgroup, oldTopicName);
-      ASSERT(pSub);
-      int32_t csz = taosArrayGetSize(pSub->consumers);
-      for (int32_t ci = 0; ci < csz; ci++) {
-        SMqSubConsumer *pSubConsumer = taosArrayGet(pSub->consumers, ci);
-        if (pSubConsumer->consumerId == consumerId) {
-          int32_t vgsz = taosArrayGetSize(pSubConsumer->vgInfo);
-          for (int32_t vgi = 0; vgi < vgsz; vgi++) {
-            SMqConsumerEp *pConsumerEp = taosArrayGet(pSubConsumer->vgInfo, vgi);
-            mndPersistCancelConnReq(pMnode, pTrans, pConsumerEp, oldTopicName);
-            taosArrayPush(pSub->unassignedVg, pConsumerEp);
-          }
-          taosArrayRemove(pSub->consumers, ci);
-          break;
-        }
-      }
-      char *oldTopicNameDup = strdup(oldTopicName);
-      taosArrayPush(pConsumer->recentRemovedTopics, &oldTopicNameDup);
-      atomic_store_32(&pConsumer->status, MQ_CONSUMER_STATUS__MODIFY);
-      /*pSub->status = MQ_SUBSCRIBE_STATUS__DELETED;*/
-    } else if (newTopicName != NULL) {
-      ASSERT(oldTopicName == NULL);
-
-      SMqTopicObj *pTopic = mndAcquireTopic(pMnode, newTopicName);
-      if (pTopic == NULL) {
-        mError("topic being subscribed not exist: %s", newTopicName);
-        continue;
-      }
-
-      SMqSubscribeObj *pSub = mndAcquireSubscribe(pMnode, cgroup, newTopicName);
-      bool             createSub = false;
-      if (pSub == NULL) {
-        mDebug("create new subscription by consumer %" PRId64 ", group: %s, topic %s", consumerId, cgroup,
-               newTopicName);
-        pSub = mndCreateSubscription(pMnode, pTopic, cgroup);
-        createSub = true;
-
-        mndCreateOffset(pTrans, cgroup, newTopicName, pSub->unassignedVg);
-      }
-
-      SMqSubConsumer mqSubConsumer;
-      mqSubConsumer.consumerId = consumerId;
-      mqSubConsumer.vgInfo = taosArrayInit(0, sizeof(SMqConsumerEp));
-      taosArrayPush(pSub->consumers, &mqSubConsumer);
-
-      // if have un assigned vg, assign one to the consumer
-      if (taosArrayGetSize(pSub->unassignedVg) > 0) {
-        SMqConsumerEp *pConsumerEp = taosArrayPop(pSub->unassignedVg);
-        pConsumerEp->oldConsumerId = pConsumerEp->consumerId;
-        pConsumerEp->consumerId = consumerId;
-        taosArrayPush(mqSubConsumer.vgInfo, pConsumerEp);
-        if (pConsumerEp->oldConsumerId == -1) {
-          mInfo("mq set conn: assign vgroup %d of topic %s to consumer %" PRId64 "", pConsumerEp->vgId, newTopicName,
-                pConsumerEp->consumerId);
-          mndPersistMqSetConnReq(pMnode, pTrans, pTopic, cgroup, pConsumerEp);
-        } else {
-          mndPersistRebalanceMsg(pMnode, pTrans, pConsumerEp, newTopicName);
-        }
-        // to trigger rebalance at once, do not set status active
-        /*atomic_store_32(&pConsumer->status, MQ_CONSUMER_STATUS__ACTIVE);*/
-      }
-
-      SSdbRaw *pRaw = mndSubActionEncode(pSub);
-      sdbSetRawStatus(pRaw, SDB_STATUS_READY);
-      mndTransAppendRedolog(pTrans, pRaw);
-
-      if (!createSub) mndReleaseSubscribe(pMnode, pSub);
-      mndReleaseTopic(pMnode, pTopic);
-    }
-  }
-
-  /*if (oldSub) taosArrayDestroyEx(oldSub, (void (*)(void *))taosMemoryFree);*/
-
-  // persist consumerObj
-  SSdbRaw *pConsumerRaw = mndConsumerActionEncode(pConsumer);
-  sdbSetRawStatus(pConsumerRaw, SDB_STATUS_READY);
-  mndTransAppendRedolog(pTrans, pConsumerRaw);
-
-  if (mndTransPrepare(pMnode, pTrans) != 0) {
-    mError("mq-subscribe-trans:%d, failed to prepare since %s", pTrans->id, terrstr());
-    mndTransDrop(pTrans);
-    if (!createConsumer) mndReleaseConsumer(pMnode, pConsumer);
-    return -1;
-  }
-
-  mndTransDrop(pTrans);
-  if (!createConsumer) mndReleaseConsumer(pMnode, pConsumer);
-  return TSDB_CODE_MND_ACTION_IN_PROGRESS;
-}
-#endif
-
 static int32_t mndProcessSubscribeInternalRsp(SNodeMsg *pRsp) {
   mndTransProcessRsp(pRsp);
   return 0;
 }
-
-static void mndCancelGetNextConsumer(SMnode *pMnode, void *pIter) {
-  SSdb *pSdb = pMnode->pSdb;
-  sdbCancelFetch(pSdb, pIter);
-}

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

@@ -82,7 +82,6 @@ SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) {
   SDB_SET_INT8(pRaw, dataPos, pTopic->withTbName, TOPIC_ENCODE_OVER);
   SDB_SET_INT8(pRaw, dataPos, pTopic->withSchema, TOPIC_ENCODE_OVER);
   SDB_SET_INT8(pRaw, dataPos, pTopic->withTag, TOPIC_ENCODE_OVER);
-  SDB_SET_INT8(pRaw, dataPos, pTopic->withTagSchema, TOPIC_ENCODE_OVER);
   SDB_SET_INT32(pRaw, dataPos, pTopic->sqlLen, TOPIC_ENCODE_OVER);
   SDB_SET_BINARY(pRaw, dataPos, pTopic->sql, pTopic->sqlLen, TOPIC_ENCODE_OVER);
   SDB_SET_INT32(pRaw, dataPos, pTopic->astLen, TOPIC_ENCODE_OVER);
@@ -146,7 +145,6 @@ SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) {
   SDB_GET_INT8(pRaw, dataPos, &pTopic->withTbName, TOPIC_DECODE_OVER);
   SDB_GET_INT8(pRaw, dataPos, &pTopic->withSchema, TOPIC_DECODE_OVER);
   SDB_GET_INT8(pRaw, dataPos, &pTopic->withTag, TOPIC_DECODE_OVER);
-  SDB_GET_INT8(pRaw, dataPos, &pTopic->withTagSchema, TOPIC_DECODE_OVER);
 
   SDB_GET_INT32(pRaw, dataPos, &pTopic->sqlLen, TOPIC_DECODE_OVER);
   pTopic->sql = taosMemoryCalloc(pTopic->sqlLen, sizeof(char));
@@ -234,6 +232,7 @@ void mndReleaseTopic(SMnode *pMnode, SMqTopicObj *pTopic) {
   sdbRelease(pSdb, pTopic);
 }
 
+#if 0
 static SDbObj *mndAcquireDbByTopic(SMnode *pMnode, char *topicName) {
   SName name = {0};
   tNameFromString(&name, topicName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
@@ -243,6 +242,7 @@ static SDbObj *mndAcquireDbByTopic(SMnode *pMnode, char *topicName) {
 
   return mndAcquireDb(pMnode, db);
 }
+#endif
 
 static SDDropTopicReq *mndBuildDropTopicMsg(SMnode *pMnode, SVgObj *pVgroup, SMqTopicObj *pTopic) {
   int32_t contLen = sizeof(SDDropTopicReq);
@@ -262,15 +262,11 @@ static SDDropTopicReq *mndBuildDropTopicMsg(SMnode *pMnode, SVgObj *pVgroup, SMq
 }
 
 static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) {
-  if (pCreate->name[0] == 0 || pCreate->sql == NULL || pCreate->sql[0] == 0) {
+  if (pCreate->name[0] == 0 || pCreate->sql == NULL || pCreate->sql[0] == 0 || pCreate->subscribeDbName[0] == 0) {
     terrno = TSDB_CODE_MND_INVALID_TOPIC_OPTION;
     return -1;
   }
 
-  if ((pCreate->ast == NULL || pCreate->ast[0] == 0) && pCreate->subscribeDbName[0] == 0) {
-    terrno = TSDB_CODE_MND_INVALID_TOPIC_OPTION;
-    return -1;
-  }
   return 0;
 }
 
@@ -386,7 +382,7 @@ static int32_t mndProcessCreateTopicReq(SNodeMsg *pReq) {
     goto CREATE_TOPIC_OVER;
   }
 
-  pDb = mndAcquireDbByTopic(pMnode, createTopicReq.name);
+  pDb = mndAcquireDb(pMnode, createTopicReq.subscribeDbName);
   if (pDb == NULL) {
     terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
     goto CREATE_TOPIC_OVER;
@@ -524,8 +520,11 @@ static int32_t mndRetrieveTopic(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pB
     int32_t cols = 0;
 
     char topicName[TSDB_TOPIC_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
-    tstrncpy(&topicName[VARSTR_HEADER_SIZE], pTopic->name, TSDB_TOPIC_NAME_LEN);
-    varDataSetLen(topicName, strlen(&topicName[VARSTR_HEADER_SIZE]));
+
+    SName n;
+    tNameFromString(&n, pTopic->name, T_NAME_ACCT|T_NAME_DB);
+    tNameGetDbName(&n, varDataVal(topicName));
+    varDataSetLen(topicName, strlen(varDataVal(topicName)));
 
     SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, numOfRows, (const char *)topicName, false);
@@ -539,7 +538,7 @@ static int32_t mndRetrieveTopic(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pB
     varDataSetLen(sql, strlen(&sql[VARSTR_HEADER_SIZE]));
     colDataAppend(pColInfo, numOfRows, (const char *)sql, false);
 
-    taosMemoryFree(sql);
+//    taosMemoryFree(sql);
 
     numOfRows++;
     sdbRelease(pSdb, pTopic);

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

@@ -159,7 +159,6 @@ typedef struct {
   int8_t        withTbName;
   int8_t        withSchema;
   int8_t        withTag;
-  int8_t        withTagSchema;
   char*         qmsg;
   STqPushHandle pushHandle;
   // SRWLatch        lock;

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

@@ -31,6 +31,7 @@ STQ* tqOpen(const char* path, SVnode* pVnode, SWal* pWal) {
   pTq->path = strdup(path);
   pTq->pVnode = pVnode;
   pTq->pWal = pWal;
+
 #if 0
   pTq->tqMeta = tqStoreOpen(pTq, path, (FTqSerialize)tqSerializeConsumer, (FTqDeserialize)tqDeserializeConsumer,
                             (FTqDelete)taosMemoryFree, 0);
@@ -401,6 +402,13 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
     consumerEpoch = atomic_val_compare_exchange_32(&pExec->epoch, consumerEpoch, reqEpoch);
   }
 
+  SWalHead* pHeadWithCkSum = taosMemoryMalloc(sizeof(SWalHead) + 2048);
+  if (pHeadWithCkSum == NULL) {
+    return -1;
+  }
+
+  walSetReaderCapacity(pExec->pWalReader, 2048);
+
   SMqDataBlkRsp rsp = {0};
   rsp.reqOffset = pReq->currentOffset;
   rsp.blockData = taosArrayInit(0, sizeof(void*));
@@ -414,6 +422,26 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
       break;
     }
 
+    taosThreadMutexLock(&pExec->pWalReader->mutex);
+
+    if (walFetchHead(pExec->pWalReader, fetchOffset, pHeadWithCkSum) < 0) {
+      vDebug("tmq poll: consumer %ld (epoch %d) vg %d offset %ld, no more log to return", consumerId, pReq->epoch,
+             TD_VID(pTq->pVnode), fetchOffset);
+      taosThreadMutexUnlock(&pExec->pWalReader->mutex);
+      break;
+    }
+
+    if (pHeadWithCkSum->head.msgType != TDMT_VND_SUBMIT) {
+      walSkipFetchBody(pExec->pWalReader, pHeadWithCkSum);
+    } else {
+      walFetchBody(pExec->pWalReader, &pHeadWithCkSum);
+    }
+
+    SWalReadHead* pHead = &pHeadWithCkSum->head;
+
+    taosThreadMutexUnlock(&pExec->pWalReader->mutex);
+
+#if 0
     SWalReadHead* pHead;
     if (walReadWithHandle_s(pExec->pWalReader, fetchOffset, &pHead) < 0) {
       // TODO: no more log, set timer to wait blocking time
@@ -443,14 +471,16 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
       return 0;
 #endif
 
-      break;
-    }
+    break;
+  }
+#endif
 
     vDebug("tmq poll: consumer %ld (epoch %d) iter log, vg %d offset %ld msgType %d", consumerId, pReq->epoch,
            TD_VID(pTq->pVnode), fetchOffset, pHead->msgType);
 
     if (pHead->msgType == TDMT_VND_SUBMIT) {
       SSubmitReq* pCont = (SSubmitReq*)&pHead->body;
+      // table subscribe
       if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
         qTaskInfo_t task = pExec->task[workerId];
         ASSERT(task);
@@ -484,6 +514,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
           taosArrayPush(rsp.blockData, &buf);
           rsp.blockNum++;
         }
+        // db subscribe
       } else if (pExec->subType == TOPIC_SUB_TYPE__DB) {
         STqReadHandle* pReader = pExec->pExecReader[workerId];
         tqReadHandleSetMsg(pReader, pCont, 0);
@@ -789,7 +820,6 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
     pExec->withTbName = req.withTbName;
     pExec->withSchema = req.withSchema;
     pExec->withTag = req.withTag;
-    pExec->withTagSchema = req.withTagSchema;
 
     pExec->qmsg = req.qmsg;
     req.qmsg = NULL;

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

@@ -3223,8 +3223,13 @@ void tsdbRetrieveDataBlockInfo(tsdbReaderT* pTsdbReadHandle, SDataBlockInfo* pDa
   tsdbDebug("data block generated, uid:%" PRIu64 " numOfRows:%d, tsrange:%" PRId64 " - %" PRId64 " %s", uid, cur->rows,
             cur->win.skey, cur->win.ekey, pHandle->idStr);
 
-  //  pDataBlockInfo->uid    = uid; // block Id may be over write by assigning uid fro this data block. Do NOT assign
-  //  the table uid
+  pDataBlockInfo->uid    = uid;
+
+#if 0
+  // for multi-group data query processing test purpose
+  pDataBlockInfo->groupId = uid;
+#endif
+
   pDataBlockInfo->rows = cur->rows;
   pDataBlockInfo->window = cur->win;
   pDataBlockInfo->numOfCols = (int32_t)(QH_GET_NUM_OF_COLS(pHandle));

source/libs/command/src/explain.c

@@ -304,8 +304,8 @@ int32_t qExplainResAppendRow(SExplainCtx *ctx, char *tbuf, int32_t len, int32_t
 
   memcpy(row.buf, tbuf, len);
   row.level = level;
-  row.len = len;
-  ctx->dataSize += len;
+  row.len   = len;
+  ctx->dataSize += row.len;
 
   if (NULL == taosArrayPush(ctx->rows, &row)) {
     qError("taosArrayPush row to explain res rows failed");
@@ -756,7 +756,7 @@ int32_t qExplainGetRspFromCtx(void *ctx, SRetrieveTableRsp **pRsp) {
   }
   
   int32_t colNum = 1;
-  int32_t rspSize = sizeof(SRetrieveTableRsp) + sizeof(int32_t) * colNum + sizeof(int32_t) * rowNum + pCtx->dataSize;
+  int32_t rspSize = sizeof(SRetrieveTableRsp) + sizeof(int32_t) + sizeof(uint64_t) + sizeof(int32_t) * colNum + sizeof(int32_t) * rowNum + pCtx->dataSize;
   SRetrieveTableRsp *rsp = (SRetrieveTableRsp *)taosMemoryCalloc(1, rspSize);
   if (NULL == rsp) {
     qError("malloc SRetrieveTableRsp failed, size:%d", rspSize);
@@ -766,29 +766,38 @@ int32_t qExplainGetRspFromCtx(void *ctx, SRetrieveTableRsp **pRsp) {
   rsp->completed = 1;
   rsp->numOfRows = htonl(rowNum);
 
-  *(int32_t *)rsp->data = htonl(pCtx->dataSize);
+  // payload length
+  *(int32_t *)rsp->data = sizeof(int32_t) + sizeof(uint64_t) + sizeof(int32_t) * colNum + sizeof(int32_t) * rowNum + pCtx->dataSize;
 
-  int32_t *offset = (int32_t *)((char *)rsp->data + sizeof(int32_t));
+  // group id
+  *(uint64_t*)(rsp->data + sizeof(int32_t)) = 0;
+
+  // column length
+  int32_t* colLength = (int32_t *)(rsp->data + sizeof(int32_t) + sizeof(uint64_t));
+
+  // varchar column offset segment
+  int32_t *offset = (int32_t *)((char *)colLength + sizeof(int32_t));
+
+  // varchar data real payload
   char *data = (char *)(offset + rowNum);
-  int32_t tOffset = 0;
-  
+
+  char* start = data;
   for (int32_t i = 0; i < rowNum; ++i) {
     SQueryExplainRowInfo *row = taosArrayGet(pCtx->rows, i);
-    *offset = tOffset;
-    tOffset += row->len;
+    offset[i] = data - start;
 
-    memcpy(data, row->buf, row->len);
-    
-    ++offset;
+    varDataCopy(data, row->buf);
+    ASSERT(varDataTLen(row->buf) == row->len);
     data += row->len;
   }
 
-  *pRsp = rsp;
+  *colLength = htonl(data - start);
+  rsp->compLen = htonl(rspSize);
 
+  *pRsp = rsp;
   return TSDB_CODE_SUCCESS;
 }
 
-
 int32_t qExplainPrepareCtx(SQueryPlan *pDag, SExplainCtx **pCtx) {
   int32_t code = 0;
   SNodeListNode *plans = NULL;
@@ -895,9 +904,7 @@ int32_t qExplainAppendPlanRows(SExplainCtx *pCtx) {
 
 int32_t qExplainGenerateRsp(SExplainCtx *pCtx, SRetrieveTableRsp **pRsp) {
   QRY_ERR_RET(qExplainAppendGroupResRows(pCtx, pCtx->rootGroupId, 0));
-
   QRY_ERR_RET(qExplainAppendPlanRows(pCtx));
-  
   QRY_ERR_RET(qExplainGetRspFromCtx(pCtx, pRsp));
 
   return TSDB_CODE_SUCCESS;
@@ -967,13 +974,10 @@ int32_t qExecStaticExplain(SQueryPlan *pDag, SRetrieveTableRsp **pRsp) {
   SExplainCtx *pCtx = NULL;
 
   QRY_ERR_RET(qExplainPrepareCtx(pDag, &pCtx));
-  
   QRY_ERR_JRET(qExplainGenerateRsp(pCtx, pRsp));
   
 _return:
-
   qExplainFreeCtx(pCtx);
-
   QRY_RET(code);
 }
 

source/libs/executor/inc/executil.h

@@ -40,8 +40,6 @@
 
 #define GET_TASKID(_t)  (((SExecTaskInfo*)(_t))->id.str)
 
-#define curTimeWindowIndex(_winres)        ((_winres)->curIndex)
-
 typedef struct SGroupResInfo {
   int32_t totalGroup;
   int32_t currentGroup;
@@ -68,11 +66,16 @@ typedef struct SResultRowPosition {
   int32_t offset;
 } SResultRowPosition;
 
+typedef struct SResKeyPos {
+  SResultRowPosition pos;
+  uint64_t  groupId;
+  char      key[];
+} SResKeyPos;
+
 typedef struct SResultRowInfo {
   SResultRowPosition *pPosition;
   int32_t      size;       // number of result set
   int32_t      capacity;   // max capacity
-//  int32_t      curPos;     // current active result row index of pResult list
   SResultRowPosition cur;
 } SResultRowInfo;
 
@@ -135,7 +138,7 @@ typedef struct {
   int32_t colId;
 } SStddevInterResult;
 
-void    initGroupResInfo(SGroupResInfo* pGroupResInfo, SResultRowInfo* pResultInfo);
+void    initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, bool sortGroupResult);
 void    initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList);
 
 void    cleanupGroupResInfo(SGroupResInfo* pGroupResInfo);

source/libs/executor/inc/executorimpl.h

@@ -347,10 +347,11 @@ typedef struct STableScanInfo {
 } STableScanInfo;
 
 typedef struct STagScanInfo {
-  SColumnInfo* pCols;
-  SSDataBlock* pRes;
+  SColumnInfo *pCols;
+  SSDataBlock *pRes;
   int32_t      totalTables;
   int32_t      curPos;
+  void        *pReader;
 } STagScanInfo;
 
 typedef struct SStreamBlockScanInfo {
@@ -376,13 +377,11 @@ typedef struct SSysTableScanInfo {
   SEpSet                 epSet;
   tsem_t                 ready;
 
-  int32_t accountId;
-  bool    showRewrite;
-  SNode*  pCondition;  // db_name filter condition, to discard data that are not in current database
-  void*   pCur;        // cursor for iterate the local table meta store.
-  SArray* scanCols;    // SArray<int16_t> scan column id list
-
-//  int32_t             type;  // show type, TODO remove it
+  int32_t             accountId;
+  bool                showRewrite;
+  SNode*              pCondition;  // db_name filter condition, to discard data that are not in current database
+  void*               pCur;        // cursor for iterate the local table meta store.
+  SArray*             scanCols;    // SArray<int16_t> scan column id list
   SName               name;
   SSDataBlock*        pRes;
   int32_t             capacity;
@@ -628,7 +627,7 @@ SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput,
 
 SOperatorInfo* createExchangeOperatorInfo(const SNodeList* pSources, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
 
-SOperatorInfo* createTableScanOperatorInfo(void* pTsdbReadHandle, int32_t order, int32_t numOfCols, int32_t dataLoadFlag, int32_t repeatTime,
+SOperatorInfo* createTableScanOperatorInfo(void* pReaderHandle, int32_t order, int32_t numOfCols, int32_t dataLoadFlag, int32_t repeatTime,
                                            int32_t reverseTime, SArray* pColMatchInfo, SSDataBlock* pResBlock, SNode* pCondition,
                                            SInterval* pInterval, double ratio, SExecTaskInfo* pTaskInfo);
 SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock, SExprInfo* pScalarExprInfo,
@@ -668,12 +667,12 @@ SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SExprInfo*
                                            SSDataBlock* pResultBlock, SExecTaskInfo* pTaskInfo);
 
 SOperatorInfo* createJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SNode* pOnCondition, SExecTaskInfo* pTaskInfo);
+SOperatorInfo* createTagScanOperatorInfo(void* pReaderHandle, SExprInfo* pExpr, int32_t numOfOutput, SExecTaskInfo* pTaskInfo);
 
 #if 0
 SOperatorInfo* createTableSeqScanOperatorInfo(void* pTsdbReadHandle, STaskRuntimeEnv* pRuntimeEnv);
 SOperatorInfo* createMultiTableTimeIntervalOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorInfo* downstream,
                                                         SExprInfo* pExpr, int32_t numOfOutput);
-SOperatorInfo* createTagScanOperatorInfo(SReaderHandle* pReaderHandle, SExprInfo* pExpr, int32_t numOfOutput);
 #endif
 
 void projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,

source/libs/executor/src/dataDispatcher.c

@@ -64,10 +64,10 @@ static bool needCompress(const SSDataBlock* pData, int32_t numOfCols) {
 }
 
 // data format:
-// +----------------+--------------------------------------+-------------+-----------+-------------+-----------+
-// |SDataCacheEntry | column#1 length, column#2 length ... | col1 bitmap | col1 data | col2 bitmap | col2 data | ....
-// |                |    sizeof(int32_t) * numOfCols       | actual size |           | actual size |           |
-// +----------------+--------------------------------------+-------------+-----------+-------------+-----------+
+// +----------------+--------------+----------+--------------------------------------+-------------+-----------+-------------+-----------+
+// |SDataCacheEntry | total length | group id | column#1 length, column#2 length ... | col1 bitmap | col1 data | col2 bitmap | col2 data | ....
+// |                |  (4 bytes)   |(8 bytes) | sizeof(int32_t) * numOfCols          | actual size |           | actual size |           |
+// +----------------+--------------+----------+--------------------------------------+-------------+-----------+-------------+-----------+
 // The length of bitmap is decided by number of rows of this data block, and the length of each column data is
 // recorded in the first segment, next to the struct header
 static void toDataCacheEntry(const SDataDispatchHandle* pHandle, const SInputData* pInput, SDataDispatchBuf* pBuf) {

source/libs/executor/src/executil.c

@@ -186,12 +186,50 @@ void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) {
   pGroupResInfo->index     = 0;
 }
 
-void initGroupResInfo(SGroupResInfo* pGroupResInfo, SResultRowInfo* pResultInfo) {
+static int32_t resultrowCompar1(const void* p1, const void* p2) {
+  SResKeyPos* pp1 = *(SResKeyPos**) p1;
+  SResKeyPos* pp2 = *(SResKeyPos**) p2;
+
+  if (pp1->groupId == pp2->groupId) {
+    int64_t pts1 = *(int64_t*) pp1->key;
+    int64_t pts2 = *(int64_t*) pp2->key;
+
+    if (pts1 == pts2) {
+      return 0;
+    } else {
+      return pts1 < pts2? -1:1;
+    }
+  } else {
+    return pp1->groupId < pp2->groupId? -1:1;
+  }
+}
+
+void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, bool sortGroupResult) {
   if (pGroupResInfo->pRows != NULL) {
     taosArrayDestroy(pGroupResInfo->pRows);
   }
 
-  pGroupResInfo->pRows = taosArrayFromList(pResultInfo->pPosition, pResultInfo->size, sizeof(SResultRowPosition));
+  // extract the result rows information from the hash map
+  void* pData = NULL;
+  pGroupResInfo->pRows = taosArrayInit(10, POINTER_BYTES);
+
+  size_t keyLen = 0;
+  while((pData = taosHashIterate(pHashmap, pData)) != NULL) {
+    void* key = taosHashGetKey(pData, &keyLen);
+
+    SResKeyPos* p = taosMemoryMalloc(keyLen + sizeof(SResultRowPosition));
+
+    p->groupId = *(uint64_t*) key;
+    p->pos = *(SResultRowPosition*) pData;
+    memcpy(p->key, key + sizeof(uint64_t), keyLen - sizeof(uint64_t));
+
+    taosArrayPush(pGroupResInfo->pRows, &p);
+  }
+
+  if (sortGroupResult) {
+    qsort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), POINTER_BYTES, resultrowCompar1);
+  }
+
   pGroupResInfo->index = 0;
   assert(pGroupResInfo->index <= getNumOfTotalRes(pGroupResInfo));
 }

source/libs/executor/src/groupoperator.c

@@ -308,7 +308,7 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator, bool* newgrou
   //  }
 
   blockDataEnsureCapacity(pRes, pOperator->resultInfo.capacity);
-  initGroupResInfo(&pInfo->groupResInfo, &pInfo->binfo.resultRowInfo);
+  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, false);
 
   while(1) {
     doBuildResultDatablock(pRes, &pInfo->groupResInfo, pOperator->pExpr, pInfo->aggSup.pResultBuf, pInfo->binfo.rowCellInfoOffset, pInfo->binfo.pCtx);

source/libs/executor/src/scanoperator.c

@@ -271,9 +271,7 @@ static void setupEnvForReverseScan(STableScanInfo* pTableScanInfo, SqlFunctionCt
 static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator, bool* newgroup) {
   STableScanInfo* pTableScanInfo = pOperator->info;
 
-  SSDataBlock*     pBlock = pTableScanInfo->pResBlock;
-  STableGroupInfo* pTableGroupInfo = &pOperator->pTaskInfo->tableqinfoGroupInfo;
-
+  SSDataBlock* pBlock = pTableScanInfo->pResBlock;
   *newgroup = false;
 
   while (tsdbNextDataBlock(pTableScanInfo->dataReader)) {
@@ -284,18 +282,6 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator, bool* newgroup) {
     pTableScanInfo->numOfBlocks += 1;
     tsdbRetrieveDataBlockInfo(pTableScanInfo->dataReader, &pBlock->info);
 
-    // todo opt
-    //    if (pTableGroupInfo->numOfTables > 1 || (pRuntimeEnv->current == NULL && pTableGroupInfo->numOfTables == 1)) {
-    //      STableQueryInfo** pTableQueryInfo =
-    //          (STableQueryInfo**)taosHashGet(pTableGroupInfo->map, &pBlock->info.uid, sizeof(pBlock->info.uid));
-    //      if (pTableQueryInfo == NULL) {
-    //        break;
-    //      }
-    //
-    //      doTableQueryInfoTimeWindowCheck(pTaskInfo, *pTableQueryInfo, pTableScanInfo->order);
-    //    }
-
-    // this function never returns error?
     uint32_t status = 0;
     int32_t  code = loadDataBlock(pOperator, pTableScanInfo, pBlock, &status);
     //    int32_t  code = loadDataBlockOnDemand(pOperator->pRuntimeEnv, pTableScanInfo, pBlock, &status);
@@ -308,6 +294,8 @@ static SSDataBlock* doTableScanImpl(SOperatorInfo* pOperator, bool* newgroup) {
       continue;
     }
 
+    // reset the block to be 0 by default, this blockId is assigned by physical plan and is used by direct upstream operator.
+    pBlock->info.blockId = 0;
     return pBlock;
   }
 
@@ -405,11 +393,11 @@ SOperatorInfo* createTableScanOperatorInfo(void* pDataReader, int32_t order, int
   pOperator->name         = "TableScanOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN;
   pOperator->blockingOptr = false;
-  pOperator->status = OP_NOT_OPENED;
-  pOperator->info = pInfo;
-  pOperator->numOfOutput = numOfOutput;
-  pOperator->getNextFn = doTableScan;
-  pOperator->pTaskInfo = pTaskInfo;
+  pOperator->status       = OP_NOT_OPENED;
+  pOperator->info         = pInfo;
+  pOperator->numOfOutput  = numOfOutput;
+  pOperator->getNextFn    = doTableScan;
+  pOperator->pTaskInfo    = pTaskInfo;
 
   static int32_t cost = 0;
   pOperator->cost.openCost = ++cost;
@@ -824,12 +812,12 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator, bool* newgroup) {
     int32_t          tableNameSlotId = 1;
     SColumnInfoData* pTableNameCol = taosArrayGet(pInfo->pRes->pDataBlock, tableNameSlotId);
 
-    char*   name = NULL;
+    char*   tb = NULL;
     int32_t numOfRows = 0;
 
     char n[TSDB_TABLE_NAME_LEN] = {0};
-    while ((name = metaTbCursorNext(pInfo->pCur)) != NULL) {
-      STR_TO_VARSTR(n, name);
+    while ((tb = metaTbCursorNext(pInfo->pCur)) != NULL) {
+      STR_TO_VARSTR(n, tb);
       colDataAppend(pTableNameCol, numOfRows, n, false);
       numOfRows += 1;
       if (numOfRows >= pInfo->capacity) {
@@ -992,3 +980,167 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* pSysTableReadHandle, SSDataB
 
   return pOperator;
 }
+
+static SSDataBlock* doTagScan(SOperatorInfo* pOperator, bool* newgroup) {
+#if 0
+  SOperatorInfo* pOperator = (SOperatorInfo*) param;
+  if (pOperator->status == OP_EXEC_DONE) {
+    return NULL;
+  }
+
+  int32_t maxNumOfTables = (int32_t)pResultInfo->capacity;
+
+  STagScanInfo *pInfo = pOperator->info;
+  SSDataBlock  *pRes = pInfo->pRes;
+  *newgroup = false;
+
+  int32_t count = 0;
+  SArray* pa = GET_TABLEGROUP(pRuntimeEnv, 0);
+
+  int32_t functionId = getExprFunctionId(&pOperator->pExpr[0]);
+  if (functionId == FUNCTION_TID_TAG) { // return the tags & table Id
+    assert(pQueryAttr->numOfOutput == 1);
+
+    SExprInfo* pExprInfo = &pOperator->pExpr[0];
+    int32_t rsize = pExprInfo->base.resSchema.bytes;
+
+    count = 0;
+
+    int16_t bytes = pExprInfo->base.resSchema.bytes;
+    int16_t type  = pExprInfo->base.resSchema.type;
+
+    for(int32_t i = 0; i < pQueryAttr->numOfTags; ++i) {
+      if (pQueryAttr->tagColList[i].colId == pExprInfo->base.pColumns->info.colId) {
+        bytes = pQueryAttr->tagColList[i].bytes;
+        type = pQueryAttr->tagColList[i].type;
+        break;
+      }
+    }
+
+    SColumnInfoData* pColInfo = taosArrayGet(pRes->pDataBlock, 0);
+
+    while(pInfo->curPos < pInfo->totalTables && count < maxNumOfTables) {
+      int32_t i = pInfo->curPos++;
+      STableQueryInfo *item = taosArrayGetP(pa, i);
+
+      char *output = pColInfo->pData + count * rsize;
+      varDataSetLen(output, rsize - VARSTR_HEADER_SIZE);
+
+      output = varDataVal(output);
+      STableId* id = TSDB_TABLEID(item->pTable);
+
+      *(int16_t *)output = 0;
+      output += sizeof(int16_t);
+
+      *(int64_t *)output = id->uid;  // memory align problem, todo serialize
+      output += sizeof(id->uid);
+
+      *(int32_t *)output = id->tid;
+      output += sizeof(id->tid);
+
+      *(int32_t *)output = pQueryAttr->vgId;
+      output += sizeof(pQueryAttr->vgId);
+
+      char* data = NULL;
+      if (pExprInfo->base.pColumns->info.colId == TSDB_TBNAME_COLUMN_INDEX) {
+        data = tsdbGetTableName(item->pTable);
+      } else {
+        data = tsdbGetTableTagVal(item->pTable, pExprInfo->base.pColumns->info.colId, type, bytes);
+      }
+
+      doSetTagValueToResultBuf(output, data, type, bytes);
+      count += 1;
+    }
+
+    //qDebug("QInfo:0x%"PRIx64" create (tableId, tag) info completed, rows:%d", GET_TASKID(pRuntimeEnv), count);
+  } else if (functionId == FUNCTION_COUNT) {// handle the "count(tbname)" query
+    SColumnInfoData* pColInfo = taosArrayGet(pRes->pDataBlock, 0);
+    *(int64_t*)pColInfo->pData = pInfo->totalTables;
+    count = 1;
+
+    pOperator->status = OP_EXEC_DONE;
+    //qDebug("QInfo:0x%"PRIx64" create count(tbname) query, res:%d rows:1", GET_TASKID(pRuntimeEnv), count);
+  } else {  // return only the tags|table name etc.
+    SExprInfo* pExprInfo = &pOperator->pExpr[0];  // todo use the column list instead of exprinfo
+
+    count = 0;
+    while(pInfo->curPos < pInfo->totalTables && count < maxNumOfTables) {
+      int32_t i = pInfo->curPos++;
+
+      STableQueryInfo* item = taosArrayGetP(pa, i);
+
+      char *data = NULL, *dst = NULL;
+      int16_t type = 0, bytes = 0;
+      for(int32_t j = 0; j < pOperator->numOfOutput; ++j) {
+        // not assign value in case of user defined constant output column
+        if (TSDB_COL_IS_UD_COL(pExprInfo[j].base.pColumns->flag)) {
+          continue;
+        }
+
+        SColumnInfoData* pColInfo = taosArrayGet(pRes->pDataBlock, j);
+        type  = pExprInfo[j].base.resSchema.type;
+        bytes = pExprInfo[j].base.resSchema.bytes;
+
+        if (pExprInfo[j].base.pColumns->info.colId == TSDB_TBNAME_COLUMN_INDEX) {
+          data = tsdbGetTableName(item->pTable);
+        } else {
+          data = tsdbGetTableTagVal(item->pTable, pExprInfo[j].base.pColumns->info.colId, type, bytes);
+        }
+
+        dst  = pColInfo->pData + count * pExprInfo[j].base.resSchema.bytes;
+        doSetTagValueToResultBuf(dst, data, type, bytes);
+      }
+
+      count += 1;
+    }
+
+    if (pInfo->curPos >= pInfo->totalTables) {
+      pOperator->status = OP_EXEC_DONE;
+    }
+
+    //qDebug("QInfo:0x%"PRIx64" create tag values results completed, rows:%d", GET_TASKID(pRuntimeEnv), count);
+  }
+
+  if (pOperator->status == OP_EXEC_DONE) {
+    setTaskStatus(pOperator->pRuntimeEnv, TASK_COMPLETED);
+  }
+
+  pRes->info.rows = count;
+  return (pRes->info.rows == 0)? NULL:pInfo->pRes;
+
+#endif
+  return TSDB_CODE_SUCCESS;
+}
+
+static void destroyTagScanOperatorInfo(void* param, int32_t numOfOutput) {
+  STagScanInfo* pInfo = (STagScanInfo*)param;
+  pInfo->pRes = blockDataDestroy(pInfo->pRes);
+}
+
+SOperatorInfo* createTagScanOperatorInfo(void* pReaderHandle, SExprInfo* pExpr, int32_t numOfOutput, SExecTaskInfo* pTaskInfo) {
+  STagScanInfo* pInfo = taosMemoryCalloc(1, sizeof(STagScanInfo));
+  SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
+  if (pInfo == NULL || pOperator == NULL) {
+    goto _error;
+  }
+
+  pInfo->pReader           = pReaderHandle;
+  pInfo->curPos            = 0;
+  pOperator->name          = "TagScanOperator";
+  pOperator->operatorType  = QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN;
+  pOperator->blockingOptr  = false;
+  pOperator->status        = OP_NOT_OPENED;
+  pOperator->info          = pInfo;
+  pOperator->getNextFn     = doTagScan;
+  pOperator->pExpr         = pExpr;
+  pOperator->numOfOutput   = numOfOutput;
+  pOperator->pTaskInfo     = pTaskInfo;
+  pOperator->closeFn       = destroyTagScanOperatorInfo;
+
+  return pOperator;
+  _error:
+  taosMemoryFree(pInfo);
+  taosMemoryFree(pOperator);
+  terrno = TSDB_CODE_OUT_OF_MEMORY;
+  return NULL;
+}

source/libs/function/inc/builtinsimpl.h

@@ -24,7 +24,7 @@ extern "C" {
 #include "functionMgt.h"
 
 bool functionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
-int32_t functionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId);
+int32_t functionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
 
 EFuncDataRequired countDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow);
 bool getCountFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
@@ -43,17 +43,17 @@ int32_t maxFunction(SqlFunctionCtx *pCtx);
 bool getAvgFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool avgFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
 int32_t avgFunction(SqlFunctionCtx* pCtx);
-int32_t avgFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId);
+int32_t avgFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
 
 bool getStddevFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool stddevFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
 int32_t stddevFunction(SqlFunctionCtx* pCtx);
-int32_t stddevFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId);
+int32_t stddevFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
 
 bool getPercentileFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool percentileFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
 int32_t percentileFunction(SqlFunctionCtx *pCtx);
-int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId);
+int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
 
 bool getDiffFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
 bool diffFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResInfo);
@@ -65,7 +65,7 @@ int32_t lastFunction(SqlFunctionCtx *pCtx);
 
 bool getTopBotFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv);
 int32_t topFunction(SqlFunctionCtx *pCtx);
-int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId);
+int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
 
 #ifdef __cplusplus
 }

source/libs/function/inc/tudf.h

@@ -42,19 +42,7 @@ enum {
   UDFC_CODE_INVALID_STATE = -5
 };
 
-typedef void *UdfcFuncHandle;
 
-/**
- * create udfd proxy, called once in process that call setupUdf/callUdfxxx/teardownUdf
- * @return error code
- */
-int32_t udfcOpen();
-
-/**
- * destroy udfd proxy
- * @return error code
- */
-int32_t udfcClose();
 
 
 /**

source/libs/function/src/builtinsimpl.c

@@ -14,7 +14,7 @@
  */
 
 #include "builtinsimpl.h"
-#include <libs/nodes/querynodes.h>
+#include "function.h"
 #include "querynodes.h"
 #include "taggfunction.h"
 #include "tdatablock.h"
@@ -44,8 +44,6 @@ typedef struct STopBotResItem {
 } STopBotResItem;
 
 typedef struct STopBotRes {
-  int32_t         pageId;
-//  int32_t   num;
   STopBotResItem *pItems;
 } STopBotRes;
 
@@ -92,18 +90,6 @@ typedef struct SDiffInfo {
     }                                                              \
   } while (0);
 
-#define DO_UPDATE_SUBSID_RES(ctx, ts)                                 \
-  do {                                                                \
-    for (int32_t _i = 0; _i < (ctx)->subsidiaryRes.numOfCols; ++_i) { \
-      SqlFunctionCtx *__ctx = (ctx)->subsidiaryRes.pCtx[_i];          \
-      if (__ctx->functionId == FUNCTION_TS_DUMMY) {                   \
-        __ctx->tag.i = (ts);                                          \
-        __ctx->tag.nType = TSDB_DATA_TYPE_BIGINT;                     \
-      }                                                               \
-      __ctx->fpSet.process(__ctx);                                    \
-    }                                                                 \
-  } while (0)
-
 #define UPDATE_DATA(ctx, left, right, num, sign, _ts) \
   do {                                                \
     if (((left) < (right)) ^ (sign)) {                \
@@ -139,7 +125,8 @@ bool functionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo) {
   return true;
 }
 
-int32_t functionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId) {
+int32_t functionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
+  int32_t slotId = pCtx->pExpr->base.resSchema.slotId;
   SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
 
   SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
@@ -406,7 +393,7 @@ int32_t avgFunction(SqlFunctionCtx* pCtx) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t avgFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId) {
+int32_t avgFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   SInputColumnInfoData* pInput = &pCtx->input;
   int32_t type = pInput->pData[0]->info.type;
   SAvgRes* pAvgRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
@@ -416,7 +403,7 @@ int32_t avgFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId) {
     pAvgRes->result = pAvgRes->sum.dsum / ((double) pAvgRes->count);
   }
 
-  return functionFinalize(pCtx, pBlock, slotId);
+  return functionFinalize(pCtx, pBlock);
 }
 
 EFuncDataRequired statisDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow){
@@ -521,6 +508,49 @@ bool getMinmaxFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
   return true;
 }
 
+#define GET_TS_LIST(x)    ((TSKEY*)((x)->ptsList))
+#define GET_TS_DATA(x, y) (GET_TS_LIST(x)[(y)])
+
+#define DO_UPDATE_TAG_COLUMNS_WITHOUT_TS(ctx)                      \
+  do {                                                             \
+    for (int32_t _i = 0; _i < (ctx)->tagInfo.numOfTagCols; ++_i) { \
+      SqlFunctionCtx *__ctx = (ctx)->tagInfo.pTagCtxList[_i];      \
+      __ctx->fpSet.process(__ctx);                                 \
+    }                                                              \
+  } while (0);
+
+#define DO_UPDATE_SUBSID_RES(ctx, ts)                                \
+  do {                                                               \
+    for (int32_t _i = 0; _i < (ctx)->subsidiaries.num; ++_i) { \
+      SqlFunctionCtx* __ctx = (ctx)->subsidiaries.pCtx[_i];          \
+      if (__ctx->functionId == FUNCTION_TS_DUMMY) {                  \
+        __ctx->tag.i = (ts);                                         \
+        __ctx->tag.nType = TSDB_DATA_TYPE_BIGINT;                    \
+      }                                                              \
+      __ctx->fpSet.process(__ctx);                                   \
+    }                                                                \
+  } while (0)
+
+#define UPDATE_DATA(ctx, left, right, num, sign, _ts) \
+  do {                                                \
+    if (((left) < (right)) ^ (sign)) {                \
+      (left) = (right);                               \
+      DO_UPDATE_SUBSID_RES(ctx, _ts);                 \
+      (num) += 1;                                     \
+    }                                                 \
+  } while (0)
+
+#define LOOPCHECK_N(val, _col, ctx, _t, _nrow, _start, sign, num)        \
+  do {                                                                   \
+    _t *d = (_t *)((_col)->pData);                                       \
+    for (int32_t i = (_start); i < (_nrow) + (_start); ++i) {            \
+      if (((_col)->hasNull) && colDataIsNull_f((_col)->nullbitmap, i)) { \
+        continue;                                                        \
+      }                                                                  \
+      TSKEY ts = (ctx)->ptsList != NULL ? GET_TS_DATA(ctx, i) : 0;       \
+      UPDATE_DATA(ctx, val, d[i], num, sign, ts);                        \
+    }                                                                    \
+  } while (0)
 
 int32_t doMinMaxHelper(SqlFunctionCtx *pCtx, int32_t isMinFunc) {
   int32_t numOfElems = 0;
@@ -564,8 +594,8 @@ int32_t doMinMaxHelper(SqlFunctionCtx *pCtx, int32_t isMinFunc) {
       int64_t val = GET_INT64_VAL(tval);
       if ((prev < val) ^ isMinFunc) {
         *(int64_t*) buf = val;
-        for (int32_t i = 0; i < (pCtx)->subsidiaryRes.numOfCols; ++i) {
-          SqlFunctionCtx* __ctx = pCtx->subsidiaryRes.pCtx[i];
+        for (int32_t i = 0; i < (pCtx)->subsidiaries.num; ++i) {
+          SqlFunctionCtx* __ctx = pCtx->subsidiaries.pCtx[i];
           if (__ctx->functionId == FUNCTION_TS_DUMMY) {  // TODO refactor
             __ctx->tag.i = key;
             __ctx->tag.nType = TSDB_DATA_TYPE_BIGINT;
@@ -581,8 +611,8 @@ int32_t doMinMaxHelper(SqlFunctionCtx *pCtx, int32_t isMinFunc) {
       uint64_t val = GET_UINT64_VAL(tval);
       if ((prev < val) ^ isMinFunc) {
         *(uint64_t*) buf = val;
-        for (int32_t i = 0; i < (pCtx)->subsidiaryRes.numOfCols; ++i) {
-          SqlFunctionCtx* __ctx = pCtx->subsidiaryRes.pCtx[i];
+        for (int32_t i = 0; i < (pCtx)->subsidiaries.num; ++i) {
+          SqlFunctionCtx* __ctx = pCtx->subsidiaries.pCtx[i];
           if (__ctx->functionId == FUNCTION_TS_DUMMY) {  // TODO refactor
             __ctx->tag.i = key;
             __ctx->tag.nType = TSDB_DATA_TYPE_BIGINT;
@@ -797,7 +827,7 @@ int32_t stddevFunction(SqlFunctionCtx* pCtx) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t stddevFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId) {
+int32_t stddevFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   SInputColumnInfoData* pInput = &pCtx->input;
   int32_t type = pInput->pData[0]->info.type;
   SStddevRes* pStddevRes = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx));
@@ -810,7 +840,7 @@ int32_t stddevFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId
     pStddevRes->result = sqrt(pStddevRes->quadraticDSum/((double)pStddevRes->count) - avg*avg);
   }
 
-  return functionFinalize(pCtx, pBlock, slotId);
+  return functionFinalize(pCtx, pBlock);
 }
 
 bool getPercentileFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
@@ -923,7 +953,7 @@ int32_t percentileFunction(SqlFunctionCtx *pCtx) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId) {
+int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   SVariant* pVal = &pCtx->param[1].param;
   double v = pVal->nType == TSDB_DATA_TYPE_INT ? pVal->i : pVal->d;
 
@@ -936,7 +966,7 @@ int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t sl
   }
 
   tMemBucketDestroy(pMemBucket);
-  return functionFinalize(pCtx, pBlock, slotId);
+  return functionFinalize(pCtx, pBlock);
 }
 
 bool getFirstLastFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv) {
@@ -1353,15 +1383,16 @@ static STopBotRes *getTopBotOutputInfo(SqlFunctionCtx *pCtx) {
   return pRes;
 }
 
-static void doAddIntoResult(STopBotRes* pRes, int32_t maxSize, void* pData, int32_t rowIndex, SSDataBlock* pSrcBlock,
+static void doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSDataBlock* pSrcBlock,
                             uint16_t type, uint64_t uid, SResultRowEntryInfo* pEntryInfo);
 
+static void saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STopBotResItem* pItem);
+static void copyTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STopBotResItem* pItem);
+
 int32_t topFunction(SqlFunctionCtx *pCtx) {
   int32_t numOfElems = 0;
   SResultRowEntryInfo *pResInfo = GET_RES_INFO(pCtx);
 
-  STopBotRes *pRes = getTopBotOutputInfo(pCtx);
-
 //  if ((void *)pRes->res[0] != (void *)((char *)pRes + sizeof(STopBotRes) + POINTER_BYTES * pCtx->param[0].i)) {
 //    buildTopBotStruct(pRes, pCtx);
 //  }
@@ -1381,7 +1412,7 @@ int32_t topFunction(SqlFunctionCtx *pCtx) {
     numOfElems++;
 
     char* data = colDataGetData(pCol, i);
-    doAddIntoResult(pRes, pCtx->param[1].param.i, data, i, NULL, type, pInput->uid, pResInfo);
+    doAddIntoResult(pCtx, data, i, pCtx->pSrcBlock, type, pInput->uid, pResInfo);
   }
 
   return TSDB_CODE_SUCCESS;
@@ -1414,9 +1445,11 @@ static int32_t topBotResComparFn(const void *p1, const void *p2, const void *par
   return (val1->v.d > val2->v.d) ? 1 : -1;
 }
 
+void doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSDataBlock* pSrcBlock, uint16_t type,
+                     uint64_t uid, SResultRowEntryInfo* pEntryInfo) {
+  STopBotRes *pRes = getTopBotOutputInfo(pCtx);
+  int32_t maxSize = pCtx->param[1].param.i;
 
-void doAddIntoResult(STopBotRes *pRes, int32_t maxSize, void *pData, int32_t rowIndex, SSDataBlock* pSrcBlock, uint16_t type,
-    uint64_t uid, SResultRowEntryInfo* pEntryInfo) {
   SVariant val = {0};
   taosVariantCreateFromBinary(&val, pData, tDataTypes[type].bytes, type);
 
@@ -1428,22 +1461,9 @@ void doAddIntoResult(STopBotRes *pRes, int32_t maxSize, void *pData, int32_t row
     STopBotResItem* pItem = &pItems[pEntryInfo->numOfRes];
     pItem->v   = val;
     pItem->uid = uid;
-    pItem->tuplePos.pageId = -1;  // todo set the corresponding tuple data in the disk-based buffer
-
-    if (pRes->pageId == -1) {
-      SFilePage* pPage = getNewBufPage(NULL, 0, &pRes->pageId);
-      pPage->num = sizeof(SFilePage);
-
-      // keep the current row data
-      for(int32_t i = 0; i < pSrcBlock->info.numOfCols; ++i) {
-        SColumnInfoData* pCol = taosArrayGet(pSrcBlock->pDataBlock, i);
-        bool isNull = colDataIsNull_s(pCol, rowIndex);
-
-
-        colDataGetData(pCol, rowIndex);
-      }
 
-    }
+    // save the data of this tuple
+    saveTupleData(pCtx, rowIndex, pSrcBlock, pItem);
 
     // allocate the buffer and keep the data of this row into the new allocated buffer
     pEntryInfo->numOfRes++;
@@ -1452,22 +1472,100 @@ void doAddIntoResult(STopBotRes *pRes, int32_t maxSize, void *pData, int32_t row
     if ((IS_SIGNED_NUMERIC_TYPE(type) && val.i > pItems[0].v.i) ||
         (IS_UNSIGNED_NUMERIC_TYPE(type) && val.u > pItems[0].v.u) ||
         (IS_FLOAT_TYPE(type) && val.d > pItems[0].v.d)) {
+      // replace the old data and the coresponding tuple data
       STopBotResItem* pItem = &pItems[0];
       pItem->v   = val;
       pItem->uid = uid;
-      pItem->tuplePos.pageId = -1;  // todo set the corresponding tuple data in the disk-based buffer
+
+      // save the data of this tuple by over writing the old data
+      copyTupleData(pCtx, rowIndex, pSrcBlock, pItem);
 
       taosheapadjust((void *) pItems, sizeof(STopBotResItem), 0, pEntryInfo->numOfRes - 1, (const void *) &type, topBotResComparFn, NULL, false);
     }
   }
 }
 
-int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId) {
+void saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STopBotResItem* pItem) {
+  SFilePage* pPage = NULL;
+
+  int32_t completeRowSize = pSrcBlock->info.rowSize + pSrcBlock->info.numOfCols * sizeof(bool);
+
+  if (pCtx->curBufPage == -1) {
+    pPage = getNewBufPage(pCtx->pBuf, 0, &pCtx->curBufPage);
+    pPage->num = sizeof(SFilePage);
+  } else {
+    pPage = getBufPage(pCtx->pBuf, pCtx->curBufPage);
+    if (pPage->num + completeRowSize > getBufPageSize(pCtx->pBuf)) {
+      pPage = getNewBufPage(pCtx->pBuf, 0, &pCtx->curBufPage);
+      pPage->num = sizeof(SFilePage);
+    }
+  }
+
+  pItem->tuplePos.pageId = pCtx->curBufPage;
+
+  // keep the current row data, extract method
+  int32_t offset = 0;
+  bool*   nullList = (bool*)((char*)pPage + pPage->num);
+  char*   pStart = (char*)(nullList + sizeof(bool) * pSrcBlock->info.numOfCols);
+  for (int32_t i = 0; i < pSrcBlock->info.numOfCols; ++i) {
+    SColumnInfoData* pCol = taosArrayGet(pSrcBlock->pDataBlock, i);
+    bool             isNull = colDataIsNull_s(pCol, rowIndex);
+    if (isNull) {
+      nullList[i] = true;
+      continue;
+    }
+
+    char* p = colDataGetData(pCol, rowIndex);
+    if (IS_VAR_DATA_TYPE(pCol->info.type)) {
+      memcpy(pStart + offset, p, varDataTLen(p));
+    } else {
+      memcpy(pStart + offset, p, pCol->info.bytes);
+    }
+
+    offset += pCol->info.bytes;
+  }
+
+  pItem->tuplePos.offset = pPage->num;
+  pPage->num += completeRowSize;
+
+  setBufPageDirty(pPage, true);
+  releaseBufPage(pCtx->pBuf, pPage);
+}
+
+void copyTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STopBotResItem* pItem) {
+  SFilePage* pPage = getBufPage(pCtx->pBuf, pItem->tuplePos.pageId);
+
+  bool* nullList = (bool*)((char*)pPage + pItem->tuplePos.offset);
+  char* pStart = (char*)(nullList + pSrcBlock->info.numOfCols * sizeof(bool));
+
+  int32_t offset = 0;
+  for(int32_t i = 0; i < pSrcBlock->info.numOfCols; ++i) {
+    SColumnInfoData* pCol = taosArrayGet(pSrcBlock->pDataBlock, i);
+    if ((nullList[i] = colDataIsNull_s(pCol, rowIndex)) == true) {
+      continue;
+    }
+
+    char* p = colDataGetData(pCol, rowIndex);
+    if (IS_VAR_DATA_TYPE(pCol->info.type)) {
+      memcpy(pStart + offset, p, varDataTLen(p));
+    } else {
+      memcpy(pStart + offset, p, pCol->info.bytes);
+    }
+
+    offset += pCol->info.bytes;
+  }
+
+  setBufPageDirty(pPage, true);
+  releaseBufPage(pCtx->pBuf, pPage);
+}
+
+int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   SResultRowEntryInfo *pEntryInfo = GET_RES_INFO(pCtx);
   STopBotRes* pRes = GET_ROWCELL_INTERBUF(pEntryInfo);
   pEntryInfo->complete = true;
 
   int32_t type = pCtx->input.pData[0]->info.type;
+  int32_t slotId = pCtx->pExpr->base.resSchema.slotId;
   SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, slotId);
 
   // todo assign the tag value and the corresponding row data
@@ -1476,19 +1574,45 @@ int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t slotId
     case TSDB_DATA_TYPE_INT: {
       for (int32_t i = 0; i < pEntryInfo->numOfRes; ++i) {
         STopBotResItem* pItem = &pRes->pItems[i];
-        colDataAppendInt32(pCol, currentRow++, (int32_t*)&pItem->v.i);
+        colDataAppendInt32(pCol, currentRow, (int32_t*)&pItem->v.i);
 
         int32_t pageId = pItem->tuplePos.pageId;
         int32_t offset = pItem->tuplePos.offset;
-        if (pageId != -1) {
-          // todo
+        if (pItem->tuplePos.pageId != -1) {
+          SFilePage* pPage = getBufPage(pCtx->pBuf, pageId);
+
+          bool* nullList = (bool*)((char*)pPage + offset);
+          char* pStart = (char*)(nullList + pCtx->pSrcBlock->info.numOfCols * sizeof(bool));
+
+          // todo set the offset value to optimize the performance.
+          for (int32_t j = 0; j < pCtx->subsidiaries.num; ++j) {
+            SqlFunctionCtx* pc = pCtx->subsidiaries.pCtx[j];
+
+            SFunctParam *pFuncParam = &pc->pExpr->base.pParam[0];
+            int32_t srcSlotId = pFuncParam->pCol->slotId;
+            int32_t dstSlotId = pCtx->pExpr->base.resSchema.slotId;
+
+            int32_t ps = 0;
+            for(int32_t k = 0; k < srcSlotId; ++k) {
+              SColumnInfoData* pSrcCol = taosArrayGet(pCtx->pSrcBlock->pDataBlock, k);
+              ps += pSrcCol->info.bytes;
+            }
+
+            SColumnInfoData* pDstCol = taosArrayGet(pBlock->pDataBlock, dstSlotId);
+            if (nullList[srcSlotId]) {
+              colDataAppendNULL(pDstCol, currentRow);
+            } else {
+              colDataAppend(pDstCol, currentRow, (pStart + ps), false);
+            }
+          }
         }
+
+        currentRow += 1;
       }
+
       break;
     }
   }
 
   return pEntryInfo->numOfRes;
-
-//  return functionFinalize(pCtx, pBlock, slotId);
 }

source/libs/qworker/src/qworkerMsg.c

@@ -46,7 +46,7 @@ int32_t qwBuildAndSendQueryRsp(SQWConnInfo *pConn, int32_t code) {
   SQueryTableRsp rsp = {.code = code};
 
   int32_t contLen = tSerializeSQueryTableRsp(NULL, 0, &rsp);
-  void *  msg = rpcMallocCont(contLen);
+  void   *msg = rpcMallocCont(contLen);
   tSerializeSQueryTableRsp(msg, contLen, &rsp);
 
   SRpcMsg rpcRsp = {
@@ -87,7 +87,7 @@ int32_t qwBuildAndSendExplainRsp(SQWConnInfo *pConn, SExplainExecInfo *execInfo,
   SExplainRsp rsp = {.numOfPlans = num, .subplanInfo = execInfo};
 
   int32_t contLen = tSerializeSExplainRsp(NULL, 0, &rsp);
-  void *  pRsp = rpcMallocCont(contLen);
+  void   *pRsp = rpcMallocCont(contLen);
   tSerializeSExplainRsp(pRsp, contLen, &rsp);
 
   SRpcMsg rpcRsp = {
@@ -107,7 +107,7 @@ int32_t qwBuildAndSendExplainRsp(SQWConnInfo *pConn, SExplainExecInfo *execInfo,
 
 int32_t qwBuildAndSendHbRsp(SQWConnInfo *pConn, SSchedulerHbRsp *pStatus, int32_t code) {
   int32_t contLen = tSerializeSSchedulerHbRsp(NULL, 0, pStatus);
-  void *  pRsp = rpcMallocCont(contLen);
+  void   *pRsp = rpcMallocCont(contLen);
   tSerializeSSchedulerHbRsp(pRsp, contLen, pStatus);
 
   SRpcMsg rpcRsp = {
@@ -223,7 +223,7 @@ int32_t qwBuildAndSendShowRsp(SRpcMsg *pMsg, int32_t code) {
   showRsp.tableMeta.numOfColumns = cols;
 
   int32_t bufLen = tSerializeSShowRsp(NULL, 0, &showRsp);
-  void *  pBuf = rpcMallocCont(bufLen);
+  void   *pBuf = rpcMallocCont(bufLen);
   tSerializeSShowRsp(pBuf, bufLen, &showRsp);
 
   SRpcMsg rpcMsg = {
@@ -403,8 +403,8 @@ int32_t qWorkerProcessCQueryMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
   bool               queryDone = false;
   SQueryContinueReq *msg = (SQueryContinueReq *)pMsg->pCont;
   bool               needStop = false;
-  SQWTaskCtx *       handles = NULL;
-  SQWorkerMgmt *     mgmt = (SQWorkerMgmt *)qWorkerMgmt;
+  SQWTaskCtx        *handles = NULL;
+  SQWorkerMgmt      *mgmt = (SQWorkerMgmt *)qWorkerMgmt;
 
   if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
     QW_ELOG("invalid cquery msg, msg:%p, msgLen:%d", msg, pMsg->contLen);
@@ -538,7 +538,7 @@ int32_t qWorkerProcessCancelMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
     return TSDB_CODE_QRY_INVALID_INPUT;
   }
 
-  SQWorkerMgmt *  mgmt = (SQWorkerMgmt *)qWorkerMgmt;
+  SQWorkerMgmt   *mgmt = (SQWorkerMgmt *)qWorkerMgmt;
   int32_t         code = 0;
   STaskCancelReq *msg = pMsg->pCont;
   if (NULL == msg || pMsg->contLen < sizeof(*msg)) {
@@ -620,7 +620,7 @@ int32_t qWorkerProcessHbMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg) {
 
   int32_t         code = 0;
   SSchedulerHbReq req = {0};
-  SQWorkerMgmt *  mgmt = (SQWorkerMgmt *)qWorkerMgmt;
+  SQWorkerMgmt   *mgmt = (SQWorkerMgmt *)qWorkerMgmt;
 
   if (NULL == pMsg->pCont) {
     QW_ELOG("invalid hb msg, msg:%p, msgLen:%d", pMsg->pCont, pMsg->contLen);

source/libs/scalar/src/sclfunc.c

@@ -1265,8 +1265,11 @@ int32_t todayFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOut
 }
 
 int32_t timezoneFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
+  char output[TD_TIMEZONE_LEN + VARSTR_HEADER_SIZE] = {0};
+  memcpy(varDataVal(output), tsTimezoneStr, TD_TIMEZONE_LEN);
+  varDataSetLen(output, strlen(tsTimezoneStr));
   for (int32_t i = 0; i < pInput->numOfRows; ++i) {
-    colDataAppend(pOutput->columnData, i, tsTimezoneStr, false);
+    colDataAppend(pOutput->columnData, i, output, false);
   }
   pOutput->numOfRows = pInput->numOfRows;
   return TSDB_CODE_SUCCESS;

source/libs/wal/src/walRead.c

@@ -138,6 +138,91 @@ static int32_t walReadSeekVer(SWalReadHandle *pRead, int64_t ver) {
   return 0;
 }
 
+void walSetReaderCapacity(SWalReadHandle *pRead, int32_t capacity) { pRead->capacity = capacity; }
+
+int32_t walFetchHead(SWalReadHandle *pRead, int64_t ver, SWalHead *pHead) {
+  int32_t code;
+  // TODO: valid ver
+
+  if (pRead->curVersion != ver) {
+    code = walReadSeekVer(pRead, ver);
+    if (code < 0) return -1;
+  }
+
+  if (!taosValidFile(pRead->pReadLogTFile)) {
+    return -1;
+  }
+
+  code = taosReadFile(pRead->pReadLogTFile, pHead, sizeof(SWalHead));
+  if (code != sizeof(SWalHead)) {
+    return -1;
+  }
+
+  code = walValidHeadCksum(pHead);
+
+  if (code != 0) {
+    wError("unexpected wal log version: % " PRId64 ", since head checksum not passed", ver);
+    terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
+    return -1;
+  }
+
+  return 0;
+}
+
+int32_t walSkipFetchBody(SWalReadHandle *pRead, const SWalHead *pHead) {
+  int32_t code;
+
+  ASSERT(pRead->curVersion == pHead->head.version);
+
+  code = taosLSeekFile(pRead->pReadLogTFile, pHead->head.bodyLen, SEEK_CUR);
+  if (code < 0) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    pRead->curVersion = -1;
+    return -1;
+  }
+
+  pRead->curVersion++;
+
+  return 0;
+}
+
+int32_t walFetchBody(SWalReadHandle *pRead, SWalHead **ppHead) {
+  SWalReadHead *pReadHead = &((*ppHead)->head);
+  int64_t       ver = pReadHead->version;
+
+  if (pRead->capacity < pReadHead->bodyLen) {
+    void *ptr = taosMemoryRealloc(*ppHead, sizeof(SWalHead) + pReadHead->bodyLen);
+    if (ptr == NULL) {
+      terrno = TSDB_CODE_WAL_OUT_OF_MEMORY;
+      return -1;
+    }
+    *ppHead = ptr;
+    pRead->capacity = pReadHead->bodyLen;
+  }
+
+  if (pReadHead->bodyLen != taosReadFile(pRead->pReadLogTFile, pReadHead->body, pReadHead->bodyLen)) {
+    return -1;
+  }
+
+  if (pReadHead->version != ver) {
+    wError("unexpected wal log version: %" PRId64 ", read request version:%" PRId64 "", pRead->pHead->head.version,
+           ver);
+    pRead->curVersion = -1;
+    terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
+    return -1;
+  }
+
+  if (walValidBodyCksum(*ppHead) != 0) {
+    wError("unexpected wal log version: % " PRId64 ", since body checksum not passed", ver);
+    pRead->curVersion = -1;
+    terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
+    return -1;
+  }
+
+  pRead->curVersion = ver + 1;
+  return 0;
+}
+
 int32_t walReadWithHandle_s(SWalReadHandle *pRead, int64_t ver, SWalReadHead **ppHead) {
   taosThreadMutexLock(&pRead->mutex);
   if (walReadWithHandle(pRead, ver) < 0) {
@@ -172,12 +257,14 @@ int32_t walReadWithHandle(SWalReadHandle *pRead, int64_t ver) {
   if (code != sizeof(SWalHead)) {
     return -1;
   }
+
   code = walValidHeadCksum(pRead->pHead);
   if (code != 0) {
     wError("unexpected wal log version: % " PRId64 ", since head checksum not passed", ver);
     terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
     return -1;
   }
+
   if (pRead->capacity < pRead->pHead->head.bodyLen) {
     void *ptr = taosMemoryRealloc(pRead->pHead, sizeof(SWalHead) + pRead->pHead->head.bodyLen);
     if (ptr == NULL) {

source/libs/wal/src/walWrite.c

@@ -13,8 +13,6 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#define _DEFAULT_SOURCE
-
 #include "os.h"
 #include "taoserror.h"
 #include "tchecksum.h"
@@ -298,14 +296,14 @@ int64_t walWriteWithSyncInfo(SWal *pWal, int64_t index, tmsg_t msgType, SSyncLog
   pWal->writeHead.head.bodyLen = bodyLen;
   pWal->writeHead.head.msgType = msgType;
 
-  // sync info
+  // sync info for sync module
   pWal->writeHead.head.syncMeta = syncMeta;
 
   pWal->writeHead.cksumHead = walCalcHeadCksum(&pWal->writeHead);
   pWal->writeHead.cksumBody = walCalcBodyCksum(body, bodyLen);
 
   if (taosWriteFile(pWal->pWriteLogTFile, &pWal->writeHead, sizeof(SWalHead)) != sizeof(SWalHead)) {
-    // ftruncate
+    // TODO ftruncate
     terrno = TAOS_SYSTEM_ERROR(errno);
     wError("vgId:%d, file:%" PRId64 ".log, failed to write since %s", pWal->cfg.vgId, walGetLastFileFirstVer(pWal),
            strerror(errno));
@@ -313,7 +311,7 @@ int64_t walWriteWithSyncInfo(SWal *pWal, int64_t index, tmsg_t msgType, SSyncLog
   }
 
   if (taosWriteFile(pWal->pWriteLogTFile, (char *)body, bodyLen) != bodyLen) {
-    // ftruncate
+    // TODO ftruncate
     terrno = TAOS_SYSTEM_ERROR(errno);
     wError("vgId:%d, file:%" PRId64 ".log, failed to write since %s", pWal->cfg.vgId, walGetLastFileFirstVer(pWal),
            strerror(errno));

source/util/src/tarray.c

@@ -476,6 +476,7 @@ void* taosDecodeArray(const void* buf, SArray** pArray, FDecode decode, int32_t
   return (void*)buf;
 }
 
+// todo remove it
 // order array<type *>
 void taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void* param) {
   taosArrayGetSize(pArray) > 8 ? taosArrayQuickSort(pArray, fn, param) : taosArrayInsertSort(pArray, fn, param);

source/util/src/terror.c

@@ -258,7 +258,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_COLUMN_ALREADY_EXIST,     "Column already exists
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_COLUMN_NOT_EXIST,         "Column does not exist")
 
 // mnode-infoSchema
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_INFOS_TBL,        "Invalid information schema table name")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_SYS_TABLENAME,    "Invalid system table name")
 
 
 // mnode-func

tests/script/tsim/parser/groupby-basic.sim

@@ -115,19 +115,19 @@ endi
 if $data00 != 10 then
  return -1
 endi
-if $data01 != 0 then
+if $data10 != 10 then
   return -1
 endi
-if $data10 != 10 then
+if $data90 != 10 then
   return -1
 endi
-if $data11 != 1 then
+if $data01 != 7 then
   return -1
 endi
-if $data90 != 10 then
+if $data11 != 6 then
   return -1
 endi
-if $data91 != 9 then
+if $data91 != 3 then
   return -1
 endi
 
@@ -143,16 +143,16 @@ if $row != 10 then
   return -1
 endi
 
-if $data00 != @22-01-01 00:00:00.000@ then
+if $data00 != @22-01-01 00:00:00.007@ then
   return -1
 endi
-if $data01 != 0 then
+if $data01 != 7 then
   return -1
 endi
-if $data90 != @22-01-01 00:00:00.009@ then
+if $data90 != @22-01-01 00:00:00.003@ then
   return -1
 endi
-if $data91 != 9 then
+if $data91 != 3 then
   return -1
 endi
 

tests/script/tsim/stable/disk.sim

@@ -53,6 +53,7 @@ endi
 sql select count(tbcol) from $mt
 print select count(tbcol) from $mt ===> $data00
 if $data00 != $totalNum then
+  print expect $totalNum , actual: $data00
   return -1
 endi