Merge remote-tracking branch 'origin/3.0' into feature/3.0_liaohj

contrib/CMakeLists.txt

@@ -219,8 +219,8 @@ endif(${BUILD_WITH_NURAFT})
 # pthread
 if(${BUILD_PTHREAD})
     set(CMAKE_BUILD_TYPE release)
-    add_definitions(-DPTW32_STATIC_LIB) 
-    add_subdirectory(pthread EXCLUDE_FROM_ALL)
+    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)

include/client/taos.h

@@ -121,13 +121,14 @@ typedef struct setConfRet {
 DLL_EXPORT void       taos_cleanup(void);
 DLL_EXPORT int        taos_options(TSDB_OPTION option, const void *arg, ...);
 DLL_EXPORT setConfRet taos_set_config(const char *config);
+DLL_EXPORT int        taos_init(void);
 DLL_EXPORT TAOS      *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port);
-DLL_EXPORT TAOS *taos_connect_l(const char *ip, int ipLen, const char *user, int userLen, const char *pass, int passLen,
+DLL_EXPORT TAOS      *taos_connect_l(const char *ip, int ipLen, const char *user, int userLen, const char *pass, int passLen,
                                 const char *db, int dbLen, uint16_t port);
-DLL_EXPORT TAOS *taos_connect_auth(const char *ip, const char *user, const char *auth, const char *db, uint16_t port);
-DLL_EXPORT void  taos_close(TAOS *taos);
+DLL_EXPORT TAOS      *taos_connect_auth(const char *ip, const char *user, const char *auth, const char *db, uint16_t port);
+DLL_EXPORT void       taos_close(TAOS *taos);
 
-const char *taos_data_type(int type);
+const char            *taos_data_type(int type);
 
 DLL_EXPORT TAOS_STMT  *taos_stmt_init(TAOS *taos);
 DLL_EXPORT int         taos_stmt_prepare(TAOS_STMT *stmt, const char *sql, unsigned long length);

include/common/tmsg.h

@@ -438,6 +438,7 @@ typedef struct {
 
 int32_t tSerializeSGetUserAuthRsp(void* buf, int32_t bufLen, SGetUserAuthRsp* pRsp);
 int32_t tDeserializeSGetUserAuthRsp(void* buf, int32_t bufLen, SGetUserAuthRsp* pRsp);
+void    tFreeSGetUserAuthRsp(SGetUserAuthRsp* pRsp);
 
 typedef struct {
   int16_t colId;     // column id
@@ -1332,7 +1333,7 @@ int32_t tDeserializeSCMCreateTopicRsp(void* buf, int32_t bufLen, SCMCreateTopicR
 
 typedef struct {
   int64_t consumerId;
-} SMqConsumerLostMsg;
+} SMqConsumerLostMsg, SMqConsumerRecoverMsg;
 
 typedef struct {
   int64_t consumerId;
@@ -1537,6 +1538,10 @@ int tDecodeSVCreateStbReq(SCoder* pCoder, SVCreateStbReq* pReq);
 
 typedef struct SVDropStbReq {
   // data
+#ifdef WINDOWS
+  size_t avoidCompilationErrors;
+#endif
+
 } SVDropStbReq;
 
 typedef struct SVCreateStbRsp {
@@ -2116,7 +2121,7 @@ static FORCE_INLINE int32_t tDecodeSSchemaWrapper(SCoder* pDecoder, SSchemaWrapp
   if (tDecodeI32v(pDecoder, &pSW->nCols) < 0) return -1;
   if (tDecodeI32v(pDecoder, &pSW->sver) < 0) return -1;
 
-  pSW->pSchema = (SSchema*)TCODER_MALLOC(pDecoder, sizeof(SSchema) * pSW->nCols);
+  pSW->pSchema = (SSchema*)tCoderMalloc(pDecoder, sizeof(SSchema) * pSW->nCols);
   if (pSW->pSchema == NULL) return -1;
   for (int32_t i = 0; i < pSW->nCols; i++) {
     if (tDecodeSSchema(pDecoder, &pSW->pSchema[i]) < 0) return -1;

include/common/tmsgdef.h

@@ -145,10 +145,11 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_MND_ALTER_TOPIC, "mnode-alter-topic", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_DROP_TOPIC, "mnode-drop-topic", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_SUBSCRIBE, "mnode-subscribe", SCMSubscribeReq, SCMSubscribeRsp)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_ASK_EP, "mnode-mq-ask-ep", SMqAskEpReq, SMqAskEpReq)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mnode-mq-tmr", SMTimerReq, SMTimerReq)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_LOST, "mnode-mq-consumer-lost", SMTimerReq, SMTimerReq)
-  TD_DEF_MSG_TYPE(TDMT_MND_MQ_DO_REBALANCE, "mnode-mq-do-rebalance", SMqDoRebalanceMsg, SMqDoRebalanceMsg)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_ASK_EP, "mnode-mq-ask-ep", SMqAskEpReq, SMqAskEpRsp)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_TIMER, "mnode-mq-tmr", SMTimerReq, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_LOST, "mnode-mq-consumer-lost", SMqConsumerLostMsg, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_CONSUMER_RECOVER, "mnode-mq-consumer-recover", SMqConsumerRecoverMsg, NULL)
+  TD_DEF_MSG_TYPE(TDMT_MND_MQ_DO_REBALANCE, "mnode-mq-do-rebalance", SMqDoRebalanceMsg, NULL)
   TD_DEF_MSG_TYPE(TDMT_MND_MQ_COMMIT_OFFSET, "mnode-mq-commit-offset", SMqCMCommitOffsetReq, SMqCMCommitOffsetRsp)
   TD_DEF_MSG_TYPE(TDMT_MND_CREATE_STREAM, "mnode-create-stream", SCMCreateStreamReq, SCMCreateStreamRsp)
   TD_DEF_MSG_TYPE(TDMT_MND_ALTER_STREAM, "mnode-alter-stream", NULL, NULL)

include/os/os.h

@@ -59,6 +59,7 @@ extern "C" {
 #include <winsock.h>
 #endif
 
+#define __typeof(a) auto
 
 #endif
 

include/os/osMath.h

@@ -23,27 +23,21 @@ extern "C" {
 #define TPOW2(x) ((x) * (x))
 #define TABS(x) ((x) > 0 ? (x) : -(x))
 
+#define TSWAP(a, b)             \
+  do {                          \
+    __typeof(a) __tmp = (a); \
+    (a) = (b);               \
+    (b) = __tmp;                \
+  } while (0)
+
 #ifdef WINDOWS
 
-  #define TSWAP(a, b, c) \
-    do {                \
-      c __tmp = (c)(a); \
-      (a) = (c)(b);     \
-      (b) = __tmp;      \
-    } while (0)
   #define TMAX(a, b) (((a) > (b)) ? (a) : (b))
   #define TMIN(a, b) (((a) < (b)) ? (a) : (b))
   #define TRANGE(aa, bb, cc) ((aa) = TMAX((aa), (bb)),(aa) = TMIN((aa), (cc)))
 
 #else
 
-  #define TSWAP(a, b, c)       \
-    do {                       \
-      __typeof(a) __tmp = (a); \
-      (a) = (b);               \
-      (b) = __tmp;             \
-    } while (0)
-
   #define TMAX(a, b)             \
     ({                           \
       __typeof(a) __a = (a);     \
@@ -51,12 +45,12 @@ extern "C" {
       (__a > __b) ? __a : __b;   \
     })
 
-  #define TMIN(a, b)             \
-    ({                           \
-      __typeof(a) __a = (a);     \
-      __typeof(b) __b = (b);     \
-      (__a < __b) ? __a : __b;   \
-    })
+#define TMIN(a, b)             \
+  ({                           \
+    __typeof(a) __a = (a);     \
+    __typeof(b) __b = (b);     \
+    (__a < __b) ? __a : __b;   \
+  })
 
 #define TRANGE(a, b, c) \
   ({                    \

include/os/osSocket.h

@@ -19,53 +19,53 @@
 // If the error is in a third-party library, place this header file under the third-party library header file.
 // When you want to use this feature, you should find or add the same function in the following section.
 #ifndef ALLOW_FORBID_FUNC
-  #define socket SOCKET_FUNC_TAOS_FORBID
-  #define bind   BIND_FUNC_TAOS_FORBID
-  #define listen LISTEN_FUNC_TAOS_FORBID
-  #define accept ACCEPT_FUNC_TAOS_FORBID
-  #define epoll_create EPOLL_CREATE_FUNC_TAOS_FORBID
-  #define epoll_ctl EPOLL_CTL_FUNC_TAOS_FORBID
-  #define epoll_wait EPOLL_WAIT_FUNC_TAOS_FORBID
-  #define inet_addr INET_ADDR_FUNC_TAOS_FORBID
-  #define inet_ntoa INET_NTOA_FUNC_TAOS_FORBID
+#define socket       SOCKET_FUNC_TAOS_FORBID
+#define bind         BIND_FUNC_TAOS_FORBID
+#define listen       LISTEN_FUNC_TAOS_FORBID
+#define accept       ACCEPT_FUNC_TAOS_FORBID
+#define epoll_create EPOLL_CREATE_FUNC_TAOS_FORBID
+#define epoll_ctl    EPOLL_CTL_FUNC_TAOS_FORBID
+#define epoll_wait   EPOLL_WAIT_FUNC_TAOS_FORBID
+#define inet_addr    INET_ADDR_FUNC_TAOS_FORBID
+#define inet_ntoa    INET_NTOA_FUNC_TAOS_FORBID
 #endif
 
 #if defined(WINDOWS)
-  #if BYTE_ORDER == LITTLE_ENDIAN
-    #include <stdlib.h>
-    #define htobe16(x) _byteswap_ushort(x)
-    #define htole16(x) (x)
-    #define be16toh(x) _byteswap_ushort(x)
-    #define le16toh(x) (x)
-                      
-    #define htobe32(x) _byteswap_ulong(x)
-    #define htole32(x) (x)
-    #define be32toh(x) _byteswap_ulong(x)
-    #define le32toh(x) (x)
-                      
-    #define htobe64(x) _byteswap_uint64(x)
-    #define htole64(x) (x)
-    #define be64toh(x) _byteswap_uint64(x)
-    #define le64toh(x) (x)
-  #else
-    #error byte order not supported
-  #endif
-
-  #define __BYTE_ORDER    BYTE_ORDER
-  #define __BIG_ENDIAN    BIG_ENDIAN
-  #define __LITTLE_ENDIAN LITTLE_ENDIAN
-  #define __PDP_ENDIAN    PDP_ENDIAN
+#if BYTE_ORDER == LITTLE_ENDIAN
+#include <stdlib.h>
+#define htobe16(x) _byteswap_ushort(x)
+#define htole16(x) (x)
+#define be16toh(x) _byteswap_ushort(x)
+#define le16toh(x) (x)
+
+#define htobe32(x) _byteswap_ulong(x)
+#define htole32(x) (x)
+#define be32toh(x) _byteswap_ulong(x)
+#define le32toh(x) (x)
+
+#define htobe64(x) _byteswap_uint64(x)
+#define htole64(x) (x)
+#define be64toh(x) _byteswap_uint64(x)
+#define le64toh(x) (x)
+#else
+#error byte order not supported
+#endif
+
+#define __BYTE_ORDER    BYTE_ORDER
+#define __BIG_ENDIAN    BIG_ENDIAN
+#define __LITTLE_ENDIAN LITTLE_ENDIAN
+#define __PDP_ENDIAN    PDP_ENDIAN
 
 #else
-  #include <netinet/in.h>
-  #include <sys/socket.h>
-
-  #if defined(_TD_DARWIN_64)
-    #include <osEok.h>
-  #else
-    #include <netinet/in.h>
-    #include <sys/epoll.h>
-  #endif
+#include <netinet/in.h>
+#include <sys/socket.h>
+
+#if defined(_TD_DARWIN_64)
+#include <osEok.h>
+#else
+#include <netinet/in.h>
+#include <sys/epoll.h>
+#endif
 #endif
 
 #ifdef __cplusplus
@@ -73,24 +73,24 @@ extern "C" {
 #endif
 
 #if defined(WINDOWS)
-  typedef int socklen_t;
-  #define TAOS_EPOLL_WAIT_TIME 100
-  typedef SOCKET eventfd_t;
-  #define eventfd(a, b) -1
-  #define EpollClose(pollFd) epoll_close(pollFd)
-  #ifndef EPOLLWAKEUP
-    #define EPOLLWAKEUP (1u << 29)
-  #endif
+typedef int socklen_t;
+#define TAOS_EPOLL_WAIT_TIME 100
+typedef SOCKET eventfd_t;
+#define eventfd(a, b)      -1
+#define EpollClose(pollFd) epoll_close(pollFd)
+#ifndef EPOLLWAKEUP
+#define EPOLLWAKEUP (1u << 29)
+#endif
 #elif defined(_TD_DARWIN_64)
-  #define TAOS_EPOLL_WAIT_TIME 500
-  typedef int32_t SOCKET;
-  typedef SOCKET EpollFd;
-  #define EpollClose(pollFd) epoll_close(pollFd)
+#define TAOS_EPOLL_WAIT_TIME 500
+typedef int32_t SOCKET;
+typedef SOCKET  EpollFd;
+#define EpollClose(pollFd)   epoll_close(pollFd)
 #else
-  #define TAOS_EPOLL_WAIT_TIME 500
-  typedef int32_t SOCKET;
-  typedef SOCKET EpollFd;
-  #define EpollClose(pollFd) taosCloseSocket(pollFd)
+#define TAOS_EPOLL_WAIT_TIME 500
+typedef int32_t SOCKET;
+typedef SOCKET  EpollFd;
+#define EpollClose(pollFd)   taosCloseSocket(pollFd)
 #endif
 
 #if defined(_TD_DARWIN_64)
@@ -119,8 +119,8 @@ extern "C" {
 #define __PDP_ENDIAN    PDP_ENDIAN
 #endif
 
-typedef int32_t SocketFd;
-typedef SocketFd  EpollFd;
+typedef int32_t  SocketFd;
+typedef SocketFd EpollFd;
 
 typedef struct TdSocket {
 #if SOCKET_WITH_LOCK
@@ -128,16 +128,17 @@ typedef struct TdSocket {
 #endif
   int      refId;
   SocketFd fd;
-} *TdSocketPtr, TdSocket;
+} * TdSocketPtr, TdSocket;
 
 typedef struct TdSocketServer *TdSocketServerPtr;
-typedef struct TdSocket *TdSocketPtr;
-typedef struct TdEpoll *TdEpollPtr;
+typedef struct TdSocket *      TdSocketPtr;
+typedef struct TdEpoll *       TdEpollPtr;
 
-int32_t taosSendto(TdSocketPtr pSocket, void * msg, int len, unsigned int flags, const struct sockaddr * to, int tolen);
+int32_t taosSendto(TdSocketPtr pSocket, void *msg, int len, unsigned int flags, const struct sockaddr *to, int tolen);
 int32_t taosWriteSocket(TdSocketPtr pSocket, void *msg, int len);
 int32_t taosReadSocket(TdSocketPtr pSocket, void *msg, int len);
-int32_t taosReadFromSocket(TdSocketPtr pSocket, void *buf, int32_t len, int32_t flags, struct sockaddr *destAddr, int *addrLen);
+int32_t taosReadFromSocket(TdSocketPtr pSocket, void *buf, int32_t len, int32_t flags, struct sockaddr *destAddr,
+                           int *addrLen);
 int32_t taosCloseSocketNoCheck1(SocketFd fd);
 int32_t taosCloseSocket(TdSocketPtr *ppSocket);
 int32_t taosCloseSocketServer(TdSocketServerPtr *ppSocketServer);
@@ -154,30 +155,32 @@ int32_t taosWriteMsg(TdSocketPtr pSocket, void *ptr, int32_t nbytes);
 int32_t taosReadMsg(TdSocketPtr pSocket, void *ptr, int32_t nbytes);
 int32_t taosNonblockwrite(TdSocketPtr pSocket, char *ptr, int32_t nbytes);
 int64_t taosCopyFds(TdSocketPtr pSrcSocket, TdSocketPtr pDestSocket, int64_t len);
-void taosWinSocketInit();
+void    taosWinSocketInit();
+
+int taosCreateSocketWithTimeOutOpt(uint32_t conn_timeout_sec);
 
-TdSocketPtr taosOpenUdpSocket(uint32_t localIp, uint16_t localPort);
-TdSocketPtr taosOpenTcpClientSocket(uint32_t ip, uint16_t port, uint32_t localIp);
+TdSocketPtr       taosOpenUdpSocket(uint32_t localIp, uint16_t localPort);
+TdSocketPtr       taosOpenTcpClientSocket(uint32_t ip, uint16_t port, uint32_t localIp);
 TdSocketServerPtr taosOpenTcpServerSocket(uint32_t ip, uint16_t port);
-int32_t taosKeepTcpAlive(TdSocketPtr pSocket);
-TdSocketPtr taosAcceptTcpConnectSocket(TdSocketServerPtr pServerSocket, struct sockaddr *destAddr, int *addrLen);
-
-int32_t taosGetSocketName(TdSocketPtr pSocket,struct sockaddr *destAddr, int *addrLen);
-
-void     taosBlockSIGPIPE();
-uint32_t taosGetIpv4FromFqdn(const char *);
-int32_t  taosGetFqdn(char *);
-void     tinet_ntoa(char *ipstr, uint32_t ip);
-uint32_t ip2uint(const char *const ip_addr);
-void     taosIgnSIGPIPE();
-void     taosSetMaskSIGPIPE();
-uint32_t taosInetAddr(const char *ipAddr);
+int32_t           taosKeepTcpAlive(TdSocketPtr pSocket);
+TdSocketPtr       taosAcceptTcpConnectSocket(TdSocketServerPtr pServerSocket, struct sockaddr *destAddr, int *addrLen);
+
+int32_t taosGetSocketName(TdSocketPtr pSocket, struct sockaddr *destAddr, int *addrLen);
+
+void        taosBlockSIGPIPE();
+uint32_t    taosGetIpv4FromFqdn(const char *);
+int32_t     taosGetFqdn(char *);
+void        tinet_ntoa(char *ipstr, uint32_t ip);
+uint32_t    ip2uint(const char *const ip_addr);
+void        taosIgnSIGPIPE();
+void        taosSetMaskSIGPIPE();
+uint32_t    taosInetAddr(const char *ipAddr);
 const char *taosInetNtoa(struct in_addr ipInt);
 
 TdEpollPtr taosCreateEpoll(int32_t size);
-int32_t taosCtlEpoll(TdEpollPtr pEpoll, int32_t epollOperate, TdSocketPtr pSocket, struct epoll_event *event);
-int32_t taosWaitEpoll(TdEpollPtr pEpoll, struct epoll_event *event, int32_t maxEvents, int32_t timeout);
-int32_t taosCloseEpoll(TdEpollPtr *ppEpoll);
+int32_t    taosCtlEpoll(TdEpollPtr pEpoll, int32_t epollOperate, TdSocketPtr pSocket, struct epoll_event *event);
+int32_t    taosWaitEpoll(TdEpollPtr pEpoll, struct epoll_event *event, int32_t maxEvents, int32_t timeout);
+int32_t    taosCloseEpoll(TdEpollPtr *ppEpoll);
 
 #ifdef __cplusplus
 }

include/util/taoserror.h

@@ -62,6 +62,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_APP_NOT_READY                 TAOS_DEF_ERROR_CODE(0, 0x0014)
 #define TSDB_CODE_RPC_FQDN_ERROR                TAOS_DEF_ERROR_CODE(0, 0x0015)
 #define TSDB_CODE_RPC_INVALID_VERSION           TAOS_DEF_ERROR_CODE(0, 0x0016)
+#define TSDB_CODE_RPC_PORT_EADDRINUSE           TAOS_DEF_ERROR_CODE(0, 0x0017)
 
 //common & util
 #define TSDB_CODE_OUT_OF_MEMORY                 TAOS_DEF_ERROR_CODE(0, 0x0100)
@@ -268,14 +269,13 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_MND_TOPIC_NOT_EXIST           TAOS_DEF_ERROR_CODE(0, 0x03E1)
 #define TSDB_CODE_MND_TOO_MANY_TOPICS           TAOS_DEF_ERROR_CODE(0, 0x03E2)
 #define TSDB_CODE_MND_INVALID_TOPIC             TAOS_DEF_ERROR_CODE(0, 0x03E3)
-#define TSDB_CODE_MND_INVALID_TOPIC_OPTION      TAOS_DEF_ERROR_CODE(0, 0x03E4)
-#define TSDB_CODE_MND_TOPIC_OPTION_UNCHNAGED    TAOS_DEF_ERROR_CODE(0, 0x03E5)
-#define TSDB_CODE_MND_NAME_CONFLICT_WITH_STB    TAOS_DEF_ERROR_CODE(0, 0x03E6)
-#define TSDB_CODE_MND_CONSUMER_NOT_EXIST        TAOS_DEF_ERROR_CODE(0, 0x03E7)
-#define TSDB_CODE_MND_UNSUPPORTED_TOPIC         TAOS_DEF_ERROR_CODE(0, 0x03E8)
-#define TSDB_CODE_MND_SUBSCRIBE_NOT_EXIST       TAOS_DEF_ERROR_CODE(0, 0x03E9)
-#define TSDB_CODE_MND_OFFSET_NOT_EXIST          TAOS_DEF_ERROR_CODE(0, 0x03EA)
-#define TSDB_CODE_MND_CONSUMER_NOT_READY        TAOS_DEF_ERROR_CODE(0, 0x03EB)
+#define TSDB_CODE_MND_INVALID_TOPIC_QUERY       TAOS_DEF_ERROR_CODE(0, 0x03E4)
+#define TSDB_CODE_MND_INVALID_TOPIC_OPTION      TAOS_DEF_ERROR_CODE(0, 0x03E5)
+#define TSDB_CODE_MND_CONSUMER_NOT_EXIST        TAOS_DEF_ERROR_CODE(0, 0x03E6)
+#define TSDB_CODE_MND_TOPIC_OPTION_UNCHNAGED    TAOS_DEF_ERROR_CODE(0, 0x03E7)
+#define TSDB_CODE_MND_SUBSCRIBE_NOT_EXIST       TAOS_DEF_ERROR_CODE(0, 0x03E8)
+#define TSDB_CODE_MND_OFFSET_NOT_EXIST          TAOS_DEF_ERROR_CODE(0, 0x03E9)
+#define TSDB_CODE_MND_CONSUMER_NOT_READY        TAOS_DEF_ERROR_CODE(0, 0x03EA)
 
 // mnode-stream
 #define TSDB_CODE_MND_STREAM_ALREADY_EXIST      TAOS_DEF_ERROR_CODE(0, 0x03F0)

include/util/tarray.h

@@ -205,7 +205,6 @@ SArray* taosArrayDup(const SArray* pSrc);
  */
 SArray* taosArrayDeepCopy(const SArray* pSrc, FCopy deepCopy);
 
-
 /**
  * clear the array (remove all element)
  * @param pArray
@@ -272,6 +271,8 @@ void taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void* par
 int32_t taosEncodeArray(void** buf, const SArray* pArray, FEncode encode);
 void*   taosDecodeArray(const void* buf, SArray** pArray, FDecode decode, int32_t dataSz);
 
+char* taosShowStrArray(const SArray* pArray);
+
 #ifdef __cplusplus
 }
 #endif

include/util/tdef.h

@@ -131,6 +131,7 @@ extern const int32_t TYPE_BYTES[15];
 #define TSDB_PERFS_TABLE_TOPICS        "topics"
 #define TSDB_PERFS_TABLE_CONSUMERS     "consumers"
 #define TSDB_PERFS_TABLE_SUBSCRIPTIONS "subscriptions"
+#define TSDB_PERFS_TABLE_OFFSETS       "offsets"
 
 #define TSDB_INDEX_TYPE_SMA      "SMA"
 #define TSDB_INDEX_TYPE_FULLTEXT "FULLTEXT"

include/util/tencode.h

@@ -79,31 +79,52 @@ typedef struct {
 #define TD_CODER_CURRENT(CODER)                        ((CODER)->data + (CODER)->pos)
 #define TD_CODER_MOVE_POS(CODER, MOVE)                 ((CODER)->pos += (MOVE))
 #define TD_CODER_CHECK_CAPACITY_FAILED(CODER, EXPSIZE) (((CODER)->size - (CODER)->pos) < (EXPSIZE))
-#define TCODER_MALLOC(PCODER, SIZE)                                         \
-  ({                                                                        \
-    void*      ptr = NULL;                                                  \
-    SCoderMem* pMem = (SCoderMem*)taosMemoryMalloc(sizeof(*pMem) + (SIZE)); \
-    if (pMem) {                                                             \
-      pMem->next = (PCODER)->mList;                                         \
-      (PCODER)->mList = pMem;                                               \
-      ptr = (void*)&pMem[1];                                                \
-    }                                                                       \
-    ptr;                                                                    \
-  })
-
-#define tEncodeSize(E, S, SIZE)                                \
-  ({                                                           \
+// #define TCODER_MALLOC(PCODER, SIZE)                                         \
+//   ({                                                                        \
+//     void*      ptr = NULL;                                                  \
+//     SCoderMem* pMem = (SCoderMem*)taosMemoryMalloc(sizeof(*pMem) + (SIZE)); \
+//     if (pMem) {                                                             \
+//       pMem->next = (PCODER)->mList;                                         \
+//       (PCODER)->mList = pMem;                                               \
+//       ptr = (void*)&pMem[1];                                                \
+//     }                                                                       \
+//     ptr;                                                                    \
+//   })
+static FORCE_INLINE void* tCoderMalloc(SCoder* pCoder, int32_t size) {
+  void*      ptr = NULL;
+  SCoderMem* pMem = (SCoderMem*)taosMemoryMalloc(sizeof(SCoderMem*) + size);
+  if (pMem) {
+    pMem->next = pCoder->mList;
+    pCoder->mList = pMem;
+    ptr = (void*)&pMem[1];
+  }
+  return ptr;
+}
+
+#define tEncodeSize(E, S, SIZE, RET)                           \
+  do{                                                          \
     SCoder coder = {0};                                        \
-    int    ret = 0;                                            \
     tCoderInit(&coder, TD_LITTLE_ENDIAN, NULL, 0, TD_ENCODER); \
     if ((E)(&coder, S) == 0) {                                 \
       SIZE = coder.pos;                                        \
     } else {                                                   \
-      ret = -1;                                                \
+      RET = -1;                                                \
     }                                                          \
     tCoderClear(&coder);                                       \
-    ret;                                                       \
-  })
+  }while(0)
+// #define tEncodeSize(E, S, SIZE)                                \
+//   ({                                                           \
+//     SCoder coder = {0};                                        \
+//     int    ret = 0;                                            \
+//     tCoderInit(&coder, TD_LITTLE_ENDIAN, NULL, 0, TD_ENCODER); \
+//     if ((E)(&coder, S) == 0) {                                 \
+//       SIZE = coder.pos;                                        \
+//     } else {                                                   \
+//       ret = -1;                                                \
+//     }                                                          \
+//     tCoderClear(&coder);                                       \
+//     ret;                                                       \
+//   })
 
 void tCoderInit(SCoder* pCoder, td_endian_t endian, uint8_t* data, int32_t size, td_coder_t type);
 void tCoderClear(SCoder* pCoder);

source/client/inc/clientInt.h

@@ -254,8 +254,6 @@ extern int (*handleRequestRspFp[TDMT_MAX])(void*, const SDataBuf* pMsg, int32_t
 int           genericRspCallback(void* param, const SDataBuf* pMsg, int32_t code);
 SMsgSendInfo* buildMsgInfoImpl(SRequestObj* pReqObj);
 
-int taos_init();
-
 void*    createTscObj(const char* user, const char* auth, const char* db, SAppInstInfo* pAppInfo);
 void     destroyTscObj(void* pObj);
 STscObj* acquireTscObj(int64_t rid);

source/client/src/clientImpl.c

@@ -103,6 +103,7 @@ TAOS* taos_connect_internal(const char* ip, const char* user, const char* pass,
 
   if (port) {
     epSet.epSet.eps[0].port = port;
+    epSet.epSet.eps[1].port = port;
   }
 
   char* key = getClusterKey(user, secretEncrypt, ip, port);
@@ -187,8 +188,8 @@ int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtC
       setResPrecision(&pRequest->body.resInfo, (*pQuery)->precision);
     }
 
-    TSWAP(pRequest->dbList, (*pQuery)->pDbList, SArray*);
-    TSWAP(pRequest->tableList, (*pQuery)->pTableList, SArray*);
+    TSWAP(pRequest->dbList, (*pQuery)->pDbList);
+    TSWAP(pRequest->tableList, (*pQuery)->pTableList);
   }
 
   return code;
@@ -245,7 +246,6 @@ void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t
   ASSERT(pSchema != NULL && numOfCols > 0);
 
   pResInfo->numOfCols = numOfCols;
-  // TODO handle memory leak
   if (pResInfo->fields != NULL) {
     taosMemoryFree(pResInfo->fields);
   }

source/client/src/taos.def

 taos_cleanup
 taos_options
 taos_set_config
+taos_init
 taos_connect
 taos_connect_l
 taos_connect_auth

source/client/src/tmq.c

@@ -666,7 +666,6 @@ tmq_resp_err_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
   code = param.rspErr;
   if (code != 0) goto FAIL;
 
-  // TODO: add max retry cnt
   while (TSDB_CODE_MND_CONSUMER_NOT_READY == tmqAskEp(tmq, false)) {
     tscDebug("not ready, retry");
     taosMsleep(500);
@@ -683,7 +682,7 @@ tmq_resp_err_t tmq_subscribe(tmq_t* tmq, const tmq_list_t* topic_list) {
   code = 0;
 FAIL:
   if (req.topicNames != NULL) taosArrayDestroyP(req.topicNames, taosMemoryFree);
-  if (code != 0) {
+  if (code != 0 && buf) {
     taosMemoryFree(buf);
   }
   return code;
@@ -1265,6 +1264,7 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t wait_time) {
     return (TAOS_RES*)rspObj;
   }
 
+  // in no topic status also need process delayed task
   if (atomic_load_8(&tmq->status) == TMQ_CONSUMER_STATUS__INIT) {
     return NULL;
   }
@@ -1285,6 +1285,9 @@ TAOS_RES* tmq_consumer_poll(tmq_t* tmq, int64_t wait_time) {
         return NULL;
       }
       tsem_timewait(&tmq->rspSem, leftTime * 1000);
+    } else {
+      // use tsem_timewait instead of tsem_wait to avoid unexpected stuck
+      tsem_timewait(&tmq->rspSem, 500 * 1000);
     }
   }
 }

source/common/src/tmsg.c

@@ -1302,6 +1302,11 @@ int32_t tDeserializeSGetUserAuthRsp(void *buf, int32_t bufLen, SGetUserAuthRsp *
   return 0;
 }
 
+void tFreeSGetUserAuthRsp(SGetUserAuthRsp *pRsp) {
+  taosHashCleanup(pRsp->readDbs);
+  taosHashCleanup(pRsp->writeDbs);
+}
+
 int32_t tSerializeSCreateDropMQSBNodeReq(void *buf, int32_t bufLen, SMCreateQnodeReq *pReq) {
   SCoder encoder = {0};
   tCoderInit(&encoder, TD_LITTLE_ENDIAN, buf, bufLen, TD_ENCODER);
@@ -3226,7 +3231,7 @@ int32_t tEncodeSMqCMCommitOffsetReq(SCoder *encoder, const SMqCMCommitOffsetReq
 int32_t tDecodeSMqCMCommitOffsetReq(SCoder *decoder, SMqCMCommitOffsetReq *pReq) {
   if (tStartDecode(decoder) < 0) return -1;
   if (tDecodeI32(decoder, &pReq->num) < 0) return -1;
-  pReq->offsets = (SMqOffset *)TCODER_MALLOC(decoder, sizeof(SMqOffset) * pReq->num);
+  pReq->offsets = (SMqOffset *)tCoderMalloc(decoder, sizeof(SMqOffset) * pReq->num);
   if (pReq->offsets == NULL) return -1;
   for (int32_t i = 0; i < pReq->num; i++) {
     tDecodeSMqOffset(decoder, &pReq->offsets[i]);
@@ -3509,7 +3514,7 @@ int tDecodeSVCreateTbBatchRsp(SCoder *pCoder, SVCreateTbBatchRsp *pRsp) {
   if (tStartDecode(pCoder) < 0) return -1;
 
   if (tDecodeI32v(pCoder, &pRsp->nRsps) < 0) return -1;
-  pRsp->pRsps = (SVCreateTbRsp *)TCODER_MALLOC(pCoder, sizeof(*pRsp->pRsps) * pRsp->nRsps);
+  pRsp->pRsps = (SVCreateTbRsp *)tCoderMalloc(pCoder, sizeof(*pRsp->pRsps) * pRsp->nRsps);
   for (int32_t i = 0; i < pRsp->nRsps; i++) {
     if (tDecodeSVCreateTbRsp(pCoder, pRsp->pRsps + i) < 0) return -1;
   }
@@ -3738,7 +3743,7 @@ int tDecodeSVCreateTbBatchReq(SCoder *pCoder, SVCreateTbBatchReq *pReq) {
   if (tStartDecode(pCoder) < 0) return -1;
 
   if (tDecodeI32v(pCoder, &pReq->nReqs) < 0) return -1;
-  pReq->pReqs = (SVCreateTbReq *)TCODER_MALLOC(pCoder, sizeof(SVCreateTbReq) * pReq->nReqs);
+  pReq->pReqs = (SVCreateTbReq *)tCoderMalloc(pCoder, sizeof(SVCreateTbReq) * pReq->nReqs);
   if (pReq->pReqs == NULL) return -1;
   for (int iReq = 0; iReq < pReq->nReqs; iReq++) {
     if (tDecodeSVCreateTbReq(pCoder, pReq->pReqs + iReq) < 0) return -1;

source/common/src/ttypes.c

@@ -651,35 +651,35 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size, void *buf
   switch (type) {
     case TSDB_DATA_TYPE_INT:
     case TSDB_DATA_TYPE_UINT: {
-      TSWAP(*(int32_t *)(pLeft), *(int32_t *)(pRight), int32_t);
+      TSWAP(*(int32_t *)(pLeft), *(int32_t *)(pRight));
       break;
     }
 
     case TSDB_DATA_TYPE_BIGINT:
     case TSDB_DATA_TYPE_UBIGINT:
     case TSDB_DATA_TYPE_TIMESTAMP: {
-      TSWAP(*(int64_t *)(pLeft), *(int64_t *)(pRight), int64_t);
+      TSWAP(*(int64_t *)(pLeft), *(int64_t *)(pRight));
       break;
     }
     case TSDB_DATA_TYPE_DOUBLE: {
-      TSWAP(*(double *)(pLeft), *(double *)(pRight), double);
+      TSWAP(*(double *)(pLeft), *(double *)(pRight));
       break;
     }
     case TSDB_DATA_TYPE_SMALLINT:
     case TSDB_DATA_TYPE_USMALLINT: {
-      TSWAP(*(int16_t *)(pLeft), *(int16_t *)(pRight), int16_t);
+      TSWAP(*(int16_t *)(pLeft), *(int16_t *)(pRight));
       break;
     }
 
     case TSDB_DATA_TYPE_FLOAT: {
-      TSWAP(*(float *)(pLeft), *(float *)(pRight), float);
+      TSWAP(*(float *)(pLeft), *(float *)(pRight));
       break;
     }
 
     case TSDB_DATA_TYPE_BOOL:
     case TSDB_DATA_TYPE_TINYINT:
     case TSDB_DATA_TYPE_UTINYINT: {
-      TSWAP(*(int8_t *)(pLeft), *(int8_t *)(pRight), int8_t);
+      TSWAP(*(int8_t *)(pLeft), *(int8_t *)(pRight));
       break;
     }
 

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

@@ -120,7 +120,7 @@ int32_t dmReadEps(SDnode *pDnode) {
       goto PRASE_DNODE_OVER;
     }
 
-    dnodeEp.id = dnodeId->valueint;
+    dnodeEp.id = did->valueint;
 
     cJSON *dnodeFqdn = cJSON_GetObjectItem(node, "fqdn");
     if (!dnodeFqdn || dnodeFqdn->type != cJSON_String || dnodeFqdn->valuestring == NULL) {
@@ -156,11 +156,6 @@ PRASE_DNODE_OVER:
   if (root != NULL) cJSON_Delete(root);
   if (pFile != NULL) taosCloseFile(&pFile);
 
-  if (dmIsEpChanged(pDnode, pDnode->data.dnodeId, pDnode->data.localEp)) {
-    dError("localEp %s different with %s and need reconfigured", pDnode->data.localEp, file);
-    return -1;
-  }
-
   if (taosArrayGetSize(pDnode->data.dnodeEps) == 0) {
     SDnodeEp dnodeEp = {0};
     dnodeEp.isMnode = 1;
@@ -170,6 +165,11 @@ PRASE_DNODE_OVER:
 
   dmResetEps(pDnode, pDnode->data.dnodeEps);
 
+  if (dmIsEpChanged(pDnode, pDnode->data.dnodeId, pDnode->data.localEp)) {
+    dError("localEp %s different with %s and need reconfigured", pDnode->data.localEp, file);
+    return -1;
+  }
+
   terrno = code;
   return code;
 }
@@ -291,13 +291,17 @@ static void dmPrintEps(SDnode *pDnode) {
 
 static bool dmIsEpChanged(SDnode *pDnode, int32_t dnodeId, const char *ep) {
   bool changed = false;
+  if (dnodeId == 0) return changed;
   taosRLockLatch(&pDnode->data.latch);
 
   SDnodeEp *pDnodeEp = taosHashGet(pDnode->data.dnodeHash, &dnodeId, sizeof(int32_t));
   if (pDnodeEp != NULL) {
-    char epstr[TSDB_EP_LEN + 1];
+    char epstr[TSDB_EP_LEN + 1] = {0};
     snprintf(epstr, TSDB_EP_LEN, "%s:%u", pDnodeEp->ep.fqdn, pDnodeEp->ep.port);
-    changed = strcmp(ep, epstr) != 0;
+    changed = (strcmp(ep, epstr) != 0);
+    if (changed) {
+      dError("dnode:%d, localEp %s different from %s", dnodeId, ep, epstr);
+    }
   }
 
   taosRUnLockLatch(&pDnode->data.latch);

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

@@ -23,10 +23,8 @@ extern "C" {
 #endif
 
 enum {
-  // MQ_CONSUMER_STATUS__INIT = 1,
   MQ_CONSUMER_STATUS__MODIFY = 1,
   MQ_CONSUMER_STATUS__MODIFY_IN_REB,
-  // MQ_CONSUMER_STATUS__IDLE,
   MQ_CONSUMER_STATUS__READY,
   MQ_CONSUMER_STATUS__LOST,
   MQ_CONSUMER_STATUS__LOST_IN_REB,

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

@@ -89,6 +89,7 @@ typedef enum {
   TRN_TYPE_DROP_STREAM = 1020,
   TRN_TYPE_ALTER_STREAM = 1021,
   TRN_TYPE_CONSUMER_LOST = 1022,
+  TRN_TYPE_CONSUMER_RECOVER = 1023,
   TRN_TYPE_BASIC_SCOPE_END,
   TRN_TYPE_GLOBAL_SCOPE = 2000,
   TRN_TYPE_CREATE_DNODE = 2001,
@@ -463,12 +464,14 @@ enum {
   CONSUMER_UPDATE__ADD,
   CONSUMER_UPDATE__REMOVE,
   CONSUMER_UPDATE__LOST,
+  CONSUMER_UPDATE__RECOVER,
   CONSUMER_UPDATE__MODIFY,
 };
 
 typedef struct {
   int64_t consumerId;
   char    cgroup[TSDB_CGROUP_LEN];
+  char    appId[TSDB_CGROUP_LEN];
   int8_t  updateType;  // used only for update
   int32_t epoch;
   int32_t status;
@@ -479,6 +482,17 @@ typedef struct {
   SArray*  currentTopics;     // SArray<char*>
   SArray*  rebNewTopics;      // SArray<char*>
   SArray*  rebRemovedTopics;  // SArray<char*>
+
+  // subscribed by user
+  SArray* assignedTopics;  // SArray<char*>
+
+  // data for display
+  int32_t pid;
+  SEpSet  ep;
+  int64_t upTime;
+  int64_t subscribeTime;
+  int64_t rebalanceTime;
+
 } SMqConsumerObj;
 
 SMqConsumerObj* tNewSMqConsumerObj(int64_t consumerId, char cgroup[TSDB_CGROUP_LEN]);

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

@@ -37,6 +37,8 @@
 
 static int8_t mqInRebFlag = 0;
 
+static const char *mndConsumerStatusName(int status);
+
 static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer);
 static int32_t mndConsumerActionDelete(SSdb *pSdb, SMqConsumerObj *pConsumer);
 static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pConsumer, SMqConsumerObj *pNewConsumer);
@@ -48,6 +50,7 @@ static int32_t mndProcessSubscribeReq(SNodeMsg *pMsg);
 static int32_t mndProcessAskEpReq(SNodeMsg *pMsg);
 static int32_t mndProcessMqTimerMsg(SNodeMsg *pMsg);
 static int32_t mndProcessConsumerLostMsg(SNodeMsg *pMsg);
+static int32_t mndProcessConsumerRecoverMsg(SNodeMsg *pMsg);
 
 int32_t mndInitConsumer(SMnode *pMnode) {
   SSdbTable table = {.sdbType = SDB_CONSUMER,
@@ -62,6 +65,11 @@ int32_t mndInitConsumer(SMnode *pMnode) {
   mndSetMsgHandle(pMnode, TDMT_MND_MQ_ASK_EP, mndProcessAskEpReq);
   mndSetMsgHandle(pMnode, TDMT_MND_MQ_TIMER, mndProcessMqTimerMsg);
   mndSetMsgHandle(pMnode, TDMT_MND_MQ_CONSUMER_LOST, mndProcessConsumerLostMsg);
+  mndSetMsgHandle(pMnode, TDMT_MND_MQ_CONSUMER_RECOVER, mndProcessConsumerRecoverMsg);
+
+  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_CONSUMERS, mndRetrieveConsumer);
+  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_CONSUMERS, mndCancelGetNextConsumer);
+
   return sdbSetTable(pMnode->pSdb, table);
 }
 
@@ -91,6 +99,30 @@ FAIL:
   return -1;
 }
 
+static int32_t mndProcessConsumerRecoverMsg(SNodeMsg *pMsg) {
+  SMnode                *pMnode = pMsg->pNode;
+  SMqConsumerRecoverMsg *pRecoverMsg = pMsg->rpcMsg.pCont;
+  SMqConsumerObj        *pConsumer = mndAcquireConsumer(pMnode, pRecoverMsg->consumerId);
+  ASSERT(pConsumer);
+
+  SMqConsumerObj *pConsumerNew = tNewSMqConsumerObj(pConsumer->consumerId, pConsumer->cgroup);
+  pConsumerNew->updateType = CONSUMER_UPDATE__RECOVER;
+
+  mndReleaseConsumer(pMnode, pConsumer);
+
+  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_CONSUMER_RECOVER, &pMsg->rpcMsg);
+  if (pTrans == NULL) goto FAIL;
+  if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) goto FAIL;
+  if (mndTransPrepare(pMnode, pTrans) != 0) goto FAIL;
+
+  mndTransDrop(pTrans);
+  return 0;
+FAIL:
+  tDeleteSMqConsumerObj(pConsumerNew);
+  mndTransDrop(pTrans);
+  return -1;
+}
+
 static SMqRebSubscribe *mndGetOrCreateRebSub(SHashObj *pHash, const char *key) {
   SMqRebSubscribe *pRebSub = taosHashGet(pHash, key, strlen(key) + 1);
   if (pRebSub == NULL) {
@@ -216,8 +248,15 @@ static int32_t mndProcessAskEpReq(SNodeMsg *pMsg) {
   // 1. check consumer status
   int32_t status = atomic_load_32(&pConsumer->status);
 
-  if (status == MQ_CONSUMER_STATUS__LOST) {
-    // TODO: recover consumer
+  if (status == MQ_CONSUMER_STATUS__LOST_REBD) {
+    SMqConsumerRecoverMsg *pRecoverMsg = rpcMallocCont(sizeof(SMqConsumerRecoverMsg));
+
+    pRecoverMsg->consumerId = consumerId;
+    SRpcMsg *pRpcMsg = taosMemoryCalloc(1, sizeof(SRpcMsg));
+    pRpcMsg->msgType = TDMT_MND_MQ_CONSUMER_RECOVER;
+    pRpcMsg->pCont = pRecoverMsg;
+    pRpcMsg->contLen = sizeof(SMqConsumerRecoverMsg);
+    tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, pRpcMsg);
   }
 
   if (status != MQ_CONSUMER_STATUS__READY) {
@@ -366,10 +405,14 @@ static int32_t mndProcessSubscribeReq(SNodeMsg *pMsg) {
   if (pConsumerOld == NULL) {
     pConsumerNew = tNewSMqConsumerObj(consumerId, cgroup);
     pConsumerNew->updateType = CONSUMER_UPDATE__MODIFY;
-    /*pConsumerNew->waitingRebTopics = newSub;*/
     pConsumerNew->rebNewTopics = newSub;
     subscribe.topicNames = NULL;
 
+    for (int32_t i = 0; i < newTopicNum; i++) {
+      char *newTopicCopy = strdup(taosArrayGetP(newSub, i));
+      taosArrayPush(pConsumerNew->assignedTopics, &newTopicCopy);
+    }
+
     STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_SUBSCRIBE, &pMsg->rpcMsg);
     if (pTrans == NULL) goto SUBSCRIBE_OVER;
     if (mndSetConsumerCommitLogs(pMnode, pTrans, pConsumerNew) != 0) goto SUBSCRIBE_OVER;
@@ -389,7 +432,11 @@ static int32_t mndProcessSubscribeReq(SNodeMsg *pMsg) {
       goto SUBSCRIBE_OVER;
     }
     pConsumerNew->updateType = CONSUMER_UPDATE__MODIFY;
-    /*pConsumerOld->waitingRebTopics = newSub;*/
+
+    for (int32_t i = 0; i < newTopicNum; i++) {
+      char *newTopicCopy = strdup(taosArrayGetP(newSub, i));
+      taosArrayPush(pConsumerNew->assignedTopics, &newTopicCopy);
+    }
 
     int32_t oldTopicNum = 0;
     if (pConsumerOld->currentTopics) {
@@ -532,6 +579,7 @@ CM_DECODE_OVER:
 
 static int32_t mndConsumerActionInsert(SSdb *pSdb, SMqConsumerObj *pConsumer) {
   mTrace("consumer:%" PRId64 ", perform insert action", pConsumer->consumerId);
+  pConsumer->subscribeTime = pConsumer->upTime;
   return 0;
 }
 
@@ -557,17 +605,45 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
     pOldConsumer->rebRemovedTopics = pNewConsumer->rebRemovedTopics;
     pNewConsumer->rebRemovedTopics = tmp;
 
+    tmp = pOldConsumer->assignedTopics;
+    pOldConsumer->assignedTopics = pNewConsumer->assignedTopics;
+    pNewConsumer->assignedTopics = tmp;
+
+    pOldConsumer->subscribeTime = pNewConsumer->upTime;
+
     pOldConsumer->status = MQ_CONSUMER_STATUS__MODIFY;
   } else if (pNewConsumer->updateType == CONSUMER_UPDATE__LOST) {
+    ASSERT(taosArrayGetSize(pOldConsumer->rebNewTopics) == 0);
+    ASSERT(taosArrayGetSize(pOldConsumer->rebRemovedTopics) == 0);
+
     int32_t sz = taosArrayGetSize(pOldConsumer->currentTopics);
-    pOldConsumer->rebRemovedTopics = taosArrayInit(sz, sizeof(void *));
+    /*pOldConsumer->rebRemovedTopics = taosArrayInit(sz, sizeof(void *));*/
     for (int32_t i = 0; i < sz; i++) {
       char *topic = strdup(taosArrayGetP(pOldConsumer->currentTopics, i));
-      taosArrayPush(pNewConsumer->rebRemovedTopics, &topic);
+      taosArrayPush(pOldConsumer->rebRemovedTopics, &topic);
     }
+
+    pOldConsumer->rebalanceTime = pNewConsumer->upTime;
+
     pOldConsumer->status = MQ_CONSUMER_STATUS__LOST;
+  } else if (pNewConsumer->updateType == CONSUMER_UPDATE__RECOVER) {
+    ASSERT(taosArrayGetSize(pOldConsumer->currentTopics) == 0);
+    ASSERT(taosArrayGetSize(pOldConsumer->rebNewTopics) == 0);
+
+    int32_t sz = taosArrayGetSize(pOldConsumer->assignedTopics);
+    for (int32_t i = 0; i < sz; i++) {
+      char *topic = strdup(taosArrayGetP(pOldConsumer->assignedTopics, i));
+      taosArrayPush(pOldConsumer->rebNewTopics, &topic);
+    }
+
+    pOldConsumer->rebalanceTime = pNewConsumer->upTime;
+
+    pOldConsumer->status = MQ_CONSUMER_STATUS__MODIFY;
   } else if (pNewConsumer->updateType == CONSUMER_UPDATE__TOUCH) {
     atomic_add_fetch_32(&pOldConsumer->epoch, 1);
+
+    pOldConsumer->rebalanceTime = pNewConsumer->upTime;
+
   } else if (pNewConsumer->updateType == CONSUMER_UPDATE__ADD) {
     ASSERT(taosArrayGetSize(pNewConsumer->rebNewTopics) == 1);
     ASSERT(taosArrayGetSize(pNewConsumer->rebRemovedTopics) == 0);
@@ -612,6 +688,9 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
         pOldConsumer->status = MQ_CONSUMER_STATUS__LOST_IN_REB;
       }
     }
+
+    pOldConsumer->rebalanceTime = pNewConsumer->upTime;
+
     atomic_add_fetch_32(&pOldConsumer->epoch, 1);
   } else if (pNewConsumer->updateType == CONSUMER_UPDATE__REMOVE) {
     ASSERT(taosArrayGetSize(pNewConsumer->rebNewTopics) == 0);
@@ -668,6 +747,9 @@ static int32_t mndConsumerActionUpdate(SSdb *pSdb, SMqConsumerObj *pOldConsumer,
         pOldConsumer->status = MQ_CONSUMER_STATUS__LOST_IN_REB;
       }
     }
+
+    pOldConsumer->rebalanceTime = pNewConsumer->upTime;
+
     atomic_add_fetch_32(&pOldConsumer->epoch, 1);
   }
 
@@ -688,3 +770,104 @@ void mndReleaseConsumer(SMnode *pMnode, SMqConsumerObj *pConsumer) {
   SSdb *pSdb = pMnode->pSdb;
   sdbRelease(pSdb, pConsumer);
 }
+
+static int32_t mndRetrieveConsumer(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rowsCapacity) {
+  SMnode         *pMnode = pReq->pNode;
+  SSdb           *pSdb = pMnode->pSdb;
+  int32_t         numOfRows = 0;
+  SMqConsumerObj *pConsumer = NULL;
+
+  while (numOfRows < rowsCapacity) {
+    pShow->pIter = sdbFetch(pSdb, SDB_CONSUMER, pShow->pIter, (void **)&pConsumer);
+    if (pShow->pIter == NULL) break;
+
+    SColumnInfoData *pColInfo;
+    int32_t          cols = 0;
+
+    taosRLockLatch(&pConsumer->lock);
+
+    // consumer id
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)&pConsumer->consumerId, false);
+
+    // group id
+    char groupId[TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE] = {0};
+    tstrncpy(varDataVal(groupId), pConsumer->cgroup, TSDB_CGROUP_LEN);
+    varDataSetLen(groupId, strlen(varDataVal(groupId)));
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)groupId, false);
+
+    // app id
+    char appId[TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE] = {0};
+    tstrncpy(varDataVal(appId), pConsumer->appId, TSDB_CGROUP_LEN);
+    varDataSetLen(appId, strlen(varDataVal(appId)));
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)appId, false);
+
+    // status
+    char status[20 + VARSTR_HEADER_SIZE] = {0};
+    tstrncpy(varDataVal(status), mndConsumerStatusName(pConsumer->status), 20);
+    varDataSetLen(status, strlen(varDataVal(status)));
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)status, false);
+
+    // subscribed topics
+    char  topics[TSDB_SHOW_LIST_LEN + VARSTR_HEADER_SIZE] = {0};
+    char *showStr = taosShowStrArray(pConsumer->assignedTopics);
+    tstrncpy(varDataVal(topics), showStr, TSDB_SHOW_LIST_LEN);
+    taosMemoryFree(showStr);
+    varDataSetLen(topics, strlen(varDataVal(topics)));
+
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)topics, false);
+
+    // pid
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)&pConsumer->pid, true);
+
+    // end point
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)&pConsumer->ep, true);
+
+    // up time
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)&pConsumer->upTime, false);
+
+    // subscribe time
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)&pConsumer->subscribeTime, false);
+
+    // rebalance time
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)&pConsumer->rebalanceTime, pConsumer->rebalanceTime == 0);
+
+    taosRUnLockLatch(&pConsumer->lock);
+    sdbRelease(pSdb, pConsumer);
+
+    numOfRows++;
+  }
+
+  pShow->numOfRows += numOfRows;
+  return numOfRows;
+}
+
+static void mndCancelGetNextConsumer(SMnode *pMnode, void *pIter) {
+  SSdb *pSdb = pMnode->pSdb;
+  sdbCancelFetch(pSdb, pIter);
+}
+
+static const char *mndConsumerStatusName(int status) {
+  switch (status) {
+    case MQ_CONSUMER_STATUS__READY:
+      return "ready";
+    case MQ_CONSUMER_STATUS__LOST:
+    case MQ_CONSUMER_STATUS__LOST_REBD:
+    case MQ_CONSUMER_STATUS__LOST_IN_REB:
+      return "lost";
+    case MQ_CONSUMER_STATUS__MODIFY:
+    case MQ_CONSUMER_STATUS__MODIFY_IN_REB:
+      return "rebalancing";
+    default:
+      return "unknown";
+  }
+}

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

@@ -34,15 +34,20 @@ SMqConsumerObj *tNewSMqConsumerObj(int64_t consumerId, char cgroup[TSDB_CGROUP_L
   pConsumer->currentTopics = taosArrayInit(0, sizeof(void *));
   pConsumer->rebNewTopics = taosArrayInit(0, sizeof(void *));
   pConsumer->rebRemovedTopics = taosArrayInit(0, sizeof(void *));
+  pConsumer->assignedTopics = taosArrayInit(0, sizeof(void *));
 
-  if (pConsumer->currentTopics == NULL || pConsumer->rebNewTopics == NULL || pConsumer->rebRemovedTopics == NULL) {
+  if (pConsumer->currentTopics == NULL || pConsumer->rebNewTopics == NULL || pConsumer->rebRemovedTopics == NULL ||
+      pConsumer->assignedTopics == NULL) {
     taosArrayDestroy(pConsumer->currentTopics);
     taosArrayDestroy(pConsumer->rebNewTopics);
     taosArrayDestroy(pConsumer->rebRemovedTopics);
+    taosArrayDestroy(pConsumer->assignedTopics);
     taosMemoryFree(pConsumer);
     return NULL;
   }
 
+  pConsumer->upTime = taosGetTimestampMs();
+
   return pConsumer;
 }
 
@@ -56,6 +61,9 @@ void tDeleteSMqConsumerObj(SMqConsumerObj *pConsumer) {
   if (pConsumer->rebRemovedTopics) {
     taosArrayDestroyP(pConsumer->rebRemovedTopics, (FDelete)taosMemoryFree);
   }
+  if (pConsumer->assignedTopics) {
+    taosArrayDestroyP(pConsumer->assignedTopics, (FDelete)taosMemoryFree);
+  }
 }
 
 int32_t tEncodeSMqConsumerObj(void **buf, const SMqConsumerObj *pConsumer) {
@@ -67,6 +75,12 @@ int32_t tEncodeSMqConsumerObj(void **buf, const SMqConsumerObj *pConsumer) {
   tlen += taosEncodeFixedI32(buf, pConsumer->epoch);
   tlen += taosEncodeFixedI32(buf, pConsumer->status);
 
+  tlen += taosEncodeFixedI32(buf, pConsumer->pid);
+  tlen += taosEncodeSEpSet(buf, &pConsumer->ep);
+  tlen += taosEncodeFixedI64(buf, pConsumer->upTime);
+  tlen += taosEncodeFixedI64(buf, pConsumer->subscribeTime);
+  tlen += taosEncodeFixedI64(buf, pConsumer->rebalanceTime);
+
   // current topics
   if (pConsumer->currentTopics) {
     sz = taosArrayGetSize(pConsumer->currentTopics);
@@ -103,6 +117,18 @@ int32_t tEncodeSMqConsumerObj(void **buf, const SMqConsumerObj *pConsumer) {
     tlen += taosEncodeFixedI32(buf, 0);
   }
 
+  // lost topics
+  if (pConsumer->assignedTopics) {
+    sz = taosArrayGetSize(pConsumer->assignedTopics);
+    tlen += taosEncodeFixedI32(buf, sz);
+    for (int32_t i = 0; i < sz; i++) {
+      char *topic = taosArrayGetP(pConsumer->assignedTopics, i);
+      tlen += taosEncodeString(buf, topic);
+    }
+  } else {
+    tlen += taosEncodeFixedI32(buf, 0);
+  }
+
   return tlen;
 }
 
@@ -114,6 +140,12 @@ void *tDecodeSMqConsumerObj(const void *buf, SMqConsumerObj *pConsumer) {
   buf = taosDecodeFixedI32(buf, &pConsumer->epoch);
   buf = taosDecodeFixedI32(buf, &pConsumer->status);
 
+  buf = taosDecodeFixedI32(buf, &pConsumer->pid);
+  buf = taosDecodeSEpSet(buf, &pConsumer->ep);
+  buf = taosDecodeFixedI64(buf, &pConsumer->upTime);
+  buf = taosDecodeFixedI64(buf, &pConsumer->subscribeTime);
+  buf = taosDecodeFixedI64(buf, &pConsumer->rebalanceTime);
+
   // current topics
   buf = taosDecodeFixedI32(buf, &sz);
   pConsumer->currentTopics = taosArrayInit(sz, sizeof(void *));
@@ -141,6 +173,15 @@ void *tDecodeSMqConsumerObj(const void *buf, SMqConsumerObj *pConsumer) {
     taosArrayPush(pConsumer->rebRemovedTopics, &topic);
   }
 
+  // reb removed topics
+  buf = taosDecodeFixedI32(buf, &sz);
+  pConsumer->assignedTopics = taosArrayInit(sz, sizeof(void *));
+  for (int32_t i = 0; i < sz; i++) {
+    char *topic;
+    buf = taosDecodeString(buf, &topic);
+    taosArrayPush(pConsumer->assignedTopics, &topic);
+  }
+
   return (void *)buf;
 }
 
@@ -329,6 +370,7 @@ int32_t tEncodeSMqSubActionLogEntry(void **buf, const SMqSubActionLogEntry *pEnt
   tlen += taosEncodeArray(buf, pEntry->consumers, (FEncode)tEncodeSMqSubActionLogEntry);
   return tlen;
 }
+
 void *tDecodeSMqSubActionLogEntry(const void *buf, SMqSubActionLogEntry *pEntry) {
   buf = taosDecodeFixedI32(buf, &pEntry->epoch);
   buf = taosDecodeArray(buf, &pEntry->consumers, (FDecode)tDecodeSMqSubActionLogEntry, sizeof(SMqSubActionLogEntry));

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

@@ -165,7 +165,6 @@ static const SInfosTableSchema userUsersSchema[] = {
     {.name = "name", .bytes = TSDB_USER_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
     {.name = "privilege", .bytes = 10 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
     {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
-    {.name = "account", .bytes = TSDB_USER_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR},
 };
 
 static const SInfosTableSchema grantsSchema[] = {
@@ -200,23 +199,6 @@ static const SInfosTableSchema vgroupsSchema[] = {
     {.name = "file_size", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
 };
 
-static const SInfosTableSchema consumerSchema[] = {
-    {.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
-    {.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
-    {.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
-    {.name = "status", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
-    // ep
-    // up time
-    // topics
-};
-
-static const SInfosTableSchema subscribeSchema[] = {
-    {.name = "topic_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
-    {.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
-    {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
-    {.name = "client_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
-};
-
 static const SInfosTableSchema smaSchema[] = {
     {.name = "sma_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR},
     {.name = "create_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
@@ -283,8 +265,6 @@ static const SInfosTableMeta infosMeta[] = {
     {TSDB_INS_TABLE_USER_USERS, userUsersSchema, tListLen(userUsersSchema)},
     {TSDB_INS_TABLE_LICENCES, grantsSchema, tListLen(grantsSchema)},
     {TSDB_INS_TABLE_VGROUPS, vgroupsSchema, tListLen(vgroupsSchema)},
-    {TSDB_INS_TABLE_CONSUMERS, consumerSchema, tListLen(consumerSchema)},
-    {TSDB_INS_TABLE_SUBSCRIBES, subscribeSchema, tListLen(subscribeSchema)},
     {TSDB_INS_TABLE_TRANS, transSchema, tListLen(transSchema)},
     {TSDB_INS_TABLE_SMAS, smaSchema, tListLen(smaSchema)},
     {TSDB_INS_TABLE_CONFIGS, configSchema, tListLen(configSchema)},

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

@@ -49,13 +49,15 @@ static const SPerfsTableSchema topicSchema[] = {
 
 static const SPerfsTableSchema consumerSchema[] = {
     {.name = "consumer_id", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
-    {.name = "app_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
-    {.name = "status", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
+    {.name = "app_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "status", .bytes = 20 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "topics", .bytes = TSDB_SHOW_LIST_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "pid", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "end_point", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "up_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
+    {.name = "subscribe_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
+    {.name = "rebalance_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP},
 };
 
 static const SPerfsTableSchema subscriptionSchema[] = {
@@ -63,6 +65,12 @@ static const SPerfsTableSchema subscriptionSchema[] = {
     {.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
     {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "consumer_id", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
+};
+
+static const SPerfsTableSchema offsetSchema[] = {
+    {.name = "topic_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "group_id", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY},
+    {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT},
     {.name = "committed_offset", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
     {.name = "current_offset", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
     {.name = "skip_log_cnt", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT},
@@ -74,6 +82,7 @@ static const SPerfsTableMeta perfsMeta[] = {
     {TSDB_PERFS_TABLE_TOPICS, topicSchema, tListLen(topicSchema)},
     {TSDB_PERFS_TABLE_CONSUMERS, consumerSchema, tListLen(consumerSchema)},
     {TSDB_PERFS_TABLE_SUBSCRIPTIONS, subscriptionSchema, tListLen(subscriptionSchema)},
+    {TSDB_PERFS_TABLE_OFFSETS, offsetSchema, tListLen(offsetSchema)},
 };
 
 // connection/application/

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

@@ -196,7 +196,7 @@ static int32_t mndProcessConnectReq(SNodeMsg *pReq) {
     goto CONN_OVER;
   }
   if (0 != strncmp(connReq.passwd, pUser->pass, TSDB_PASSWORD_LEN - 1)) {
-    mError("user:%s, failed to auth while acquire user\n %s \r\n %s", pReq->user, connReq.passwd, pUser->pass);
+    mError("user:%s, failed to auth while acquire user, input:%s saved:%s", pReq->user, connReq.passwd, pUser->pass);
     code = TSDB_CODE_RPC_AUTH_FAILURE;
     goto CONN_OVER;
   }

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

@@ -489,7 +489,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
     int32_t levelNum = LIST_LENGTH(pPlan->pSubplans);
     if (levelNum != 1) {
       qDestroyQueryPlan(pPlan);
-      terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC;
+      terrno = TSDB_CODE_MND_INVALID_TOPIC_QUERY;
       return -1;
     }
 
@@ -498,7 +498,7 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
     int32_t opNum = LIST_LENGTH(inner->pNodeList);
     if (opNum != 1) {
       qDestroyQueryPlan(pPlan);
-      terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC;
+      terrno = TSDB_CODE_MND_INVALID_TOPIC_QUERY;
       return -1;
     }
     plan = nodesListGetNode(inner->pNodeList, 0);

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

@@ -395,7 +395,9 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt
     }
   }
   // get length
-  if (tEncodeSize(tEncodeSVCreateStbReq, &req, contLen) < 0) {
+  int32_t ret = 0;
+  tEncodeSize(tEncodeSVCreateStbReq, &req, contLen, ret);
+  if (ret < 0) {
     return NULL;
   }
 

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

@@ -261,7 +261,7 @@ 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 || pCreate->subscribeDbName[0] == 0) {
-    terrno = TSDB_CODE_MND_INVALID_TOPIC_OPTION;
+    terrno = TSDB_CODE_MND_INVALID_TOPIC;
     return -1;
   }
 

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

@@ -165,8 +165,9 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
   int32_t numOfWriteDbs = 0;
   SDB_GET_INT32(pRaw, dataPos, &numOfReadDbs, _OVER)
   SDB_GET_INT32(pRaw, dataPos, &numOfWriteDbs, _OVER)
-  pUser->readDbs = taosHashInit(numOfReadDbs, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, true);
-  pUser->writeDbs = taosHashInit(numOfWriteDbs, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, true);
+  pUser->readDbs = taosHashInit(numOfReadDbs, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
+  pUser->writeDbs =
+      taosHashInit(numOfWriteDbs, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
   if (pUser->readDbs == NULL || pUser->writeDbs == NULL) goto _OVER;
 
   for (int32_t i = 0; i < numOfReadDbs; ++i) {
@@ -229,8 +230,8 @@ static int32_t mndUserActionUpdate(SSdb *pSdb, SUserObj *pOld, SUserObj *pNew) {
   memcpy(pOld->pass, pNew->pass, TSDB_PASSWORD_LEN);
   pOld->updateTime = pNew->updateTime;
 
-  TSWAP(pOld->readDbs, pNew->readDbs, (void *));
-  TSWAP(pOld->writeDbs, pNew->writeDbs, (void *));
+  TSWAP(pOld->readDbs, pNew->readDbs);
+  TSWAP(pOld->writeDbs, pNew->writeDbs);
 
   return 0;
 }
@@ -340,13 +341,13 @@ _OVER:
   return code;
 }
 
-static int32_t mndUpdateUser(SMnode *pMnode, SUserObj *pOld, SUserObj *pNew, SNodeMsg *pReq) {
+static int32_t mndAlterUser(SMnode *pMnode, SUserObj *pOld, SUserObj *pNew, SNodeMsg *pReq) {
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_ALTER_USER, &pReq->rpcMsg);
   if (pTrans == NULL) {
-    mError("user:%s, failed to update since %s", pOld->user, terrstr());
+    mError("user:%s, failed to alter since %s", pOld->user, terrstr());
     return -1;
   }
-  mDebug("trans:%d, used to update user:%s", pTrans->id, pOld->user);
+  mDebug("trans:%d, used to alter user:%s", pTrans->id, pOld->user);
 
   SSdbRaw *pRedoRaw = mndUserActionEncode(pNew);
   if (pRedoRaw == NULL || mndTransAppendRedolog(pTrans, pRedoRaw) != 0) {
@@ -367,7 +368,8 @@ static int32_t mndUpdateUser(SMnode *pMnode, SUserObj *pOld, SUserObj *pNew, SNo
 }
 
 static SHashObj *mndDupDbHash(SHashObj *pOld) {
-  SHashObj *pNew = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, true);
+  SHashObj *pNew =
+      taosHashInit(taosHashGetSize(pOld), taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
   if (pNew == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return NULL;
@@ -378,8 +380,8 @@ static SHashObj *mndDupDbHash(SHashObj *pOld) {
     int32_t len = strlen(db) + 1;
     if (taosHashPut(pNew, db, len, db, TSDB_DB_FNAME_LEN) != 0) {
       taosHashCancelIterate(pOld, db);
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
       taosHashCleanup(pNew);
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
       return NULL;
     }
     db = taosHashIterate(pOld, db);
@@ -439,7 +441,7 @@ static int32_t mndProcessAlterUserReq(SNodeMsg *pReq) {
   if (alterReq.alterType == TSDB_ALTER_USER_PASSWD) {
     char pass[TSDB_PASSWORD_LEN + 1] = {0};
     taosEncryptPass_c((uint8_t *)alterReq.pass, strlen(alterReq.pass), pass);
-    memcpy(pUser->pass, pass, TSDB_PASSWORD_LEN);
+    memcpy(newUser.pass, pass, TSDB_PASSWORD_LEN);
   } else if (alterReq.alterType == TSDB_ALTER_USER_SUPERUSER) {
     newUser.superUser = alterReq.superUser;
   } else if (alterReq.alterType == TSDB_ALTER_USER_ADD_READ_DB) {
@@ -485,7 +487,7 @@ static int32_t mndProcessAlterUserReq(SNodeMsg *pReq) {
     goto _OVER;
   }
 
-  code = mndUpdateUser(pMnode, pUser, &newUser, pReq);
+  code = mndAlterUser(pMnode, pUser, &newUser, pReq);
   if (code == 0) code = TSDB_CODE_MND_ACTION_IN_PROGRESS;
 
 _OVER:
@@ -632,8 +634,7 @@ static int32_t mndProcessGetUserAuthReq(SNodeMsg *pReq) {
 
 _OVER:
   mndReleaseUser(pMnode, pUser);
-  taosHashCleanup(authRsp.readDbs);
-  taosHashCleanup(authRsp.writeDbs);
+  tFreeSGetUserAuthRsp(&authRsp);
 
   return code;
 }
@@ -670,11 +671,6 @@ static int32_t mndRetrieveUsers(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock *pB
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
     colDataAppend(pColInfo, numOfRows, (const char *)&pUser->createdTime, false);
 
-    cols++;
-    pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
-    STR_WITH_MAXSIZE_TO_VARSTR(name, pUser->acct, pShow->bytes[cols]);
-    colDataAppend(pColInfo, numOfRows, (const char *)name, false);
-
     numOfRows++;
     sdbRelease(pSdb, pUser);
   }

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

@@ -43,7 +43,7 @@
 #include "mndUser.h"
 #include "mndVgroup.h"
 
-#define MQ_TIMER_MS    3000
+#define MQ_TIMER_MS    2000
 #define TRNAS_TIMER_MS 6000
 
 static void *mndBuildTimerMsg(int32_t *pContLen) {
@@ -418,7 +418,6 @@ int64_t mndGenerateUid(char *name, int32_t len) {
   } while (true);
 }
 
-
 int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
                           SMonGrantInfo *pGrantInfo) {
   if (!mndIsMaster(pMnode)) return -1;
@@ -528,4 +527,4 @@ int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgr
 int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad) {
   pLoad->syncState = pMnode->syncMgmt.state;
   return 0;
-}
+}
\ No newline at end of file

source/dnode/mnode/impl/test/CMakeLists.txt

@@ -8,6 +8,7 @@ add_subdirectory(func)
 add_subdirectory(mnode)
 add_subdirectory(profile)
 add_subdirectory(qnode)
+add_subdirectory(sdb)
 add_subdirectory(show)
 add_subdirectory(sma)
 add_subdirectory(snode)

source/dnode/mnode/impl/test/sdb/CMakeLists.txt

+aux_source_directory(. MNODE_SDB_TEST_SRC)
+add_executable(sdbTest ${MNODE_SDB_TEST_SRC})
+target_link_libraries(
+    sdbTest
+    PUBLIC sut
+    PUBLIC sdb
+)
+
+add_test(
+    NAME sdbTest
+    COMMAND sdbTest
+)

source/dnode/mnode/impl/test/sdb/sdbTest.cpp

+/**
+ * @file sdbTest.cpp
+ * @author slguan (slguan@taosdata.com)
+ * @brief MNODE module sdb tests
+ * @version 1.0
+ * @date 2022-04-27
+ *
+ * @copyright Copyright (c) 2022
+ *
+ */
+
+#include "sut.h"
+
+class MndTestShow : public ::testing::Test {
+ protected:
+  static void SetUpTestSuite() { test.Init("/tmp/mnode_test_show", 9021); }
+  static void TearDownTestSuite() { test.Cleanup(); }
+
+  static Testbase test;
+
+ public:
+  void SetUp() override {}
+  void TearDown() override {}
+};
+
+Testbase MndTestShow::test;
+
+TEST_F(MndTestShow, 01_ShowMsg_InvalidMsgMax) {
+  SShowReq showReq = {0};
+  showReq.type = TSDB_MGMT_TABLE_MAX;
+
+  int32_t contLen = tSerializeSShowReq(NULL, 0, &showReq);
+  void*   pReq = rpcMallocCont(contLen);
+  tSerializeSShowReq(pReq, contLen, &showReq);
+  tFreeSShowReq(&showReq);
+
+  SRpcMsg* pRsp = test.SendReq(TDMT_MND_SYSTABLE_RETRIEVE, pReq, contLen);
+  ASSERT_NE(pRsp, nullptr);
+  ASSERT_NE(pRsp->code, 0);
+}
+
+TEST_F(MndTestShow, 02_ShowMsg_InvalidMsgStart) {
+  SShowReq showReq = {0};
+  showReq.type = TSDB_MGMT_TABLE_START;
+
+  int32_t contLen = tSerializeSShowReq(NULL, 0, &showReq);
+  void*   pReq = rpcMallocCont(contLen);
+  tSerializeSShowReq(pReq, contLen, &showReq);
+  tFreeSShowReq(&showReq);
+
+  SRpcMsg* pRsp = test.SendReq(TDMT_MND_SYSTABLE_RETRIEVE, pReq, contLen);
+  ASSERT_NE(pRsp, nullptr);
+  ASSERT_NE(pRsp->code, 0);
+}
+
+TEST_F(MndTestShow, 03_ShowMsg_Conn) {
+  char passwd[] = "taosdata";
+  char secretEncrypt[TSDB_PASSWORD_LEN] = {0};
+  taosEncryptPass_c((uint8_t*)passwd, strlen(passwd), secretEncrypt);
+
+  SConnectReq connectReq = {0};
+  connectReq.pid = 1234;
+  strcpy(connectReq.app, "mnode_test_show");
+  strcpy(connectReq.db, "");
+  strcpy(connectReq.user, "root");
+  strcpy(connectReq.passwd, secretEncrypt);
+
+  int32_t contLen = tSerializeSConnectReq(NULL, 0, &connectReq);
+  void*   pReq = rpcMallocCont(contLen);
+  tSerializeSConnectReq(pReq, contLen, &connectReq);
+
+  SRpcMsg* pRsp = test.SendReq(TDMT_MND_CONNECT, pReq, contLen);
+  ASSERT_NE(pRsp, nullptr);
+  ASSERT_EQ(pRsp->code, 0);
+
+  test.SendShowReq(TSDB_MGMT_TABLE_CONNS, "connections", "");
+  // EXPECT_EQ(test.GetShowRows(), 1);
+}
+
+TEST_F(MndTestShow, 04_ShowMsg_Cluster) {
+  test.SendShowReq(TSDB_MGMT_TABLE_CLUSTER, "cluster", "");
+  EXPECT_EQ(test.GetShowRows(), 1);
+}

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

@@ -77,21 +77,25 @@ typedef struct {
   tb_uid_t uid;
 } STbDbKey;
 
-typedef struct __attribute__((__packed__)) {
+#pragma pack(push, 1)
+typedef struct {
   tb_uid_t uid;
   int32_t  sver;
 } SSkmDbKey;
+#pragma pack(pop)
 
 typedef struct {
   tb_uid_t suid;
   tb_uid_t uid;
 } SCtbIdxKey;
 
-typedef struct __attribute__((__packed__)) {
+#pragma pack(push, 1)
+typedef struct {
   tb_uid_t suid;
   int16_t  cid;
   char     data[];
 } STagIdxKey;
+#pragma pack(pop)
 
 typedef struct {
   int64_t  dtime;

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

@@ -158,7 +158,9 @@ static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME) {
   pKey = &tbDbKey;
   kLen = sizeof(tbDbKey);
 
-  if (tEncodeSize(metaEncodeEntry, pME, vLen) < 0) {
+  int32_t ret = 0;
+  tEncodeSize(metaEncodeEntry, pME, vLen, ret);
+  if (ret < 0) {
     goto _err;
   }
 
@@ -250,7 +252,9 @@ static int metaSaveToSkmDb(SMeta *pMeta, const SMetaEntry *pME) {
   skmDbKey.sver = pSW->sver;
 
   // encode schema
-  if (tEncodeSize(tEncodeSSchemaWrapper, pSW, vLen) < 0) return -1;
+  int32_t ret = 0;
+  tEncodeSize(tEncodeSSchemaWrapper, pSW, vLen, ret);
+  if (ret < 0) return -1;
   pVal = taosMemoryMalloc(vLen);
   if (pVal == NULL) {
     rcode = -1;

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

@@ -454,7 +454,7 @@ void tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond* pCond) {
   if (emptyQueryTimewindow(pTsdbReadHandle)) {
     if (pCond->order != pTsdbReadHandle->order) {
       pTsdbReadHandle->order = pCond->order;
-      TSWAP(pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey, int64_t);
+      TSWAP(pTsdbReadHandle->window.skey, pTsdbReadHandle->window.ekey);
     }
 
     return;
@@ -924,7 +924,7 @@ static bool hasMoreDataInCache(STsdbReadHandle* pHandle) {
   pHandle->cur.mixBlock = true;
 
   if (!ASCENDING_TRAVERSE(pHandle->order)) {
-    TSWAP(win->skey, win->ekey, TSKEY);
+    TSWAP(win->skey, win->ekey);
   }
 
   return true;
@@ -1203,7 +1203,7 @@ static int32_t handleDataMergeIfNeeded(STsdbReadHandle* pTsdbReadHandle, SBlock*
       // update the last key value
       pCheckInfo->lastKey = cur->win.ekey + step;
       if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
-        TSWAP(cur->win.skey, cur->win.ekey, TSKEY);
+        TSWAP(cur->win.skey, cur->win.ekey);
       }
 
       cur->mixBlock = true;
@@ -1701,7 +1701,7 @@ static void copyAllRemainRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, STa
   int32_t end = endPos;
 
   if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
-    TSWAP(start, end, int32_t);
+    TSWAP(start, end);
   }
 
   assert(pTsdbReadHandle->outputCapacity >= (end - start + 1));
@@ -1932,7 +1932,7 @@ static void doMergeTwoLevelData(STsdbReadHandle* pTsdbReadHandle, STableCheckInf
        ((pos < endPos || cur->lastKey < pTsdbReadHandle->window.ekey) && !ASCENDING_TRAVERSE(pTsdbReadHandle->order)));
 
   if (!ASCENDING_TRAVERSE(pTsdbReadHandle->order)) {
-    TSWAP(cur->win.skey, cur->win.ekey, TSKEY);
+    TSWAP(cur->win.skey, cur->win.ekey);
   }
 
   moveDataToFront(pTsdbReadHandle, numOfRows, numOfCols);

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

@@ -369,7 +369,8 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
   tCoderClear(&coder);
 
   // prepare rsp
-  tEncodeSize(tEncodeSVCreateTbBatchRsp, &rsp, pRsp->contLen);
+  int32_t ret = 0;
+  tEncodeSize(tEncodeSVCreateTbBatchRsp, &rsp, pRsp->contLen, ret);
   pRsp->pCont = rpcMallocCont(pRsp->contLen);
   if (pRsp->pCont == NULL) {
     terrno = TSDB_CODE_OUT_OF_MEMORY;

source/libs/executor/src/executorimpl.c

@@ -2100,7 +2100,7 @@ static int32_t updateBlockLoadStatus(STaskAttr* pQuery, int32_t status) {
 //
 //     pQueryAttr->order.order = TSDB_ORDER_ASC;
 //     if (pQueryAttr->window.skey > pQueryAttr->window.ekey) {
-//       TSWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey, TSKEY);
+//       TSWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey);
 //     }
 //
 //     pQueryAttr->needReverseScan = false;
@@ -2110,7 +2110,7 @@ static int32_t updateBlockLoadStatus(STaskAttr* pQuery, int32_t status) {
 //   if (pQueryAttr->groupbyColumn && pQueryAttr->order.order == TSDB_ORDER_DESC) {
 //     pQueryAttr->order.order = TSDB_ORDER_ASC;
 //     if (pQueryAttr->window.skey > pQueryAttr->window.ekey) {
-//       TSWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey, TSKEY);
+//       TSWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey);
 //     }
 //
 //     pQueryAttr->needReverseScan = false;
@@ -2135,7 +2135,7 @@ static int32_t updateBlockLoadStatus(STaskAttr* pQuery, int32_t status) {
 //         //qDebug(msg, pQInfo->qId, "only-first", pQueryAttr->order.order, TSDB_ORDER_ASC, pQueryAttr->window.skey,
 ////               pQueryAttr->window.ekey, pQueryAttr->window.ekey, pQueryAttr->window.skey);
 //
-//        TSWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey, TSKEY);
+//        TSWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey);
 //        doUpdateLastKey(pQueryAttr);
 //      }
 //
@@ -2146,7 +2146,7 @@ static int32_t updateBlockLoadStatus(STaskAttr* pQuery, int32_t status) {
 //        //qDebug(msg, pQInfo->qId, "only-last", pQueryAttr->order.order, TSDB_ORDER_DESC, pQueryAttr->window.skey,
 ////               pQueryAttr->window.ekey, pQueryAttr->window.ekey, pQueryAttr->window.skey);
 //
-//        TSWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey, TSKEY);
+//        TSWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey);
 //        doUpdateLastKey(pQueryAttr);
 //      }
 //
@@ -2162,7 +2162,7 @@ static int32_t updateBlockLoadStatus(STaskAttr* pQuery, int32_t status) {
 ////                 pQueryAttr->window.skey, pQueryAttr->window.ekey, pQueryAttr->window.ekey,
 /// pQueryAttr->window.skey);
 //
-//          TSWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey, TSKEY);
+//          TSWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey);
 //          doUpdateLastKey(pQueryAttr);
 //        }
 //
@@ -2174,7 +2174,7 @@ static int32_t updateBlockLoadStatus(STaskAttr* pQuery, int32_t status) {
 ////                 pQueryAttr->window.skey, pQueryAttr->window.ekey, pQueryAttr->window.ekey,
 /// pQueryAttr->window.skey);
 //
-//          TSWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey, TSKEY);
+//          TSWAP(pQueryAttr->window.skey, pQueryAttr->window.ekey);
 //          doUpdateLastKey(pQueryAttr);
 //        }
 //
@@ -2673,7 +2673,7 @@ static void updateTableQueryInfoForReverseScan(STableQueryInfo* pTableQueryInfo)
     return;
   }
 
-  //  TSWAP(pTableQueryInfo->win.skey, pTableQueryInfo->win.ekey, TSKEY);
+  //  TSWAP(pTableQueryInfo->win.skey, pTableQueryInfo->win.ekey);
   //  pTableQueryInfo->lastKey = pTableQueryInfo->win.skey;
 
   //  SWITCH_ORDER(pTableQueryInfo->cur.order);
@@ -6660,7 +6660,7 @@ static  int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableS
   //todo work around a problem, remove it later
   if ((pCond->order == TSDB_ORDER_ASC && pCond->twindow.skey > pCond->twindow.ekey) ||
       (pCond->order == TSDB_ORDER_DESC && pCond->twindow.skey < pCond->twindow.ekey)) {
-    TSWAP(pCond->twindow.skey, pCond->twindow.ekey, int64_t);
+    TSWAP(pCond->twindow.skey, pCond->twindow.ekey);
   }
 #endif
 

source/libs/executor/src/scanoperator.c

@@ -266,7 +266,7 @@ static void prepareForDescendingScan(STableScanInfo* pTableScanInfo, SqlFunction
   //  setupQueryRangeForReverseScan(pTableScanInfo);
 
   STimeWindow* pTWindow = &pTableScanInfo->cond.twindow;
-  TSWAP(pTWindow->skey, pTWindow->ekey, int64_t);
+  TSWAP(pTWindow->skey, pTWindow->ekey);
   pTableScanInfo->cond.order = TSDB_ORDER_DESC;
 }
 

source/libs/parser/src/parInsert.c

@@ -307,7 +307,7 @@ static int32_t buildOutput(SInsertParseContext* pCxt) {
     taosHashGetDup(pCxt->pVgroupsHashObj, (const char*)&src->vgId, sizeof(src->vgId), &dst->vg);
     dst->numOfTables = src->numOfTables;
     dst->size = src->size;
-    TSWAP(dst->pData, src->pData, char*);
+    TSWAP(dst->pData, src->pData);
     buildMsgHeader(src, dst);
     taosArrayPush(pCxt->pOutput->pDataBlocks, &dst);
   }

source/libs/parser/src/parInsertData.c

@@ -159,9 +159,10 @@ static int32_t createDataBlock(size_t defaultSize, int32_t rowSize, int32_t star
 int32_t buildCreateTbMsg(STableDataBlocks* pBlocks, SVCreateTbReq* pCreateTbReq) {
   SCoder coder = {0};
   char* pBuf;
-   int32_t len;
+  int32_t len;
 
-  tEncodeSize(tEncodeSVCreateTbReq, pCreateTbReq, len);
+  int32_t ret = 0;
+  tEncodeSize(tEncodeSVCreateTbReq, pCreateTbReq, len, ret);
   if (pBlocks->nAllocSize - pBlocks->size < len) {
     pBlocks->nAllocSize += len + pBlocks->rowSize;
     char* pTmp = taosMemoryRealloc(pBlocks->pData, pBlocks->nAllocSize);

source/libs/parser/src/parTranslater.c

@@ -2049,10 +2049,10 @@ static int32_t buildSampleAst(STranslateContext* pCxt, SSampleAstInfo* pInfo, ch
   }
   strcpy(pTable->table.dbName, pInfo->pDbName);
   strcpy(pTable->table.tableName, pInfo->pTableName);
-  TSWAP(pTable->pMeta, pInfo->pRollupTableMeta, STableMeta*);
+  TSWAP(pTable->pMeta, pInfo->pRollupTableMeta);
   pSelect->pFromTable = (SNode*)pTable;
 
-  TSWAP(pSelect->pProjectionList, pInfo->pFuncs, SNodeList*);
+  TSWAP(pSelect->pProjectionList, pInfo->pFuncs);
   SFunctionNode* pFunc = nodesMakeNode(QUERY_NODE_FUNCTION);
   if (NULL == pSelect->pProjectionList || NULL == pFunc) {
     nodesDestroyNode(pSelect);
@@ -2069,9 +2069,9 @@ static int32_t buildSampleAst(STranslateContext* pCxt, SSampleAstInfo* pInfo, ch
     return TSDB_CODE_OUT_OF_MEMORY;
   }
   pSelect->pWindow = (SNode*)pInterval;
-  TSWAP(pInterval->pInterval, pInfo->pInterval, SNode*);
-  TSWAP(pInterval->pOffset, pInfo->pOffset, SNode*);
-  TSWAP(pInterval->pSliding, pInfo->pSliding, SNode*);
+  TSWAP(pInterval->pInterval, pInfo->pInterval);
+  TSWAP(pInterval->pOffset, pInfo->pOffset);
+  TSWAP(pInterval->pSliding, pInfo->pSliding);
   pInterval->pCol = nodesMakeNode(QUERY_NODE_COLUMN);
   if (NULL == pInterval->pCol) {
     nodesDestroyNode(pSelect);
@@ -3282,7 +3282,8 @@ static int32_t serializeVgroupTablesBatch(SVgroupTablesBatch* pTbBatch, SArray*
   int    tlen;
   SCoder coder = {0};
 
-  tEncodeSize(tEncodeSVCreateTbBatchReq, &pTbBatch->req, tlen);
+  int32_t ret = 0;
+  tEncodeSize(tEncodeSVCreateTbBatchReq, &pTbBatch->req, tlen, ret);
   tlen += sizeof(SMsgHead);  //+ tSerializeSVCreateTbBatchReq(NULL, &(pTbBatch->req));
   void* buf = taosMemoryMalloc(tlen);
   if (NULL == buf) {
@@ -3696,7 +3697,7 @@ static int32_t setQuery(STranslateContext* pCxt, SQuery* pQuery) {
     default:
       pQuery->execMode = QUERY_EXEC_MODE_RPC;
       if (NULL != pCxt->pCmdMsg) {
-        TSWAP(pQuery->pCmdMsg, pCxt->pCmdMsg, SCmdMsgInfo*);
+        TSWAP(pQuery->pCmdMsg, pCxt->pCmdMsg);
         pQuery->msgType = pQuery->pCmdMsg->msgType;
       }
       break;

source/libs/planner/src/planLogicCreater.c

@@ -251,8 +251,8 @@ static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
     return TSDB_CODE_OUT_OF_MEMORY;
   }
 
-  TSWAP(pScan->pMeta, pRealTable->pMeta, STableMeta*);
-  TSWAP(pScan->pVgroupList, pRealTable->pVgroupList, SVgroupsInfo*);
+  TSWAP(pScan->pMeta, pRealTable->pMeta);
+  TSWAP(pScan->pVgroupList, pRealTable->pVgroupList);
   pScan->scanSeq[0] = 1;
   pScan->scanSeq[1] = 0;
   pScan->scanRange = TSWINDOW_INITIALIZER;
@@ -954,7 +954,7 @@ static int32_t createVnodeModifLogicNode(SLogicPlanContext* pCxt, SVnodeModifOpS
   if (NULL == pModif) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
-  TSWAP(pModif->pDataBlocks, pStmt->pDataBlocks, SArray*);
+  TSWAP(pModif->pDataBlocks, pStmt->pDataBlocks);
   pModif->msgType = getMsgType(pStmt->sqlNodeType);
   *pLogicNode = (SLogicNode*)pModif;
   return TSDB_CODE_SUCCESS;

source/libs/planner/src/planOptimizer.c

@@ -246,7 +246,7 @@ static int32_t cpdMergeConds(SNode** pDst, SNodeList** pSrc) {
 
 static int32_t cpdCondAppend(SNode** pCond, SNode** pAdditionalCond) {
   if (NULL == *pCond) {
-    TSWAP(*pCond, *pAdditionalCond, SNode*);
+    TSWAP(*pCond, *pAdditionalCond);
     return TSDB_CODE_SUCCESS;
   }
 

source/libs/planner/src/planPhysiCreater.c

@@ -1097,7 +1097,7 @@ static int32_t createDataInserter(SPhysiPlanContext* pCxt, SVgDataBlocks* pBlock
 
   pInserter->numOfTables = pBlocks->numOfTables;
   pInserter->size = pBlocks->size;
-  TSWAP(pInserter->pData, pBlocks->pData, char*);
+  TSWAP(pInserter->pData, pBlocks->pData);
 
   *pSink = (SDataSinkNode*)pInserter;
   return TSDB_CODE_SUCCESS;

source/libs/planner/src/planSpliter.c

@@ -65,7 +65,7 @@ static SLogicSubplan* splCreateScanSubplan(SSplitContext* pCxt, SScanLogicNode*
   pSubplan->id.groupId = pCxt->groupId;
   pSubplan->subplanType = SUBPLAN_TYPE_SCAN;
   pSubplan->pNode = (SLogicNode*)nodesCloneNode(pScan);
-  TSWAP(pSubplan->pVgroupList, ((SScanLogicNode*)pSubplan->pNode)->pVgroupList, SVgroupsInfo*);
+  TSWAP(pSubplan->pVgroupList, ((SScanLogicNode*)pSubplan->pNode)->pVgroupList);
   SPLIT_FLAG_SET_MASK(pSubplan->splitFlag, flag);
   return pSubplan;
 }
@@ -406,8 +406,7 @@ int32_t splitLogicPlan(SPlanContext* pCxt, SLogicNode* pLogicNode, SLogicSubplan
   }
   if (QUERY_NODE_LOGIC_PLAN_VNODE_MODIF == nodeType(pLogicNode)) {
     pSubplan->subplanType = SUBPLAN_TYPE_MODIFY;
-    TSWAP(((SVnodeModifLogicNode*)pLogicNode)->pDataBlocks, ((SVnodeModifLogicNode*)pSubplan->pNode)->pDataBlocks,
-          SArray*);
+    TSWAP(((SVnodeModifLogicNode*)pLogicNode)->pDataBlocks, ((SVnodeModifLogicNode*)pSubplan->pNode)->pDataBlocks);
   } else {
     pSubplan->subplanType = SUBPLAN_TYPE_SCAN;
   }

source/libs/transport/inc/transComm.h

@@ -21,6 +21,7 @@ extern "C" {
 #include <uv.h>
 #include "lz4.h"
 #include "os.h"
+#include "osSocket.h"
 #include "rpcCache.h"
 #include "rpcHead.h"
 #include "rpcLog.h"
@@ -105,6 +106,7 @@ typedef void* queue[2];
 
 #define TRANS_RETRY_COUNT_LIMIT 10  // retry count limit
 #define TRANS_RETRY_INTERVAL    5   // ms retry interval
+#define TRANS_CONN_TIMEOUT      3   // connect timeout
 
 typedef struct {
   SRpcInfo* pRpc;     // associated SRpcInfo

source/libs/transport/src/trans.c

@@ -49,6 +49,10 @@ void* rpcOpen(const SRpcInit* pInit) {
   pRpc->connType = pInit->connType;
   pRpc->idleTime = pInit->idleTime;
   pRpc->tcphandle = (*taosInitHandle[pRpc->connType])(0, pInit->localPort, pRpc->label, pRpc->numOfThreads, NULL, pRpc);
+  if (pRpc->tcphandle == NULL) {
+    taosMemoryFree(pRpc);
+    return NULL;
+  }
   pRpc->parent = pInit->parent;
   if (pInit->user) {
     memcpy(pRpc->user, pInit->user, strlen(pInit->user));

source/libs/transport/src/transCli.c

@@ -103,6 +103,10 @@ static void      cliDestroyConn(SCliConn* pConn, bool clear /*clear tcp handle o
 static void      cliDestroy(uv_handle_t* handle);
 static void      cliSend(SCliConn* pConn);
 
+/*
+ * set TCP connection timeout per-socket level
+ */
+static int cliCreateSocket();
 // process data read from server, add decompress etc later
 static void cliHandleResp(SCliConn* conn);
 // handle except about conn
@@ -729,9 +733,15 @@ void cliHandleReq(SCliMsg* pMsg, SCliThrdObj* pThrd) {
     if (ret) {
       tError("%s cli conn %p failed to set conn option, errmsg %s", pTransInst->label, conn, uv_err_name(ret));
     }
+    int fd = taosCreateSocketWithTimeOutOpt(TRANS_CONN_TIMEOUT);
+    if (fd == -1) {
+      tTrace("%s cli conn %p failed to create socket", pTransInst->label, conn);
+      cliHandleExcept(conn);
+      return;
+    }
+    uv_tcp_open((uv_tcp_t*)conn->stream, fd);
 
     struct sockaddr_in addr;
-
     addr.sin_family = AF_INET;
     addr.sin_addr.s_addr = taosGetIpv4FromFqdn(conn->ip);
     addr.sin_port = (uint16_t)htons((uint16_t)conn->port);
@@ -902,7 +912,6 @@ int cliAppCb(SCliConn* pConn, STransMsg* pResp, SCliMsg* pMsg) {
         cliDestroy((uv_handle_t*)pConn->stream);
         return -1;
       }
-
     } else if (pCtx->retryCount < TRANS_RETRY_COUNT_LIMIT) {
       if (pResp->contLen == 0) {
         pEpSet->inUse = (pEpSet->inUse++) % pEpSet->numOfEps;

source/libs/transport/src/transSrv.c

@@ -93,6 +93,8 @@ typedef struct SServerObj {
   uint32_t    ip;
   uint32_t    port;
   uv_async_t* pAcceptAsync;  // just to quit from from accept thread
+
+  bool inited;
 } SServerObj;
 
 // handle
@@ -143,7 +145,7 @@ static void (*transAsyncHandle[])(SSrvMsg* msg, SWorkThrdObj* thrd) = {uvHandleR
 
 static int32_t exHandlesMgt;
 
-void       uvInitExHandleMgt();
+void       uvInitEnv();
 void       uvOpenExHandleMgt(int size);
 void       uvCloseExHandleMgt();
 int64_t    uvAddExHandle(void* p);
@@ -716,6 +718,7 @@ static bool addHandleToAcceptloop(void* arg) {
   }
   if ((err = uv_listen((uv_stream_t*)&srv->server, 512, uvOnAcceptCb)) != 0) {
     tError("failed to listen: %s", uv_err_name(err));
+    terrno = TSDB_CODE_RPC_PORT_EADDRINUSE;
     return false;
   }
   return true;
@@ -800,7 +803,7 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
   srv->port = port;
   uv_loop_init(srv->loop);
 
-  taosThreadOnce(&transModuleInit, uvInitExHandleMgt);
+  taosThreadOnce(&transModuleInit, uvInitEnv);
   transSrvInst++;
 
   for (int i = 0; i < srv->numOfThreads; i++) {
@@ -844,15 +847,15 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads,
     goto End;
     // clear all resource later
   }
-
+  srv->inited = true;
   return srv;
 End:
   transCloseServer(srv);
   return NULL;
 }
 
-void uvInitExHandleMgt() {
-  // init exhandle mgt
+void uvInitEnv() {
+  uv_os_setenv("UV_TCP_SINGLE_ACCEPT", "1");
   uvOpenExHandleMgt(10000);
 }
 void uvOpenExHandleMgt(int size) {
@@ -958,9 +961,10 @@ void transCloseServer(void* arg) {
   SServerObj* srv = arg;
 
   tDebug("send quit msg to accept thread");
-  uv_async_send(srv->pAcceptAsync);
-  taosThreadJoin(srv->thread, NULL);
-
+  if (srv->inited) {
+    uv_async_send(srv->pAcceptAsync);
+    taosThreadJoin(srv->thread, NULL);
+  }
   SRV_RELEASE_UV(srv->loop);
 
   for (int i = 0; i < srv->numOfThreads; i++) {

source/os/src/osSemaphore.c

@@ -67,6 +67,12 @@ int32_t tsem_wait(tsem_t* sem) {
   return ret;
 }
 
+int32_t tsem_timewait(tsem_t* sem, int64_t nanosecs) {
+  int ret = 0;
+
+  return ret;
+}
+
 #elif defined(_TD_DARWIN_64)
 
 /*

source/os/src/osSocket.c

@@ -37,31 +37,35 @@
 #include <unistd.h>
 
 #if defined(DARWIN)
-  #include <dispatch/dispatch.h>
-  #include "osEok.h"
+#include <dispatch/dispatch.h>
+#include "osEok.h"
 #else
-  #include <sys/epoll.h>
+#include <sys/epoll.h>
 #endif
 #endif
 
+#ifndef INVALID_SOCKET
+#define INVALID_SOCKET -1
+#endif
+
 typedef struct TdSocketServer {
 #if SOCKET_WITH_LOCK
   TdThreadRwlock rwlock;
 #endif
   int      refId;
   SocketFd fd;
-} *TdSocketServerPtr, TdSocketServer;
+} * TdSocketServerPtr, TdSocketServer;
 
 typedef struct TdEpoll {
 #if SOCKET_WITH_LOCK
   TdThreadRwlock rwlock;
 #endif
-  int      refId;
-  EpollFd  fd;
-} *TdEpollPtr, TdEpoll;
+  int     refId;
+  EpollFd fd;
+} * TdEpollPtr, TdEpoll;
 
 int32_t taosSendto(TdSocketPtr pSocket, void *buf, int len, unsigned int flags, const struct sockaddr *dest_addr,
-                    int addrlen) {
+                   int addrlen) {
   if (pSocket == NULL || pSocket->fd < 0) {
     return -1;
   }
@@ -94,7 +98,8 @@ int32_t taosReadSocket(TdSocketPtr pSocket, void *buf, int len) {
 #endif
 }
 
-int32_t taosReadFromSocket(TdSocketPtr pSocket, void *buf, int32_t len, int32_t flags, struct sockaddr *destAddr, int *addrLen) {
+int32_t taosReadFromSocket(TdSocketPtr pSocket, void *buf, int32_t len, int32_t flags, struct sockaddr *destAddr,
+                           int *addrLen) {
   if (pSocket == NULL || pSocket->fd < 0) {
     return -1;
   }
@@ -318,7 +323,7 @@ int32_t taosWriteMsg(TdSocketPtr pSocket, void *buf, int32_t nbytes) {
     return -1;
   }
   int32_t nleft, nwritten;
-  char   *ptr = (char *)buf;
+  char *  ptr = (char *)buf;
 
   nleft = nbytes;
 
@@ -347,7 +352,7 @@ int32_t taosReadMsg(TdSocketPtr pSocket, void *buf, int32_t nbytes) {
     return -1;
   }
   int32_t nleft, nread;
-  char   *ptr = (char *)buf;
+  char *  ptr = (char *)buf;
 
   nleft = nbytes;
 
@@ -689,8 +694,7 @@ TdSocketServerPtr taosOpenTcpServerSocket(uint32_t ip, uint16_t port) {
   return (TdSocketServerPtr)pSocket;
 }
 
-TdSocketPtr taosAcceptTcpConnectSocket(TdSocketServerPtr pServerSocket, struct sockaddr *destAddr,
-                                       int *addrLen) {
+TdSocketPtr taosAcceptTcpConnectSocket(TdSocketServerPtr pServerSocket, struct sockaddr *destAddr, int *addrLen) {
   if (pServerSocket == NULL || pServerSocket->fd < 0) {
     return NULL;
   }
@@ -771,7 +775,7 @@ uint32_t taosGetIpv4FromFqdn(const char *fqdn) {
 
   int32_t ret = getaddrinfo(fqdn, NULL, &hints, &result);
   if (result) {
-    struct sockaddr    *sa = result->ai_addr;
+    struct sockaddr *   sa = result->ai_addr;
     struct sockaddr_in *si = (struct sockaddr_in *)sa;
     struct in_addr      ia = si->sin_addr;
     uint32_t            ip = ia.s_addr;
@@ -887,7 +891,6 @@ int32_t taosGetSocketName(TdSocketPtr pSocket, struct sockaddr *destAddr, int *a
   return getsockname(pSocket->fd, destAddr, addrLen);
 }
 
-
 TdEpollPtr taosCreateEpoll(int32_t size) {
   EpollFd fd = -1;
 #ifdef WINDOWS
@@ -939,3 +942,28 @@ int32_t taosCloseEpoll(TdEpollPtr *ppEpoll) {
   taosMemoryFree(*ppEpoll);
   return code;
 }
+/*
+ * Set TCP connection timeout per-socket level.
+ * ref [https://github.com/libuv/help/issues/54]
+ */
+int taosCreateSocketWithTimeOutOpt(uint32_t conn_timeout_sec) {
+#if defined(WINDOWS)
+  SOCKET fd;
+#else
+  int      fd;
+#endif
+  if ((fd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP)) == INVALID_SOCKET) {
+    return -1;
+  }
+#if defined(WINDOWS)
+  if (0 != setsockopt(fd, IPPROTO_TCP, TCP_MAXRT, (char *)&conn_timeout_sec, sizeof(conn_timeout_sec))) {
+    return -1;
+  }
+#else  // Linux like systems
+  uint32_t conn_timeout_ms = conn_timeout_sec * 1000;
+  if (0 != setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, (char *)&conn_timeout_ms, sizeof(conn_timeout_ms))) {
+    return -1;
+  }
+#endif
+  return (int)fd;
+}

source/util/src/tarray.c

@@ -482,3 +482,22 @@ void taosArraySortPWithExt(SArray* pArray, __ext_compar_fn_t fn, const void* par
   taosArrayGetSize(pArray) > 8 ? taosArrayQuickSort(pArray, fn, param) : taosArrayInsertSort(pArray, fn, param);
 }
 // TODO(yihaoDeng) add order array<type>
+//
+
+char* taosShowStrArray(const SArray* pArray) {
+  int32_t sz = pArray->size;
+  int32_t tlen = 0;
+  for (int32_t i = 0; i < sz; i++) {
+    tlen += strlen(taosArrayGetP(pArray, i)) + 1;
+  }
+  char* res = taosMemoryCalloc(1, tlen);
+  char* buf = res;
+  for (int32_t i = 0; i < sz; i++) {
+    char*   str = taosArrayGetP(pArray, i);
+    int32_t len = strlen(str);
+    memcpy(buf, str, len);
+    buf += len;
+    if (i != sz - 1) *buf = ',';
+  }
+  return res;
+}

source/util/src/terror.c

@@ -68,6 +68,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_RPC_INVALID_TIME_STAMP,       "Client and server's t
 TAOS_DEFINE_ERROR(TSDB_CODE_APP_NOT_READY,                "Database not ready")
 TAOS_DEFINE_ERROR(TSDB_CODE_RPC_FQDN_ERROR,               "Unable to resolve FQDN")
 TAOS_DEFINE_ERROR(TSDB_CODE_RPC_INVALID_VERSION,          "Invalid app version")
+TAOS_DEFINE_ERROR(TSDB_CODE_RPC_PORT_EADDRINUSE,           "port already in use")
 
 //common & util
 TAOS_DEFINE_ERROR(TSDB_CODE_OUT_OF_MEMORY,                "Out of Memory")
@@ -271,9 +272,14 @@ TAOS_DEFINE_ERROR(TSDB_CODE_MND_TRANS_INVALID_STAGE,      "Invalid stage to kill
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_TRANS_CANT_PARALLEL,      "Invalid stage to kill")
 
 // mnode-mq
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_UNSUPPORTED_TOPIC,        "Topic with aggregation is unsupported")
-TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_READY,       "Consumer waiting for rebalance")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_ALREADY_EXIST,      "Topic already exists")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOPIC_NOT_EXIST,          "Topic not exist")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_TOO_MANY_TOPICS,          "Too many Topics")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TOPIC,            "Invalid topic")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TOPIC_QUERY,      "Topic with invalid query")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_INVALID_TOPIC_OPTION,     "Topic with invalid option")
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_EXIST,       "Consumer not exist")
+TAOS_DEFINE_ERROR(TSDB_CODE_MND_CONSUMER_NOT_READY,       "Consumer waiting for rebalance")
 
 // mnode-sma
 TAOS_DEFINE_ERROR(TSDB_CODE_MND_SMA_ALREADY_EXIST,        "SMA already exists")

source/util/test/encodeTest.cpp

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

tests/script/general/user/authority.sim

-system sh/stop_dnodes.sh
-system sh/deploy.sh -n dnode1 -i 1
-system sh/exec.sh -n dnode1 -s start
-sql connect
-sleep 2000
-
-print ============= step1
-sql create user read pass 'taosdata'
-sql create user write pass 'taosdata'
-sql show users
-if $rows != 5 then
-  return -1
-endi
-
-print ============= step2
-sql close
-sql connect read
-sleep 2000
-
-sql create database dread
-sql show databases
-if $rows != 1 then
-  return -1
-endi
-
-print ============= step3
-sql close
-sql connect write
-sleep 2000
-
-sql create database dwrite
-sql show databases
-if $rows != 2 then
-  return -1
-endi
-
-print ============ step4
-sql close
-sql connect
-sleep 2000
-
-sql show databases
-if $row != 2 then
-  return -1
-endi
-
-print ============ step5
-sql close
-sql connect read
-sleep 2000
-sql drop database dread
-sql drop database dwrite
-
-
-sql close
-sql connect
-sql show databases
-if $rows != 0 then
-  return -1
-endi
-
-sql close
-sql connect
-sleep 2000
-
-system sh/exec.sh -n dnode1 -s stop  -x SIGINT
\ No newline at end of file

tests/script/general/user/user_create.sim

-system sh/stop_dnodes.sh
-
-system sh/deploy.sh -n dnode1 -i 1
-system sh/cfg.sh -n dnode1 -c wallevel -v 0
-system sh/exec.sh -n dnode1 -s start
-sql connect
-
-print =============== step1
-sql show users
-if $rows != 3 then 
-  return -1
-endi
-
-sql create user read PASS 'pass123'
-sql create user read PASS 'pass123' -x step1
-	return -1
-step1:
-
-sql show users
-if $rows != 4 then 
-  return -1
-endi
-
-sql alter user read PASS 'taosdata'
-
-print =============== step2
-sql close
-sql connect read
-sleep 2000
-
-sql alter user read PASS 'taosdata'
-
-print =============== step3
-sql drop user read -x step31
-  return -1
-step31:
-sql drop user _root -x step32
-  return -1
-step32:
-sql drop user monitor -x step33
-  return -1
-step33:
-
-print =============== step4
-sql close
-sql connect
-sleep 2000
-
-sql alter user read privilege read
-sql show users
-print $data1_read
-if $data1_read != readable then 
-  return -1
-endi
-
-sql_error alter user read privilege super
-sql show users
-print $data1_read
-if $data1_read != readable then
-  return -1
-endi
-
-sql alter user read privilege write
-sql show users
-print $data1_read
-if $data1_read != writable then 
-  return -1
-endi
-
-sql alter user read privilege 1 -x step43
-  return -1
-step43:
-
-sql drop user _root -x step41
-  return -1
-step41:
-
-sql drop user monitor -x step42
-  return -1
-step42:
-
-sql drop user read
-
-system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file

tests/script/jenkins/basic.txt

@@ -3,6 +3,9 @@
 
 # ---- user
 ./test.sh -f tsim/user/basic1.sim
+./test.sh -f tsim/user/pass_alter.sim
+./test.sh -f tsim/user/pass_len.sim
+./test.sh -f tsim/user/user_len.sim
 
 # ---- db
 ./test.sh -f tsim/db/create_all_options.sim

tests/script/general/user/pass_alter.sim → tests/script/tsim/user/pass_alter.sim

 system sh/stop_dnodes.sh
 system sh/deploy.sh -n dnode1 -i 1
-system sh/cfg.sh -n dnode1 -c wallevel -v 0
 system sh/exec.sh -n dnode1 -s start
-sleep 2000
 sql connect
 
 print ============= step1
@@ -13,7 +11,7 @@ sql alter user read pass 'taosdata'
 sql alter user write pass 'taosdata'
 
 sql show users
-if $rows != 5 then 
+if $rows != 3 then 
   return -1
 endi
 
@@ -27,11 +25,11 @@ sql alter user write pass 'taosdata1' -x step2
   return -1
 step2:
 
-sql_error create user read pass 'taosdata1'
-sql_error create user write pass 'taosdata1'
+sql_error create user read1 pass 'taosdata1'
+sql_error create user write1 pass 'taosdata1'
 
 sql show users
-if $rows != 5 then 
+if $rows != 3 then 
   return -1
 endi
 
@@ -41,27 +39,27 @@ sleep 2500
 print user write login
 sql connect write
 
-sql_error create user read pass 'taosdata1'
-sql_error create user write pass 'taosdata1'
+sql_error create user read2 pass 'taosdata1'
+sql_error create user write2 pass 'taosdata1'
 sql alter user write pass 'taosdata'
 sql alter user read pass 'taosdata' -x step3
   return -1
 step3:
 
 sql show users
-if $rows != 5 then 
+if $rows != 3 then 
   return -1
 endi
 
 print ============= step4
 sql close 
 sleep 2500
-print root write login
+print user root login
 sql connect
 sql create user oroot pass 'taosdata'
 
 sql show users
-if $rows != 6 then 
+if $rows != 4 then 
   return -1
 endi
 

tests/script/general/user/pass_len.sim → tests/script/tsim/user/pass_len.sim

 system sh/stop_dnodes.sh
-
 system sh/deploy.sh -n dnode1 -i 1
-system sh/cfg.sh -n dnode1 -c wallevel -v 0
 system sh/exec.sh -n dnode1 -s start
 sql connect
 
@@ -50,15 +48,16 @@ step3:
 
 sql create user $user PASS 'abc0123456789'
 sql show users
-if $rows != 3 then 
+if $rows != 4 then 
   return -1
 endi
 
 print =============== step4
 $i = 3
 $user = $userPrefix . $i
-sql create user $user PASS 'abcd012345678901234567891234567890' -x step4
+sql create user $user PASS 'abcd012345678901234567891234567890abcd012345678901234567891234567890abcd012345678901234567891234567890abcd012345678901234567891234567890123' -x step4
 	return -1
+
 step4:
 sql show users
 if $rows != 4 then 

tests/script/general/user/user_len.sim → tests/script/tsim/user/user_len.sim

 system sh/stop_dnodes.sh
-
-
 system sh/deploy.sh -n dnode1 -i 1
-system sh/cfg.sh -n dnode1 -c wallevel -v 0
 system sh/exec.sh -n dnode1 -s start
-
-sleep 2000
 sql connect
 
 $i = 0
@@ -24,7 +19,7 @@ sql create user PASS '123' -x step1
 step1:
 
 sql show users
-if $rows != 3 then 
+if $rows != 1 then 
   return -1
 endi
 
@@ -33,13 +28,13 @@ sql drop user a -x step2
 step2:
 sql create user a PASS '123'
 sql show users
-if $rows != 4 then 
+if $rows != 2 then 
   return -1
 endi
 
 sql drop user a
 sql show users
-if $rows != 3 then 
+if $rows != 1 then 
   return -1
 endi
 
@@ -49,13 +44,13 @@ step3:
 
 sql create user abc01234567890123456789 PASS '123'
 sql show users
-if $rows != 4 then 
+if $rows != 2 then 
   return -1
 endi
 
 sql drop user abc01234567890123456789
 sql show users
-if $rows != 3 then 
+if $rows != 1 then 
   return -1
 endi
 
@@ -64,7 +59,7 @@ sql create user abcd0123456789012345678901234567890111 PASS '123' -x step4
 	return -1
 step4:
 sql show users
-if $rows != 3 then 
+if $rows != 1 then 
   return -1
 endi
 
@@ -77,13 +72,13 @@ step61:
 
 sql create user a123 PASS '123'
 sql show users
-if $rows != 4 then 
+if $rows != 2 then 
   return -1
 endi
 
 sql drop user a123
 sql show users
-if $rows != 3 then 
+if $rows != 1 then 
   return -1
 endi
 

tests/system-test/0-others/taosShell.py

@@ -11,7 +11,7 @@ from util.sql import *
 from util.cases import *
 from util.dnodes import *
 
-def taos_command (key, value, expectString, cfgDir, dbName, key1='', value1=''):
+def taos_command (key, value, expectString, cfgDir, sqlString='', key1='', value1=''):
     if len(key) == 0:
         tdLog.exit("taos test key is null!")
 
@@ -37,30 +37,53 @@ def taos_command (key, value, expectString, cfgDir, dbName, key1='', value1=''):
     child = pexpect.spawn(taosCmd, timeout=3)
     #output = child.readline()
     #print (output.decode())
-    i = child.expect([expectString, pexpect.TIMEOUT, pexpect.EOF], timeout=1)
+    if len(expectString) != 0:
+        i = child.expect([expectString, pexpect.TIMEOUT, pexpect.EOF], timeout=6)
+    else:
+        i = child.expect([pexpect.TIMEOUT, pexpect.EOF], timeout=6)
+
     retResult = child.before.decode()
     print(retResult)
     #print(child.after.decode())
     if i == 0:
         print ('taos login success! Here can run sql, taos> ')
-        if len(dbName) != 0:
-            child.sendline ('create database %s;'%(dbName))
+        if len(sqlString) != 0:
+            child.sendline (sqlString)
             w = child.expect(["Query OK", pexpect.TIMEOUT, pexpect.EOF], timeout=1)
             if w == 0:
                 return "TAOS_OK"
             else:
                 return "TAOS_FAIL"
         else:
-            return  "TAOS_OK"
+            if key == 'A' or key1 == 'A' or key == 'C' or key1 == 'C':
+                return "TAOS_OK", retResult
+            else:
+                return  "TAOS_OK"
     else:
-        if key == 'A' or key1 == 'A':
+        if key == 'A' or key1 == 'A' or key == 'C' or key1 == 'C':
             return "TAOS_OK", retResult
         else:
             return "TAOS_FAIL"
 
 class TDTestCase:
-
-    #updatecfgDict = {'serverPort': 7080, 'firstEp': 'localhost:7080'}
+    #updatecfgDict = {'clientCfg': {'serverPort': 7080, 'firstEp': 'trd02:7080', 'secondEp':'trd02:7080'},\
+    #                 'serverPort': 7080, 'firstEp': 'trd02:7080'}
+    hostname = socket.gethostname()
+    serverPort = '7080'
+    rpcDebugFlagVal = '143'
+    clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135'}
+    clientCfgDict["serverPort"]    = serverPort
+    clientCfgDict["firstEp"]       = hostname + ':' + serverPort
+    clientCfgDict["secondEp"]      = hostname + ':' + serverPort
+    clientCfgDict["rpcDebugFlag"]  = rpcDebugFlagVal
+
+    updatecfgDict = {'clientCfg': {}, 'serverPort': '', 'firstEp': '', 'secondEp':''}
+    updatecfgDict["clientCfg"]  = clientCfgDict
+    updatecfgDict["serverPort"] = serverPort
+    updatecfgDict["firstEp"]    = hostname + ':' + serverPort
+    updatecfgDict["secondEp"]   = hostname + ':' + serverPort
+
+    print ("===================: ", updatecfgDict)
 
     def init(self, conn, logSql):
         tdLog.debug(f"start to excute {__file__}")
@@ -87,8 +110,8 @@ class TDTestCase:
         # time.sleep(2)
         tdSql.query("create user testpy pass 'testpy'")
 
-        hostname = socket.gethostname()
-        tdLog.info ("hostname: %s" % hostname)
+        #hostname = socket.gethostname()
+        #tdLog.info ("hostname: %s" % hostname)
 
         buildPath = self.getBuildPath()
         if (buildPath == ""):
@@ -104,12 +127,14 @@ class TDTestCase:
         keyDict = {'h':'', 'P':'6030', 'p':'testpy', 'u':'testpy', 'a':'', 'A':'', 'c':'', 'C':'', 's':'', 'r':'', 'f':'', \
                    'k':'', 't':'', 'n':'', 'l':'1024', 'N':'100', 'V':'', 'd':'db', 'w':'30', '-help':'', '-usage':'', '?':''}
 
-        keyDict['h'] = hostname
+        keyDict['h'] = self.hostname
         keyDict['c'] = cfgPath
+        keyDict['P'] = self.serverPort
 
         tdLog.printNoPrefix("================================ parameter: -h")
         newDbName="dbh"
-        retCode = taos_command("h", keyDict['h'], "taos>", keyDict['c'], newDbName)
+        sqlString = 'create database ' + newDbName + ';'
+        retCode = taos_command("h", keyDict['h'], "taos>", keyDict['c'], sqlString)
         if retCode != "TAOS_OK":
             tdLog.exit("taos -h %s fail"%keyDict['h'])
         else:
@@ -131,7 +156,8 @@ class TDTestCase:
         #sleep(3)
         #keyDict['P'] = 6030
         newDbName = "dbpp"
-        retCode = taos_command("P", keyDict['P'], "taos>", keyDict['c'], newDbName)
+        sqlString = 'create database ' + newDbName + ';'
+        retCode = taos_command("P", keyDict['P'], "taos>", keyDict['c'], sqlString)
         if retCode != "TAOS_OK":
             tdLog.exit("taos -P %s fail"%keyDict['P'])
         else:
@@ -146,7 +172,8 @@ class TDTestCase:
 
         tdLog.printNoPrefix("================================ parameter: -u")
         newDbName="dbu"
-        retCode = taos_command("u", keyDict['u'], "taos>", keyDict['c'], newDbName, "p", keyDict['p'])
+        sqlString = 'create database ' + newDbName + ';'
+        retCode = taos_command("u", keyDict['u'], "taos>", keyDict['c'], sqlString, "p", keyDict['p'])
         if retCode != "TAOS_OK":
             tdLog.exit("taos -u %s -p%s fail"%(keyDict['u'], keyDict['p']))
         else:
@@ -164,8 +191,9 @@ class TDTestCase:
         retCode, retVal = taos_command("p", keyDict['p'], "taos>", keyDict['c'], '', "A", '')
         if retCode != "TAOS_OK":
             tdLog.exit("taos -A fail")
-        
-        retCode = taos_command("u", keyDict['u'], "taos>", keyDict['c'], newDbName, 'a', retVal)
+                
+        sqlString = 'create database ' + newDbName + ';'
+        retCode = taos_command("u", keyDict['u'], "taos>", keyDict['c'], sqlString, 'a', retVal)
         if retCode != "TAOS_OK":
             tdLog.exit("taos -u %s -a %s"%(keyDict['u'], retVal))
 
@@ -220,10 +248,91 @@ class TDTestCase:
         tdSql.checkData(0, 1, 11)
         tdSql.checkData(1, 0, '2021-04-01 08:00:01.000')
         tdSql.checkData(1, 1, 21)
+        
+        keyDict['s'] = "\"select * from " + newDbName + ".ctb0\""
+        retCode = taos_command("s", keyDict['s'], "2021-04-01 08:00:01.000", keyDict['c'], '', '', '')
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -r show fail")
+        
+        tdLog.printNoPrefix("================================ parameter: -r")
+        keyDict['s'] = "\"select * from " + newDbName + ".ctb0\""
+        retCode = taos_command("s", keyDict['s'], "1617235200000", keyDict['c'], '', 'r', '')
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -r show fail")
 
-        #tdSql.query('drop database %s'%newDbName)
+        keyDict['s'] = "\"select * from " + newDbName + ".ctb1\""
+        retCode = taos_command("s", keyDict['s'], "1617235201000", keyDict['c'], '', 'r', '')
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -r show fail")
+        
+        tdSql.query('drop database %s'%newDbName)
+ 
+        tdLog.printNoPrefix("================================ parameter: -f")
+        pwd=os.getcwd()
+        newDbName="dbf"
+        sqlFile = pwd + "/0-others/sql.txt"
+        sql1 = "echo 'create database " + newDbName + "' > " + sqlFile
+        sql2 = "echo 'use " + newDbName + "' >> " + sqlFile
+        sql3 = "echo 'create table ntbf (ts timestamp, c binary(40))' >> " + sqlFile
+        sql4 = "echo 'insert into ntbf values (\"2021-04-01 08:00:00.000\", \"test taos -f1\")(\"2021-04-01 08:00:01.000\", \"test taos -f2\")' >> " + sqlFile 
+        sql5 = "echo 'show databases' >> " + sqlFile       
+        os.system(sql1)       
+        os.system(sql2)       
+        os.system(sql3)       
+        os.system(sql4)      
+        os.system(sql5)
+
+        keyDict['f'] = pwd + "/0-others/sql.txt"
+        retCode = taos_command("f", keyDict['f'], 'performance_schema', keyDict['c'], '', '', '')
+        print("============ ret code: ", retCode)
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -s fail")
 
+        print ("========== check new db ==========")
+        tdSql.query("show databases")        
+        for i in range(tdSql.queryRows):
+            #print ("dbseq: %d, dbname: %s"%(i, tdSql.getData(i, 0)))
+            if tdSql.getData(i, 0) == newDbName:
+                break
+        else:
+            tdLog.exit("create db fail after taos -f fail")
 
+        sqlString = "select * from " + newDbName + ".ntbf"    
+        tdSql.query(sqlString)
+        tdSql.checkData(0, 0, '2021-04-01 08:00:00.000')
+        tdSql.checkData(0, 1, 'test taos -f1')
+        tdSql.checkData(1, 0, '2021-04-01 08:00:01.000')
+        tdSql.checkData(1, 1, 'test taos -f2')
+        
+        shellCmd = "rm -f " + sqlFile
+        os.system(shellCmd)
+        tdSql.query('drop database %s'%newDbName)
+
+        tdLog.printNoPrefix("================================ parameter: -C")
+        newDbName="dbcc"
+        retCode, retVal = taos_command("C", keyDict['C'], "buildinfo", keyDict['c'], '', '', '')
+        if retCode != "TAOS_OK":
+            tdLog.exit("taos -C fail")
+
+
+        print ("-C return content:\n ", retVal)
+        totalCfgItem = {"firstEp":['', '', ''], }
+        for line in retVal.splitlines():
+            strList = line.split() 
+            if (len(strList) > 2):
+                totalCfgItem[strList[1]] = strList 
+
+        #print ("dict content:\n ", totalCfgItem)
+        firstEp = keyDict["h"] + ':' + keyDict['P']
+        if (totalCfgItem["firstEp"][2] != firstEp) and (totalCfgItem["firstEp"][0] != 'cfg_file'):
+            tdLog.exit("taos -C return firstEp error!")
+
+        if (totalCfgItem["rpcDebugFlag"][2] != self.rpcDebugFlagVal) and (totalCfgItem["rpcDebugFlag"][0] != 'cfg_file'):
+            tdLog.exit("taos -C return rpcDebugFlag error!")
+                
+        count = os.cpu_count()        
+        if (totalCfgItem["numOfCores"][2] != count) and (totalCfgItem["numOfCores"][0] != 'default'):
+            tdLog.exit("taos -C return numOfCores error!")
 
     def stop(self):
         tdSql.close()

tests/system-test/2-query/Today.py

@@ -8,6 +8,7 @@ from util.cases import *
 
 class TDTestCase:
 
+    updatecfgDict = {'rpcDebugFlag': '143'}
     def init(self, conn, logSql):
         tdLog.debug(f"start to excute {__file__}")
         tdSql.init(conn.cursor())

tools/shell/inc/shellInt.h

@@ -111,6 +111,5 @@ void shellTestNetWork();
 
 // shellMain.c
 extern SShellObj shell;
-extern void      taos_init();
 
 #endif /*_TD_SHELL_INT_H_*/