tmsg.h

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

#ifndef _TD_COMMON_TAOS_MSG_H_
#define _TD_COMMON_TAOS_MSG_H_

#ifdef __cplusplus
extern "C" {
#endif

#include "encode.h"
#include "taosdef.h"
#include "taoserror.h"
#include "tarray.h"
#include "tcoding.h"
#include "thash.h"
#include "tlist.h"
#include "trow.h"

/* ------------------------ MESSAGE DEFINITIONS ------------------------ */
#define TD_MSG_NUMBER_
#undef TD_MSG_DICT_
#undef TD_MSG_INFO_
#undef TD_MSG_SEG_CODE_
#include "tmsgdef.h"

#undef TD_MSG_NUMBER_
#undef TD_MSG_DICT_
#undef TD_MSG_INFO_
#define TD_MSG_SEG_CODE_
#include "tmsgdef.h"

#undef TD_MSG_NUMBER_
#undef TD_MSG_DICT_
#undef TD_MSG_INFO_
#undef TD_MSG_SEG_CODE_
#include "tmsgdef.h"

extern char* tMsgInfo[];
extern int   tMsgDict[];

#define TMSG_SEG_CODE(TYPE) (((TYPE)&0xff00) >> 8)
#define TMSG_SEG_SEQ(TYPE) ((TYPE)&0xff)
#define TMSG_INFO(TYPE) tMsgInfo[tMsgDict[TMSG_SEG_CODE(TYPE)] + TMSG_SEG_SEQ(TYPE)]
#define TMSG_INDEX(TYPE) (tMsgDict[TMSG_SEG_CODE(TYPE)] + TMSG_SEG_SEQ(TYPE))

typedef uint16_t tmsg_t;

/* ------------------------ OTHER DEFINITIONS ------------------------ */
// IE type
#define TSDB_IE_TYPE_SEC 1
#define TSDB_IE_TYPE_META 2
#define TSDB_IE_TYPE_MGMT_IP 3
#define TSDB_IE_TYPE_DNODE_CFG 4
#define TSDB_IE_TYPE_NEW_VERSION 5
#define TSDB_IE_TYPE_DNODE_EXT 6
#define TSDB_IE_TYPE_DNODE_STATE 7

typedef enum {
  HEARTBEAT_TYPE_MQ = 0,
  HEARTBEAT_TYPE_QUERY = 1,
  // types can be added here
  //
  HEARTBEAT_TYPE_MAX
} EHbType;

enum {
  HEARTBEAT_KEY_DBINFO = 1,
  HEARTBEAT_KEY_STBINFO,
  HEARTBEAT_KEY_MQ_TMP,
};

typedef enum _mgmt_table {
  TSDB_MGMT_TABLE_START,
  TSDB_MGMT_TABLE_ACCT,
  TSDB_MGMT_TABLE_USER,
  TSDB_MGMT_TABLE_DB,
  TSDB_MGMT_TABLE_TABLE,
  TSDB_MGMT_TABLE_DNODE,
  TSDB_MGMT_TABLE_MNODE,
  TSDB_MGMT_TABLE_QNODE,
  TSDB_MGMT_TABLE_SNODE,
  TSDB_MGMT_TABLE_BNODE,
  TSDB_MGMT_TABLE_VGROUP,
  TSDB_MGMT_TABLE_STB,
  TSDB_MGMT_TABLE_MODULE,
  TSDB_MGMT_TABLE_QUERIES,
  TSDB_MGMT_TABLE_STREAMS,
  TSDB_MGMT_TABLE_VARIABLES,
  TSDB_MGMT_TABLE_CONNS,
  TSDB_MGMT_TABLE_SCORES,
  TSDB_MGMT_TABLE_GRANTS,
  TSDB_MGMT_TABLE_VNODES,
  TSDB_MGMT_TABLE_CLUSTER,
  TSDB_MGMT_TABLE_STREAMTABLES,
  TSDB_MGMT_TABLE_TP,
  TSDB_MGMT_TABLE_FUNC,
  TSDB_MGMT_TABLE_MAX,
} EShowType;

#define TSDB_ALTER_TABLE_ADD_TAG 1
#define TSDB_ALTER_TABLE_DROP_TAG 2
#define TSDB_ALTER_TABLE_UPDATE_TAG_NAME 3
#define TSDB_ALTER_TABLE_UPDATE_TAG_VAL 4

#define TSDB_ALTER_TABLE_ADD_COLUMN 5
#define TSDB_ALTER_TABLE_DROP_COLUMN 6
#define TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES 7
#define TSDB_ALTER_TABLE_UPDATE_TAG_BYTES 8

#define TSDB_FILL_NONE 0
#define TSDB_FILL_NULL 1
#define TSDB_FILL_SET_VALUE 2
#define TSDB_FILL_LINEAR 3
#define TSDB_FILL_PREV 4
#define TSDB_FILL_NEXT 5

#define TSDB_ALTER_USER_PASSWD 0x1
#define TSDB_ALTER_USER_SUPERUSER 0x2
#define TSDB_ALTER_USER_ADD_READ_DB 0x3
#define TSDB_ALTER_USER_REMOVE_READ_DB 0x4
#define TSDB_ALTER_USER_CLEAR_READ_DB 0x5
#define TSDB_ALTER_USER_ADD_WRITE_DB 0x6
#define TSDB_ALTER_USER_REMOVE_WRITE_DB 0x7
#define TSDB_ALTER_USER_CLEAR_WRITE_DB 0x8

#define TSDB_ALTER_USER_PRIVILEGES 0x2

#define TSDB_KILL_MSG_LEN 30

#define TSDB_VN_READ_ACCCESS ((char)0x1)
#define TSDB_VN_WRITE_ACCCESS ((char)0x2)
#define TSDB_VN_ALL_ACCCESS (TSDB_VN_READ_ACCCESS | TSDB_VN_WRITE_ACCCESS)

#define TSDB_COL_NORMAL 0x0u  // the normal column of the table
#define TSDB_COL_TAG 0x1u     // the tag column type
#define TSDB_COL_UDC 0x2u     // the user specified normal string column, it is a dummy column
#define TSDB_COL_TMP 0x4u     // internal column generated by the previous operators
#define TSDB_COL_NULL 0x8u    // the column filter NULL or not

#define TSDB_COL_IS_TAG(f) (((f & (~(TSDB_COL_NULL))) & TSDB_COL_TAG) != 0)
#define TSDB_COL_IS_NORMAL_COL(f) ((f & (~(TSDB_COL_NULL))) == TSDB_COL_NORMAL)
#define TSDB_COL_IS_UD_COL(f) ((f & (~(TSDB_COL_NULL))) == TSDB_COL_UDC)
#define TSDB_COL_REQ_NULL(f) (((f)&TSDB_COL_NULL) != 0)

#define TD_SUPER_TABLE TSDB_SUPER_TABLE
#define TD_CHILD_TABLE TSDB_CHILD_TABLE
#define TD_NORMAL_TABLE TSDB_NORMAL_TABLE

typedef struct {
  int32_t vgId;
  char*   dbFName;
  char*   tbName;
} SBuildTableMetaInput;

typedef struct {
  char    db[TSDB_DB_FNAME_LEN];
  int32_t vgVersion;
} SBuildUseDBInput;

typedef struct SField {
  char    name[TSDB_COL_NAME_LEN];
  uint8_t type;
  int32_t bytes;
} SField;

#pragma pack(push, 1)

// null-terminated string instead of char array to avoid too many memory consumption in case of more than 1M tableMeta
typedef struct SEp {
  char     fqdn[TSDB_FQDN_LEN];
  uint16_t port;
} SEp;

typedef struct {
  int32_t contLen;
  int32_t vgId;
} SMsgHead;

// Submit message for one table
typedef struct SSubmitBlk {
  int64_t uid;        // table unique id
  int32_t tid;        // table id
  int32_t padding;    // TODO just for padding here
  int32_t sversion;   // data schema version
  int32_t dataLen;    // data part length, not including the SSubmitBlk head
  int32_t schemaLen;  // schema length, if length is 0, no schema exists
  int16_t numOfRows;  // total number of rows in current submit block
  char    data[];
} SSubmitBlk;

typedef struct {
  /* data */
} SSubmitReq;

typedef struct {
  /* data */
} SSubmitRsp;

typedef struct {
  /* data */
} SSubmitReqReader;

// Submit message for this TSDB
typedef struct {
  SMsgHead header;
  int64_t  version;
  int32_t  length;
  int32_t  numOfBlocks;
  char     blocks[];
} SSubmitMsg;

typedef struct {
  int32_t totalLen;
  int32_t len;
  STSRow* row;
} SSubmitBlkIter;

typedef struct {
  int32_t totalLen;
  int32_t len;
  void*   pMsg;
} SSubmitMsgIter;

int32_t tInitSubmitMsgIter(SSubmitMsg* pMsg, SSubmitMsgIter* pIter);
int32_t tGetSubmitMsgNext(SSubmitMsgIter* pIter, SSubmitBlk** pPBlock);
int32_t tInitSubmitBlkIter(SSubmitBlk* pBlock, SSubmitBlkIter* pIter);
STSRow* tGetSubmitBlkNext(SSubmitBlkIter* pIter);

typedef struct {
  int32_t index;  // index of failed block in submit blocks
  int32_t vnode;  // vnode index of failed block
  int32_t sid;    // table index of failed block
  int32_t code;   // errorcode while write data to vnode, such as not created, dropped, no space, invalid table
} SShellSubmitRspBlock;

typedef struct {
  int32_t              code;          // 0-success, > 0 error code
  int32_t              numOfRows;     // number of records the client is trying to write
  int32_t              affectedRows;  // number of records actually written
  int32_t              failedRows;    // number of failed records (exclude duplicate records)
  int32_t              numOfFailedBlocks;
  SShellSubmitRspBlock failedBlocks[];
} SShellSubmitRsp;

typedef struct SSchema {
  int8_t  type;
  int32_t colId;
  int32_t bytes;
  char    name[TSDB_COL_NAME_LEN];
} SSchema;

typedef struct {
  char    name[TSDB_TABLE_FNAME_LEN];
  int8_t  igExists;
  int32_t numOfColumns;
  int32_t numOfTags;
  SArray* pColumns;
  SArray* pTags;
  char    comment[TSDB_STB_COMMENT_LEN];
} SMCreateStbReq;

int32_t tSerializeSMCreateStbReq(void* buf, int32_t bufLen, SMCreateStbReq* pReq);
int32_t tDeserializeSMCreateStbReq(void* buf, int32_t bufLen, SMCreateStbReq* pReq);
void    tFreeSMCreateStbReq(SMCreateStbReq* pReq);

typedef struct {
  char   name[TSDB_TABLE_FNAME_LEN];
  int8_t igNotExists;
} SMDropStbReq;

int32_t tSerializeSMDropStbReq(void* buf, int32_t bufLen, SMDropStbReq* pReq);
int32_t tDeserializeSMDropStbReq(void* buf, int32_t bufLen, SMDropStbReq* pReq);

typedef struct {
  char    name[TSDB_TABLE_FNAME_LEN];
  int8_t  alterType;
  int32_t numOfFields;
  SArray* pFields;
} SMAltertbReq;

int32_t tSerializeSMAlterStbReq(void* buf, int32_t bufLen, SMAltertbReq* pReq);
int32_t tDeserializeSMAlterStbReq(void* buf, int32_t bufLen, SMAltertbReq* pReq);
void    tFreeSMAltertbReq(SMAltertbReq* pReq);

typedef struct SEpSet {
  int8_t inUse;
  int8_t numOfEps;
  SEp    eps[TSDB_MAX_REPLICA];
} SEpSet;

int32_t tEncodeSEpSet(SCoder* pEncoder, const SEpSet* pEp);
int32_t tDecodeSEpSet(SCoder* pDecoder, SEpSet* pEp);
int32_t taosEncodeSEpSet(void** buf, const SEpSet* pEp);
void*   taosDecodeSEpSet(void* buf, SEpSet* pEp);

typedef struct {
  int32_t pid;
  char    app[TSDB_APP_NAME_LEN];
  char    db[TSDB_DB_NAME_LEN];
  int64_t startTime;
} SConnectReq;

int32_t tSerializeSConnectReq(void* buf, int32_t bufLen, SConnectReq* pReq);
int32_t tDeserializeSConnectReq(void* buf, int32_t bufLen, SConnectReq* pReq);

typedef struct {
  int32_t acctId;
  int64_t clusterId;
  int32_t connId;
  int8_t  superUser;
  SEpSet  epSet;
  char    sVersion[128];
} SConnectRsp;

int32_t tSerializeSConnectRsp(void* buf, int32_t bufLen, SConnectRsp* pRsp);
int32_t tDeserializeSConnectRsp(void* buf, int32_t bufLen, SConnectRsp* pRsp);

typedef struct {
  char    user[TSDB_USER_LEN];
  char    pass[TSDB_PASSWORD_LEN];
  int32_t maxUsers;
  int32_t maxDbs;
  int32_t maxTimeSeries;
  int32_t maxStreams;
  int32_t accessState;  // Configured only by command
  int64_t maxStorage;   // In unit of GB
} SCreateAcctReq, SAlterAcctReq;

int32_t tSerializeSCreateAcctReq(void* buf, int32_t bufLen, SCreateAcctReq* pReq);
int32_t tDeserializeSCreateAcctReq(void* buf, int32_t bufLen, SCreateAcctReq* pReq);

typedef struct {
  char user[TSDB_USER_LEN];
} SDropUserReq, SDropAcctReq;

int32_t tSerializeSDropUserReq(void* buf, int32_t bufLen, SDropUserReq* pReq);
int32_t tDeserializeSDropUserReq(void* buf, int32_t bufLen, SDropUserReq* pReq);

typedef struct {
  int8_t createType;
  int8_t superUser;  // denote if it is a super user or not
  char   user[TSDB_USER_LEN];
  char   pass[TSDB_PASSWORD_LEN];
} SCreateUserReq;

int32_t tSerializeSCreateUserReq(void* buf, int32_t bufLen, SCreateUserReq* pReq);
int32_t tDeserializeSCreateUserReq(void* buf, int32_t bufLen, SCreateUserReq* pReq);

typedef struct {
  int8_t alterType;
  int8_t superUser;
  char   user[TSDB_USER_LEN];
  char   pass[TSDB_PASSWORD_LEN];
  char   dbname[TSDB_DB_FNAME_LEN];
} SAlterUserReq;

int32_t tSerializeSAlterUserReq(void* buf, int32_t bufLen, SAlterUserReq* pReq);
int32_t tDeserializeSAlterUserReq(void* buf, int32_t bufLen, SAlterUserReq* pReq);

typedef struct {
  char user[TSDB_USER_LEN];
} SGetUserAuthReq;

int32_t tSerializeSGetUserAuthReq(void* buf, int32_t bufLen, SGetUserAuthReq* pReq);
int32_t tDeserializeSGetUserAuthReq(void* buf, int32_t bufLen, SGetUserAuthReq* pReq);

typedef struct {
  char      user[TSDB_USER_LEN];
  int8_t    superAuth;
  SHashObj* readDbs;
  SHashObj* writeDbs;
} SGetUserAuthRsp;

int32_t tSerializeSGetUserAuthRsp(void* buf, int32_t bufLen, SGetUserAuthRsp* pRsp);
int32_t tDeserializeSGetUserAuthRsp(void* buf, int32_t bufLen, SGetUserAuthRsp* pRsp);

typedef struct {
  int16_t colId;     // column id
  int16_t colIndex;  // column index in colList if it is a normal column or index in tagColList if a tag
  int16_t flag;      // denote if it is a tag or a normal column
  char    name[TSDB_DB_FNAME_LEN];
} SColIndex;

typedef struct {
  int16_t lowerRelOptr;
  int16_t upperRelOptr;
  int16_t filterstr;  // denote if current column is char(binary/nchar)

  union {
    struct {
      int64_t lowerBndi;
      int64_t upperBndi;
    };
    struct {
      double lowerBndd;
      double upperBndd;
    };
    struct {
      int64_t pz;
      int64_t len;
    };
  };
} SColumnFilterInfo;

typedef struct {
  int16_t numOfFilters;
  union {
    int64_t            placeholder;
    SColumnFilterInfo* filterInfo;
  };
} SColumnFilterList;
/*
 * for client side struct, we only need the column id, type, bytes are not necessary
 * But for data in vnode side, we need all the following information.
 */
typedef struct {
  int16_t           colId;
  int16_t           type;
  int16_t           bytes;
  SColumnFilterList flist;
} SColumnInfo;

typedef struct {
  uint64_t uid;
  TSKEY    key;  // last accessed ts, for subscription
} STableIdInfo;

typedef struct STimeWindow {
  TSKEY skey;
  TSKEY ekey;
} STimeWindow;

typedef struct {
  int32_t tsOffset;       // offset value in current msg body, NOTE: ts list is compressed
  int32_t tsLen;          // total length of ts comp block
  int32_t tsNumOfBlocks;  // ts comp block numbers
  int32_t tsOrder;        // ts comp block order
} STsBufInfo;

typedef struct {
  int32_t tz;  // query client timezone
  char    intervalUnit;
  char    slidingUnit;
  char    offsetUnit;
  int64_t interval;
  int64_t sliding;
  int64_t offset;
} SInterval;

typedef struct {
  SMsgHead head;
  char     version[TSDB_VERSION_LEN];

  bool stableQuery;        // super table query or not
  bool topBotQuery;        // TODO used bitwise flag
  bool interpQuery;        // interp query or not
  bool groupbyColumn;      // denote if this is a groupby normal column query
  bool hasTagResults;      // if there are tag values in final result or not
  bool timeWindowInterpo;  // if the time window start/end required interpolation
  bool queryBlockDist;     // if query data block distribution
  bool stabledev;          // super table stddev query
  bool tsCompQuery;        // is tscomp query
  bool simpleAgg;
  bool pointInterpQuery;  // point interpolation query
  bool needReverseScan;   // need reverse scan
  bool stateWindow;       // state window flag

  STimeWindow window;
  int32_t     numOfTables;
  int16_t     order;
  int16_t     orderColId;
  int16_t     numOfCols;  // the number of columns will be load from vnode
  SInterval   interval;
  //  SSessionWindow sw;            // session window
  int16_t tagCondLen;      // tag length in current query
  int16_t colCondLen;      // column length in current query
  int16_t numOfGroupCols;  // num of group by columns
  int16_t orderByIdx;
  int16_t orderType;    // used in group by xx order by xxx
  int64_t vgroupLimit;  // limit the number of rows for each table, used in order by + limit in stable projection query.
  int16_t prjOrder;     // global order in super table projection query.
  int64_t limit;
  int64_t offset;
  int32_t queryType;    // denote another query process
  int16_t numOfOutput;  // final output columns numbers
  int16_t fillType;     // interpolate type
  int64_t fillVal;      // default value array list
  int32_t secondStageOutput;
  STsBufInfo  tsBuf;          // tsBuf info
  int32_t     numOfTags;      // number of tags columns involved
  int32_t     sqlstrLen;      // sql query string
  int32_t     prevResultLen;  // previous result length
  int32_t     numOfOperator;
  int32_t     tableScanOperator;  // table scan operator. -1 means no scan operator
  int32_t     udfNum;             // number of udf function
  int32_t     udfContentOffset;
  int32_t     udfContentLen;
  SColumnInfo tableCols[];
} SQueryTableReq;

typedef struct {
  int32_t code;
} SQueryTableRsp;

typedef struct {
  char    db[TSDB_DB_FNAME_LEN];
  int32_t numOfVgroups;
  int32_t cacheBlockSize;  // MB
  int32_t totalBlocks;
  int32_t daysPerFile;
  int32_t daysToKeep0;
  int32_t daysToKeep1;
  int32_t daysToKeep2;
  int32_t minRows;
  int32_t maxRows;
  int32_t commitTime;
  int32_t fsyncPeriod;
  int8_t  walLevel;
  int8_t  precision;  // time resolution
  int8_t  compression;
  int8_t  replications;
  int8_t  quorum;
  int8_t  update;
  int8_t  cacheLastRow;
  int8_t  ignoreExist;
} SCreateDbReq;

int32_t tSerializeSCreateDbReq(void* buf, int32_t bufLen, SCreateDbReq* pReq);
int32_t tDeserializeSCreateDbReq(void* buf, int32_t bufLen, SCreateDbReq* pReq);

typedef struct {
  char    db[TSDB_DB_FNAME_LEN];
  int32_t totalBlocks;
  int32_t daysToKeep0;
  int32_t daysToKeep1;
  int32_t daysToKeep2;
  int32_t fsyncPeriod;
  int8_t  walLevel;
  int8_t  quorum;
  int8_t  cacheLastRow;
} SAlterDbReq;

int32_t tSerializeSAlterDbReq(void* buf, int32_t bufLen, SAlterDbReq* pReq);
int32_t tDeserializeSAlterDbReq(void* buf, int32_t bufLen, SAlterDbReq* pReq);

typedef struct {
  char   db[TSDB_DB_FNAME_LEN];
  int8_t ignoreNotExists;
} SDropDbReq;

int32_t tSerializeSDropDbReq(void* buf, int32_t bufLen, SDropDbReq* pReq);
int32_t tDeserializeSDropDbReq(void* buf, int32_t bufLen, SDropDbReq* pReq);

typedef struct {
  char     db[TSDB_DB_FNAME_LEN];
  uint64_t uid;
} SDropDbRsp;

int32_t tSerializeSDropDbRsp(void* buf, int32_t bufLen, SDropDbRsp* pRsp);
int32_t tDeserializeSDropDbRsp(void* buf, int32_t bufLen, SDropDbRsp* pRsp);

typedef struct {
  char    db[TSDB_DB_FNAME_LEN];
  int32_t vgVersion;
} SUseDbReq;

int32_t tSerializeSUseDbReq(void* buf, int32_t bufLen, SUseDbReq* pReq);
int32_t tDeserializeSUseDbReq(void* buf, int32_t bufLen, SUseDbReq* pReq);

typedef struct {
  char     db[TSDB_DB_FNAME_LEN];
  uint64_t uid;
  int32_t  vgVersion;
  int32_t  vgNum;
  int8_t   hashMethod;
  SArray*  pVgroupInfos;  // Array of SVgroupInfo
} SUseDbRsp;

int32_t tSerializeSUseDbRsp(void* buf, int32_t bufLen, SUseDbRsp* pRsp);
int32_t tDeserializeSUseDbRsp(void* buf, int32_t bufLen, SUseDbRsp* pRsp);
void    tFreeSUsedbRsp(SUseDbRsp* pRsp);

typedef struct {
  SArray* pArray;  // Array of SUseDbRsp
} SUseDbBatchRsp;

int32_t tSerializeSUseDbBatchRsp(void* buf, int32_t bufLen, SUseDbBatchRsp* pRsp);
int32_t tDeserializeSUseDbBatchRsp(void* buf, int32_t bufLen, SUseDbBatchRsp* pRsp);
void    tFreeSUseDbBatchRsp(SUseDbBatchRsp* pRsp);

typedef struct {
  char db[TSDB_DB_FNAME_LEN];
} SSyncDbReq, SCompactDbReq;

int32_t tSerializeSSyncDbReq(void* buf, int32_t bufLen, SSyncDbReq* pReq);
int32_t tDeserializeSSyncDbReq(void* buf, int32_t bufLen, SSyncDbReq* pReq);

typedef struct {
  char    name[TSDB_FUNC_NAME_LEN];
  int8_t  igExists;
  int8_t  funcType;
  int8_t  scriptType;
  int8_t  outputType;
  int32_t outputLen;
  int32_t bufSize;
  int64_t signature;
  int32_t commentSize;
  int32_t codeSize;
  char    pComment[TSDB_FUNC_COMMENT_LEN];
  char    pCode[TSDB_FUNC_CODE_LEN];
} SCreateFuncReq;

int32_t tSerializeSCreateFuncReq(void* buf, int32_t bufLen, SCreateFuncReq* pReq);
int32_t tDeserializeSCreateFuncReq(void* buf, int32_t bufLen, SCreateFuncReq* pReq);

typedef struct {
  char   name[TSDB_FUNC_NAME_LEN];
  int8_t igNotExists;
} SDropFuncReq;

int32_t tSerializeSDropFuncReq(void* buf, int32_t bufLen, SDropFuncReq* pReq);
int32_t tDeserializeSDropFuncReq(void* buf, int32_t bufLen, SDropFuncReq* pReq);

typedef struct {
  int32_t numOfFuncs;
  SArray* pFuncNames;
} SRetrieveFuncReq;

int32_t tSerializeSRetrieveFuncReq(void* buf, int32_t bufLen, SRetrieveFuncReq* pReq);
int32_t tDeserializeSRetrieveFuncReq(void* buf, int32_t bufLen, SRetrieveFuncReq* pReq);

typedef struct {
  char    name[TSDB_FUNC_NAME_LEN];
  int8_t  funcType;
  int8_t  scriptType;
  int8_t  outputType;
  int32_t outputLen;
  int32_t bufSize;
  int64_t signature;
  int32_t commentSize;
  int32_t codeSize;
  char    pComment[TSDB_FUNC_COMMENT_LEN];
  char    pCode[TSDB_FUNC_CODE_LEN];
} SFuncInfo;

typedef struct {
  int32_t numOfFuncs;
  SArray* pFuncInfos;
} SRetrieveFuncRsp;

int32_t tSerializeSRetrieveFuncRsp(void* buf, int32_t bufLen, SRetrieveFuncRsp* pRsp);
int32_t tDeserializeSRetrieveFuncRsp(void* buf, int32_t bufLen, SRetrieveFuncRsp* pRsp);

typedef struct {
  int32_t statusInterval;
  int64_t checkTime;                    // 1970-01-01 00:00:00.000
  char    timezone[TSDB_TIMEZONE_LEN];  // tsTimezone
  char    locale[TSDB_LOCALE_LEN];      // tsLocale
  char    charset[TSDB_LOCALE_LEN];     // tsCharset
} SClusterCfg;

typedef struct {
  int32_t vgId;
  int8_t  role;
  int64_t numOfTables;
  int64_t numOfTimeSeries;
  int64_t totalStorage;
  int64_t compStorage;
  int64_t pointsWritten;
} SVnodeLoad;

typedef struct {
  int32_t     sver;  // software version
  int64_t     dver;  // dnode table version in sdb
  int32_t     dnodeId;
  int64_t     clusterId;
  int64_t     rebootTime;
  int64_t     updateTime;
  int32_t     numOfCores;
  int32_t     numOfSupportVnodes;
  char        dnodeEp[TSDB_EP_LEN];
  SClusterCfg clusterCfg;
  SArray*     pVloads;  // array of SVnodeLoad
} SStatusReq;

int32_t tSerializeSStatusReq(void* buf, int32_t bufLen, SStatusReq* pReq);
int32_t tDeserializeSStatusReq(void* buf, int32_t bufLen, SStatusReq* pReq);

typedef struct {
  int32_t dnodeId;
  int64_t clusterId;
} SDnodeCfg;

typedef struct {
  int32_t id;
  int8_t  isMnode;
  SEp     ep;
} SDnodeEp;

typedef struct {
  int64_t   dver;
  SDnodeCfg dnodeCfg;
  SArray*   pDnodeEps;  // Array of SDnodeEp
} SStatusRsp;

int32_t tSerializeSStatusRsp(void* buf, int32_t bufLen, SStatusRsp* pRsp);
int32_t tDeserializeSStatusRsp(void* buf, int32_t bufLen, SStatusRsp* pRsp);

typedef struct {
  int32_t reserved;
} SMTimerReq;

int32_t tSerializeSMTimerMsg(void* buf, int32_t bufLen, SMTimerReq* pReq);
int32_t tDeserializeSMTimerMsg(void* buf, int32_t bufLen, SMTimerReq* pReq);

typedef struct {
  int32_t  id;
  uint16_t port;                 // node sync Port
  char     fqdn[TSDB_FQDN_LEN];  // node FQDN
} SReplica;

typedef struct {
  int32_t  vgId;
  int32_t  dnodeId;
  char     db[TSDB_DB_FNAME_LEN];
  uint64_t dbUid;
  int32_t  vgVersion;
  int32_t  cacheBlockSize;
  int32_t  totalBlocks;
  int32_t  daysPerFile;
  int32_t  daysToKeep0;
  int32_t  daysToKeep1;
  int32_t  daysToKeep2;
  int32_t  minRows;
  int32_t  maxRows;
  int32_t  commitTime;
  int32_t  fsyncPeriod;
  int8_t   walLevel;
  int8_t   precision;
  int8_t   compression;
  int8_t   quorum;
  int8_t   update;
  int8_t   cacheLastRow;
  int8_t   replica;
  int8_t   selfIndex;
  SReplica replicas[TSDB_MAX_REPLICA];
} SCreateVnodeReq, SAlterVnodeReq;

int32_t tSerializeSCreateVnodeReq(void* buf, int32_t bufLen, SCreateVnodeReq* pReq);
int32_t tDeserializeSCreateVnodeReq(void* buf, int32_t bufLen, SCreateVnodeReq* pReq);

typedef struct {
  int32_t  vgId;
  int32_t  dnodeId;
  uint64_t dbUid;
  char     db[TSDB_DB_FNAME_LEN];
} SDropVnodeReq, SSyncVnodeReq, SCompactVnodeReq;

int32_t tSerializeSDropVnodeReq(void* buf, int32_t bufLen, SDropVnodeReq* pReq);
int32_t tDeserializeSDropVnodeReq(void* buf, int32_t bufLen, SDropVnodeReq* pReq);

typedef struct {
  SMsgHead header;
  char     dbFName[TSDB_DB_FNAME_LEN];
  char     tbName[TSDB_TABLE_NAME_LEN];
} STableInfoReq;

int32_t tSerializeSTableInfoReq(void* buf, int32_t bufLen, STableInfoReq* pReq);
int32_t tDeserializeSTableInfoReq(void* buf, int32_t bufLen, STableInfoReq* pReq);

typedef struct {
  int8_t  metaClone;  // create local clone of the cached table meta
  int32_t numOfVgroups;
  int32_t numOfTables;
  int32_t numOfUdfs;
  char    tableNames[];
} SMultiTableInfoReq;

// todo refactor
typedef struct SVgroupInfo {
  int32_t  vgId;
  uint32_t hashBegin;
  uint32_t hashEnd;
  SEpSet   epset;
} SVgroupInfo;

typedef struct {
  int32_t     numOfVgroups;
  SVgroupInfo vgroups[];
} SVgroupsInfo;

typedef struct {
  char     tbName[TSDB_TABLE_NAME_LEN];
  char     stbName[TSDB_TABLE_NAME_LEN];
  char     dbFName[TSDB_DB_FNAME_LEN];
  uint64_t dbId;
  int32_t  numOfTags;
  int32_t  numOfColumns;
  int8_t   precision;
  int8_t   tableType;
  int8_t   update;
  int32_t  sversion;
  int32_t  tversion;
  uint64_t suid;
  uint64_t tuid;
  int32_t  vgId;
  SSchema* pSchemas;
} STableMetaRsp;

int32_t tSerializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp);
int32_t tDeserializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp);
void    tFreeSTableMetaRsp(STableMetaRsp* pRsp);

typedef struct {
  SArray* pArray;  // Array of STableMetaRsp
} STableMetaBatchRsp;

int32_t tSerializeSTableMetaBatchRsp(void* buf, int32_t bufLen, STableMetaBatchRsp* pRsp);
int32_t tDeserializeSTableMetaBatchRsp(void* buf, int32_t bufLen, STableMetaBatchRsp* pRsp);
void    tFreeSTableMetaBatchRsp(STableMetaBatchRsp* pRsp);

typedef struct {
  int32_t numOfTables;
  int32_t numOfVgroup;
  int32_t numOfUdf;
  int32_t contLen;
  int8_t  compressed;  // denote if compressed or not
  int32_t rawLen;      // size before compress
  uint8_t metaClone;   // make meta clone after retrieve meta from mnode
  char    meta[];
} SMultiTableMeta;

typedef struct {
  int32_t dataLen;
  char    name[TSDB_TABLE_FNAME_LEN];
  char*   data;
} STagData;

/*
 * sql: show tables like '%a_%'
 * payload is the query condition, e.g., '%a_%'
 * payloadLen is the length of payload
 */
typedef struct {
  int32_t type;
  char    db[TSDB_DB_FNAME_LEN];
  int32_t payloadLen;
  char*   payload;
} SShowReq;

int32_t tSerializeSShowReq(void* buf, int32_t bufLen, SShowReq* pReq);
int32_t tDeserializeSShowReq(void* buf, int32_t bufLen, SShowReq* pReq);
void    tFreeSShowReq(SShowReq* pReq);

typedef struct {
  int64_t       showId;
  STableMetaRsp tableMeta;
} SShowRsp, SVShowTablesRsp;

int32_t tSerializeSShowRsp(void* buf, int32_t bufLen, SShowRsp* pRsp);
int32_t tDeserializeSShowRsp(void* buf, int32_t bufLen, SShowRsp* pRsp);
void    tFreeSShowRsp(SShowRsp* pRsp);

typedef struct {
  int64_t showId;
  int8_t  free;
} SRetrieveTableReq;

int32_t tSerializeSRetrieveTableReq(void* buf, int32_t bufLen, SRetrieveTableReq* pReq);
int32_t tDeserializeSRetrieveTableReq(void* buf, int32_t bufLen, SRetrieveTableReq* pReq);

typedef struct {
  int64_t useconds;
  int8_t  completed;  // all results are returned to client
  int8_t  precision;
  int8_t  compressed;
  int32_t compLen;

  int32_t numOfRows;
  char    data[];
} SRetrieveTableRsp;

typedef struct {
  char    fqdn[TSDB_FQDN_LEN];  // end point, hostname:port
  int32_t port;
} SCreateDnodeReq;

int32_t tSerializeSCreateDnodeReq(void* buf, int32_t bufLen, SCreateDnodeReq* pReq);
int32_t tDeserializeSCreateDnodeReq(void* buf, int32_t bufLen, SCreateDnodeReq* pReq);

typedef struct {
  int32_t dnodeId;
} SDropDnodeReq;

int32_t tSerializeSDropDnodeReq(void* buf, int32_t bufLen, SDropDnodeReq* pReq);
int32_t tDeserializeSDropDnodeReq(void* buf, int32_t bufLen, SDropDnodeReq* pReq);

typedef struct {
  int32_t dnodeId;
  char    config[TSDB_DNODE_CONFIG_LEN];
  char    value[TSDB_DNODE_VALUE_LEN];
} SMCfgDnodeReq, SDCfgDnodeReq;

int32_t tSerializeSMCfgDnodeReq(void* buf, int32_t bufLen, SMCfgDnodeReq* pReq);
int32_t tDeserializeSMCfgDnodeReq(void* buf, int32_t bufLen, SMCfgDnodeReq* pReq);

typedef struct {
  int32_t dnodeId;
} SMCreateMnodeReq, SMDropMnodeReq, SDDropMnodeReq;

int32_t tSerializeSMCreateDropMnodeReq(void* buf, int32_t bufLen, SMCreateMnodeReq* pReq);
int32_t tDeserializeSMCreateDropMnodeReq(void* buf, int32_t bufLen, SMCreateMnodeReq* pReq);

typedef struct {
  int32_t  dnodeId;
  int8_t   replica;
  SReplica replicas[TSDB_MAX_REPLICA];
} SDCreateMnodeReq, SDAlterMnodeReq;

typedef struct {
  int32_t dnodeId;
} SMCreateQnodeReq, SMDropQnodeReq, SDCreateQnodeReq, SDDropQnodeReq, SMCreateSnodeReq, SMDropSnodeReq,
    SDCreateSnodeReq, SDDropSnodeReq, SMCreateBnodeReq, SMDropBnodeReq, SDCreateBnodeReq, SDDropBnodeReq;

int32_t tSerializeSMCreateDropQSBNodeReq(void* buf, int32_t bufLen, SMCreateQnodeReq* pReq);
int32_t tDeserializeSMCreateDropQSBNodeReq(void* buf, int32_t bufLen, SMCreateQnodeReq* pReq);

typedef struct {
  char    sql[TSDB_SHOW_SQL_LEN];
  int32_t queryId;
  int64_t useconds;
  int64_t stime;
  int64_t qId;
  int64_t sqlObjId;
  int32_t pid;
  char    fqdn[TSDB_FQDN_LEN];
  int8_t  stableQuery;
  int32_t numOfSub;
  char    subSqlInfo[TSDB_SHOW_SUBQUERY_LEN];  // include subqueries' index, Obj IDs and states(C-complete/I-imcomplete)
} SQueryDesc;

typedef struct {
  int32_t connId;
  int32_t pid;
  int32_t numOfQueries;
  int32_t numOfStreams;
  char    app[TSDB_APP_NAME_LEN];
  char    pData[];
} SHeartBeatReq;

typedef struct {
  int32_t connId;
  int32_t queryId;
  int32_t streamId;
  int32_t totalDnodes;
  int32_t onlineDnodes;
  int8_t  killConnection;
  int8_t  align[3];
  SEpSet  epSet;
} SHeartBeatRsp;

typedef struct {
  int32_t connId;
  int32_t queryId;
} SKillQueryReq;

typedef struct {
  int32_t connId;
} SKillConnReq;

typedef struct {
  char user[TSDB_USER_LEN];
  char spi;
  char encrypt;
  char secret[TSDB_PASSWORD_LEN];
  char ckey[TSDB_PASSWORD_LEN];
} SAuthReq, SAuthRsp;

typedef struct {
  int8_t finished;
  int8_t align[7];
  char   name[TSDB_STEP_NAME_LEN];
  char   desc[TSDB_STEP_DESC_LEN];
} SStartupReq;

/**
 * The layout of the query message payload is as following:
 * +--------------------+---------------------------------+
 * |Sql statement       | Physical plan                   |
 * |(denoted by sqlLen) |(In JSON, denoted by contentLen) |
 * +--------------------+---------------------------------+
 */
typedef struct SSubQueryMsg {
  SMsgHead header;
  uint64_t sId;
  uint64_t queryId;
  uint64_t taskId;
  int8_t   taskType;
  uint32_t sqlLen;  // the query sql,
  uint32_t phyLen;
  char     msg[];
} SSubQueryMsg;

typedef struct {
  SMsgHead header;
  uint64_t sId;
  uint64_t queryId;
  uint64_t taskId;
} SSinkDataReq;

typedef struct {
  SMsgHead header;
  uint64_t sId;
  uint64_t queryId;
  uint64_t taskId;
} SQueryContinueReq;

typedef struct {
  SMsgHead header;
  uint64_t sId;
  uint64_t queryId;
  uint64_t taskId;
} SResReadyReq;

typedef struct {
  int32_t code;
} SResReadyRsp;

typedef struct {
  SMsgHead header;
  uint64_t sId;
  uint64_t queryId;
  uint64_t taskId;
} SResFetchReq;

typedef struct {
  SMsgHead header;
  uint64_t sId;
} SSchTasksStatusReq;

typedef struct {
  uint64_t queryId;
  uint64_t taskId;
  int8_t   status;
} STaskStatus;

typedef struct {
  uint32_t    num;
  STaskStatus status[];
} SSchedulerStatusRsp;

typedef struct {
  SMsgHead header;
  uint64_t sId;
  uint64_t queryId;
  uint64_t taskId;
} STaskCancelReq;

typedef struct {
  int32_t code;
} STaskCancelRsp;

typedef struct {
  SMsgHead header;
  uint64_t sId;
  uint64_t queryId;
  uint64_t taskId;
} STaskDropReq;

typedef struct {
  int32_t code;
} STaskDropRsp;

typedef struct {
  int8_t igExists;
  char*  name;
  char*  sql;
  char*  physicalPlan;
  char*  logicalPlan;
} SMCreateTopicReq;

int32_t tSerializeMCreateTopicReq(void** buf, const SMCreateTopicReq* pReq);
void*   tDeserializeSMCreateTopicReq(void* buf, SMCreateTopicReq* pReq);

typedef struct {
  int64_t topicId;
} SMCreateTopicRsp;

int32_t tSerializeSMCreateTopicRsp(void* buf, int32_t bufLen, const SMCreateTopicRsp* pRsp);
int32_t tDeserializeSMCreateTopicRsp(void* buf, int32_t bufLen, SMCreateTopicRsp* pRsp);

typedef struct {
  int32_t topicNum;
  int64_t consumerId;
  char*   consumerGroup;
  SArray* topicNames;  // SArray<char*>
} SCMSubscribeReq;

static FORCE_INLINE int tSerializeSCMSubscribeReq(void** buf, const SCMSubscribeReq* pReq) {
  int tlen = 0;
  tlen += taosEncodeFixedI32(buf, pReq->topicNum);
  tlen += taosEncodeFixedI64(buf, pReq->consumerId);
  tlen += taosEncodeString(buf, pReq->consumerGroup);

  for (int i = 0; i < pReq->topicNum; i++) {
    tlen += taosEncodeString(buf, (char*)taosArrayGetP(pReq->topicNames, i));
  }
  return tlen;
}

static FORCE_INLINE void* tDeserializeSCMSubscribeReq(void* buf, SCMSubscribeReq* pReq) {
  buf = taosDecodeFixedI32(buf, &pReq->topicNum);
  buf = taosDecodeFixedI64(buf, &pReq->consumerId);
  buf = taosDecodeString(buf, &pReq->consumerGroup);
  pReq->topicNames = taosArrayInit(pReq->topicNum, sizeof(void*));
  for (int i = 0; i < pReq->topicNum; i++) {
    char* name;
    buf = taosDecodeString(buf, &name);
    taosArrayPush(pReq->topicNames, &name);
  }
  return buf;
}

typedef struct SMqSubTopic {
  int32_t vgId;
  int64_t topicId;
  SEpSet  epSet;
} SMqSubTopic;

typedef struct {
  int32_t     topicNum;
  SMqSubTopic topics[];
} SCMSubscribeRsp;

static FORCE_INLINE int tSerializeSCMSubscribeRsp(void** buf, const SCMSubscribeRsp* pRsp) {
  int tlen = 0;
  tlen += taosEncodeFixedI32(buf, pRsp->topicNum);
  for (int i = 0; i < pRsp->topicNum; i++) {
    tlen += taosEncodeFixedI32(buf, pRsp->topics[i].vgId);
    tlen += taosEncodeFixedI64(buf, pRsp->topics[i].topicId);
    tlen += taosEncodeSEpSet(buf, &pRsp->topics[i].epSet);
  }
  return tlen;
}

static FORCE_INLINE void* tDeserializeSCMSubscribeRsp(void* buf, SCMSubscribeRsp* pRsp) {
  buf = taosDecodeFixedI32(buf, &pRsp->topicNum);
  for (int i = 0; i < pRsp->topicNum; i++) {
    buf = taosDecodeFixedI32(buf, &pRsp->topics[i].vgId);
    buf = taosDecodeFixedI64(buf, &pRsp->topics[i].topicId);
    buf = taosDecodeSEpSet(buf, &pRsp->topics[i].epSet);
  }
  return buf;
}

typedef struct {
  int64_t topicId;
  int64_t consumerId;
  int64_t consumerGroupId;
  int64_t offset;
  char*   sql;
  char*   logicalPlan;
  char*   physicalPlan;
} SMVSubscribeReq;

static FORCE_INLINE int tSerializeSMVSubscribeReq(void** buf, SMVSubscribeReq* pReq) {
  int tlen = 0;
  tlen += taosEncodeFixedI64(buf, pReq->topicId);
  tlen += taosEncodeFixedI64(buf, pReq->consumerId);
  tlen += taosEncodeFixedI64(buf, pReq->consumerGroupId);
  tlen += taosEncodeFixedI64(buf, pReq->offset);
  tlen += taosEncodeString(buf, pReq->sql);
  tlen += taosEncodeString(buf, pReq->logicalPlan);
  tlen += taosEncodeString(buf, pReq->physicalPlan);
  return tlen;
}

static FORCE_INLINE void* tDeserializeSMVSubscribeReq(void* buf, SMVSubscribeReq* pReq) {
  buf = taosDecodeFixedI64(buf, &pReq->topicId);
  buf = taosDecodeFixedI64(buf, &pReq->consumerId);
  buf = taosDecodeFixedI64(buf, &pReq->consumerGroupId);
  buf = taosDecodeFixedI64(buf, &pReq->offset);
  buf = taosDecodeString(buf, &pReq->sql);
  buf = taosDecodeString(buf, &pReq->logicalPlan);
  buf = taosDecodeString(buf, &pReq->physicalPlan);
  return buf;
}

typedef struct {
  const char* key;
  SArray*     lostConsumers;     // SArray<int64_t>
  SArray*     removedConsumers;  // SArray<int64_t>
  SArray*     newConsumers;      // SArray<int64_t>
} SMqRebSubscribe;

static FORCE_INLINE SMqRebSubscribe* tNewSMqRebSubscribe(const char* key) {
  SMqRebSubscribe* pRebSub = (SMqRebSubscribe*)calloc(1, sizeof(SMqRebSubscribe));
  if (pRebSub == NULL) {
    goto _err;
  }
  pRebSub->key = key;
  pRebSub->lostConsumers = taosArrayInit(0, sizeof(int64_t));
  if (pRebSub->lostConsumers == NULL) {
    goto _err;
  }
  pRebSub->removedConsumers = taosArrayInit(0, sizeof(int64_t));
  if (pRebSub->removedConsumers == NULL) {
    goto _err;
  }
  pRebSub->newConsumers = taosArrayInit(0, sizeof(int64_t));
  if (pRebSub->newConsumers == NULL) {
    goto _err;
  }
  return pRebSub;
_err:
  taosArrayDestroy(pRebSub->lostConsumers);
  taosArrayDestroy(pRebSub->removedConsumers);
  taosArrayDestroy(pRebSub->newConsumers);
  tfree(pRebSub);
  return NULL;
}

// this message is sent from mnode to mnode(read thread to write thread), so there is no need for serialization /
// deserialization
typedef struct {
  // SArray* rebSubscribes;     //SArray<SMqRebSubscribe>
  SHashObj* rebSubHash;  // SHashObj<key, SMqRebSubscribe>
} SMqDoRebalanceMsg;

#if 0
static FORCE_INLINE SMqDoRebalanceMsg* tNewSMqDoRebalanceMsg() {
  SMqDoRebalanceMsg *pMsg = malloc(sizeof(SMqDoRebalanceMsg));
  if (pMsg == NULL) {
    return NULL;
  }
  pMsg->rebSubscribes = taosArrayInit(0, sizeof(SMqRebSubscribe));
  if (pMsg->rebSubscribes == NULL) {
    free(pMsg);
    return NULL;
  }
  return pMsg;
}
#endif

typedef struct {
  int64_t status;
} SMVSubscribeRsp;

typedef struct {
  char   name[TSDB_TABLE_FNAME_LEN];
  int8_t igNotExists;
} SMDropTopicReq;

int32_t tSerializeSMDropTopicReqq(void* buf, int32_t bufLen, SMDropTopicReq* pReq);
int32_t tDeserializeSMDropTopicReq(void* buf, int32_t bufLen, SMDropTopicReq* pReq);

typedef struct {
  char    name[TSDB_TABLE_FNAME_LEN];
  int8_t  alterType;
  SSchema schema;
} SAlterTopicReq;

typedef struct {
  SMsgHead head;
  char     name[TSDB_TABLE_FNAME_LEN];
  uint64_t tuid;
  int32_t  sverson;
  int32_t  execLen;
  char*    executor;
  int32_t  sqlLen;
  char*    sql;
} SDCreateTopicReq;

typedef struct {
  SMsgHead head;
  char     name[TSDB_TABLE_FNAME_LEN];
  uint64_t tuid;
} SDDropTopicReq;

typedef struct SVCreateTbReq {
  uint64_t ver;  // use a general definition
  char*    name;
  uint32_t ttl;
  uint32_t keep;
  uint8_t  type;
  union {
    struct {
      tb_uid_t suid;
      uint32_t nCols;
      SSchema* pSchema;
      uint32_t nTagCols;
      SSchema* pTagSchema;
    } stbCfg;
    struct {
      tb_uid_t suid;
      SKVRow   pTag;
    } ctbCfg;
    struct {
      uint32_t nCols;
      SSchema* pSchema;
    } ntbCfg;
  };
} SVCreateTbReq, SVUpdateTbReq;

typedef struct {
} SVCreateTbRsp, SVUpdateTbRsp;

int32_t tSerializeSVCreateTbReq(void** buf, SVCreateTbReq* pReq);
void*   tDeserializeSVCreateTbReq(void* buf, SVCreateTbReq* pReq);

typedef struct {
  uint64_t ver;  // use a general definition
  SArray*  pArray;
} SVCreateTbBatchReq;

typedef struct {
} SVCreateTbBatchRsp;

int32_t tSerializeSVCreateTbBatchReq(void** buf, SVCreateTbBatchReq* pReq);
void*   tDeserializeSVCreateTbBatchReq(void* buf, SVCreateTbBatchReq* pReq);

typedef struct {
  uint64_t ver;
  char*    name;
  uint8_t  type;
  tb_uid_t suid;
} SVDropTbReq;

typedef struct {
} SVDropTbRsp;

int32_t tSerializeSVDropTbReq(void** buf, SVDropTbReq* pReq);
void*   tDeserializeSVDropTbReq(void* buf, SVDropTbReq* pReq);

typedef struct {
  SMsgHead head;
  int64_t  uid;
  int32_t  tid;
  int16_t  tversion;
  int16_t  colId;
  int8_t   type;
  int16_t  bytes;
  int32_t  tagValLen;
  int16_t  numOfTags;
  int32_t  schemaLen;
  char     data[];
} SUpdateTagValReq;

typedef struct {
  SMsgHead head;
} SUpdateTagValRsp;

typedef struct {
  SMsgHead head;
} SVShowTablesReq;

typedef struct {
  SMsgHead head;
  int32_t  id;
} SVShowTablesFetchReq;

typedef struct {
  int64_t useconds;
  int8_t  completed;  // all results are returned to client
  int8_t  precision;
  int8_t  compressed;
  int32_t compLen;

  int32_t numOfRows;
  char    data[];
} SVShowTablesFetchRsp;

typedef struct SMqCMGetSubEpReq {
  int64_t consumerId;
  int32_t epoch;
  char    cgroup[TSDB_CONSUMER_GROUP_LEN];
} SMqCMGetSubEpReq;

#pragma pack(pop)

static FORCE_INLINE int32_t tEncodeSMsgHead(void** buf, const SMsgHead* pMsg) {
  int32_t tlen = 0;
  tlen += taosEncodeFixedI32(buf, pMsg->contLen);
  tlen += taosEncodeFixedI32(buf, pMsg->vgId);
  return tlen;
}

typedef struct SMqHbRsp {
  int8_t status;  // idle or not
  int8_t vnodeChanged;
  int8_t epChanged;  // should use new epset
  int8_t reserved;
  SEpSet epSet;
} SMqHbRsp;

static FORCE_INLINE int taosEncodeSMqHbRsp(void** buf, const SMqHbRsp* pRsp) {
  int tlen = 0;
  tlen += taosEncodeFixedI8(buf, pRsp->status);
  tlen += taosEncodeFixedI8(buf, pRsp->vnodeChanged);
  tlen += taosEncodeFixedI8(buf, pRsp->epChanged);
  tlen += taosEncodeSEpSet(buf, &pRsp->epSet);
  return tlen;
}

static FORCE_INLINE void* taosDecodeSMqHbRsp(void* buf, SMqHbRsp* pRsp) {
  buf = taosDecodeFixedI8(buf, &pRsp->status);
  buf = taosDecodeFixedI8(buf, &pRsp->vnodeChanged);
  buf = taosDecodeFixedI8(buf, &pRsp->epChanged);
  buf = taosDecodeSEpSet(buf, &pRsp->epSet);
  return buf;
}

typedef struct SMqHbOneTopicBatchRsp {
  char    topicName[TSDB_TOPIC_FNAME_LEN];
  SArray* rsps;  // SArray<SMqHbRsp>
} SMqHbOneTopicBatchRsp;

static FORCE_INLINE int taosEncodeSMqHbOneTopicBatchRsp(void** buf, const SMqHbOneTopicBatchRsp* pBatchRsp) {
  int tlen = 0;
  tlen += taosEncodeString(buf, pBatchRsp->topicName);
  int32_t sz = taosArrayGetSize(pBatchRsp->rsps);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    SMqHbRsp* pRsp = (SMqHbRsp*)taosArrayGet(pBatchRsp->rsps, i);
    tlen += taosEncodeSMqHbRsp(buf, pRsp);
  }
  return tlen;
}

static FORCE_INLINE void* taosDecodeSMqHbOneTopicBatchRsp(void* buf, SMqHbOneTopicBatchRsp* pBatchRsp) {
  int32_t sz;
  buf = taosDecodeStringTo(buf, pBatchRsp->topicName);
  buf = taosDecodeFixedI32(buf, &sz);
  pBatchRsp->rsps = taosArrayInit(sz, sizeof(SMqHbRsp));
  for (int32_t i = 0; i < sz; i++) {
    SMqHbRsp rsp;
    buf = taosDecodeSMqHbRsp(buf, &rsp);
    buf = taosArrayPush(pBatchRsp->rsps, &rsp);
  }
  return buf;
}

typedef struct SMqHbBatchRsp {
  int64_t consumerId;
  SArray* batchRsps;  // SArray<SMqHbOneTopicBatchRsp>
} SMqHbBatchRsp;

static FORCE_INLINE int taosEncodeSMqHbBatchRsp(void** buf, const SMqHbBatchRsp* pBatchRsp) {
  int tlen = 0;
  tlen += taosEncodeFixedI64(buf, pBatchRsp->consumerId);
  int32_t sz;
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    SMqHbOneTopicBatchRsp* pRsp = (SMqHbOneTopicBatchRsp*)taosArrayGet(pBatchRsp->batchRsps, i);
    tlen += taosEncodeSMqHbOneTopicBatchRsp(buf, pRsp);
  }
  return tlen;
}

static FORCE_INLINE void* taosDecodeSMqHbBatchRsp(void* buf, SMqHbBatchRsp* pBatchRsp) {
  buf = taosDecodeFixedI64(buf, &pBatchRsp->consumerId);
  int32_t sz;
  buf = taosDecodeFixedI32(buf, &sz);
  pBatchRsp->batchRsps = taosArrayInit(sz, sizeof(SMqHbOneTopicBatchRsp));
  for (int32_t i = 0; i < sz; i++) {
    SMqHbOneTopicBatchRsp rsp;
    buf = taosDecodeSMqHbOneTopicBatchRsp(buf, &rsp);
    buf = taosArrayPush(pBatchRsp->batchRsps, &rsp);
  }
  return buf;
}

typedef struct {
  int32_t key;
  int32_t valueLen;
  void*   value;
} SKv;

typedef struct {
  int32_t connId;
  int32_t hbType;
} SClientHbKey;

typedef struct {
  SClientHbKey connKey;
  SHashObj*    info;  // hash<Skv.key, Skv>
} SClientHbReq;

typedef struct {
  int64_t reqId;
  SArray* reqs;  // SArray<SClientHbReq>
} SClientHbBatchReq;

typedef struct {
  SClientHbKey connKey;
  int32_t      status;
  SArray*      info;  // Array<Skv>
} SClientHbRsp;

typedef struct {
  int64_t reqId;
  int64_t rspId;
  SArray* rsps;  // SArray<SClientHbRsp>
} SClientHbBatchRsp;

static FORCE_INLINE uint32_t hbKeyHashFunc(const char* key, uint32_t keyLen) { return taosIntHash_64(key, keyLen); }

static FORCE_INLINE void tFreeReqKvHash(SHashObj* info) {
  void* pIter = taosHashIterate(info, NULL);
  while (pIter != NULL) {
    SKv* kv = (SKv*)pIter;
    tfree(kv->value);
    pIter = taosHashIterate(info, pIter);
  }
}

static FORCE_INLINE void tFreeClientHbReq(void* pReq) {
  SClientHbReq* req = (SClientHbReq*)pReq;
  if (req->info) {
    tFreeReqKvHash(req->info);
    taosHashCleanup(req->info);
  }
}

int32_t tSerializeSClientHbBatchReq(void* buf, int32_t bufLen, const SClientHbBatchReq* pReq);
int32_t tDeserializeSClientHbBatchReq(void* buf, int32_t bufLen, SClientHbBatchReq* pReq);

static FORCE_INLINE void tFreeClientHbBatchReq(void* pReq, bool deep) {
  SClientHbBatchReq* req = (SClientHbBatchReq*)pReq;
  if (deep) {
    taosArrayDestroyEx(req->reqs, tFreeClientHbReq);
  } else {
    taosArrayDestroy(req->reqs);
  }
  free(pReq);
}

static FORCE_INLINE void tFreeClientKv(void* pKv) {
  SKv* kv = (SKv*)pKv;
  if (kv) {
    tfree(kv->value);
  }
}

static FORCE_INLINE void tFreeClientHbRsp(void* pRsp) {
  SClientHbRsp* rsp = (SClientHbRsp*)pRsp;
  if (rsp->info) taosArrayDestroyEx(rsp->info, tFreeClientKv);
}

static FORCE_INLINE void tFreeClientHbBatchRsp(void* pRsp) {
  SClientHbBatchRsp* rsp = (SClientHbBatchRsp*)pRsp;
  taosArrayDestroyEx(rsp->rsps, tFreeClientHbRsp);
}

int32_t tSerializeSClientHbBatchRsp(void* buf, int32_t bufLen, const SClientHbBatchRsp* pBatchRsp);
int32_t tDeserializeSClientHbBatchRsp(void* buf, int32_t bufLen, SClientHbBatchRsp* pBatchRsp);

static FORCE_INLINE int32_t tEncodeSKv(SCoder* pEncoder, const SKv* pKv) {
  if (tEncodeI32(pEncoder, pKv->key) < 0) return -1;
  if (tEncodeI32(pEncoder, pKv->valueLen) < 0) return -1;
  if (tEncodeBinary(pEncoder, (const char*)pKv->value, pKv->valueLen) < 0) return -1;
  return 0;
}

static FORCE_INLINE int32_t tDecodeSKv(SCoder* pDecoder, SKv* pKv) {
  if (tDecodeI32(pDecoder, &pKv->key) < 0) return -1;
  if (tDecodeI32(pDecoder, &pKv->valueLen) < 0) return -1;
  pKv->value = malloc(pKv->valueLen + 1);
  if (pKv->value == NULL) return -1;
  if (tDecodeCStrTo(pDecoder, (char*)pKv->value) < 0) return -1;
  return 0;
}

static FORCE_INLINE int32_t tEncodeSClientHbKey(SCoder* pEncoder, const SClientHbKey* pKey) {
  if (tEncodeI32(pEncoder, pKey->connId) < 0) return -1;
  if (tEncodeI32(pEncoder, pKey->hbType) < 0) return -1;
  return 0;
}

static FORCE_INLINE int32_t tDecodeSClientHbKey(SCoder* pDecoder, SClientHbKey* pKey) {
  if (tDecodeI32(pDecoder, &pKey->connId) < 0) return -1;
  if (tDecodeI32(pDecoder, &pKey->hbType) < 0) return -1;
  return 0;
}

typedef struct SMqHbVgInfo {
  int32_t vgId;
} SMqHbVgInfo;

static FORCE_INLINE int taosEncodeSMqVgInfo(void** buf, const SMqHbVgInfo* pVgInfo) {
  int tlen = 0;
  tlen += taosEncodeFixedI32(buf, pVgInfo->vgId);
  return tlen;
}

static FORCE_INLINE void* taosDecodeSMqVgInfo(void* buf, SMqHbVgInfo* pVgInfo) {
  buf = taosDecodeFixedI32(buf, &pVgInfo->vgId);
  return buf;
}

typedef struct SMqHbTopicInfo {
  int32_t epoch;
  int64_t topicUid;
  char    name[TSDB_TOPIC_FNAME_LEN];
  SArray* pVgInfo;
} SMqHbTopicInfo;

static FORCE_INLINE int taosEncodeSMqHbTopicInfoMsg(void** buf, const SMqHbTopicInfo* pTopicInfo) {
  int tlen = 0;
  tlen += taosEncodeFixedI32(buf, pTopicInfo->epoch);
  tlen += taosEncodeFixedI64(buf, pTopicInfo->topicUid);
  tlen += taosEncodeString(buf, pTopicInfo->name);
  int32_t sz = taosArrayGetSize(pTopicInfo->pVgInfo);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    SMqHbVgInfo* pVgInfo = (SMqHbVgInfo*)taosArrayGet(pTopicInfo->pVgInfo, i);
    tlen += taosEncodeSMqVgInfo(buf, pVgInfo);
  }
  return tlen;
}

static FORCE_INLINE void* taosDecodeSMqHbTopicInfoMsg(void* buf, SMqHbTopicInfo* pTopicInfo) {
  buf = taosDecodeFixedI32(buf, &pTopicInfo->epoch);
  buf = taosDecodeFixedI64(buf, &pTopicInfo->topicUid);
  buf = taosDecodeStringTo(buf, pTopicInfo->name);
  int32_t sz;
  buf = taosDecodeFixedI32(buf, &sz);
  pTopicInfo->pVgInfo = taosArrayInit(sz, sizeof(SMqHbVgInfo));
  for (int32_t i = 0; i < sz; i++) {
    SMqHbVgInfo vgInfo;
    buf = taosDecodeSMqVgInfo(buf, &vgInfo);
    taosArrayPush(pTopicInfo->pVgInfo, &vgInfo);
  }
  return buf;
}

typedef struct SMqHbMsg {
  int32_t status;  // ask hb endpoint
  int32_t epoch;
  int64_t consumerId;
  SArray* pTopics;  // SArray<SMqHbTopicInfo>
} SMqHbMsg;

static FORCE_INLINE int taosEncodeSMqMsg(void** buf, const SMqHbMsg* pMsg) {
  int tlen = 0;
  tlen += taosEncodeFixedI32(buf, pMsg->status);
  tlen += taosEncodeFixedI32(buf, pMsg->epoch);
  tlen += taosEncodeFixedI64(buf, pMsg->consumerId);
  int32_t sz = taosArrayGetSize(pMsg->pTopics);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int i = 0; i < sz; i++) {
    SMqHbTopicInfo* topicInfo = (SMqHbTopicInfo*)taosArrayGet(pMsg->pTopics, i);
    tlen += taosEncodeSMqHbTopicInfoMsg(buf, topicInfo);
  }
  return tlen;
}

static FORCE_INLINE void* taosDecodeSMqMsg(void* buf, SMqHbMsg* pMsg) {
  buf = taosDecodeFixedI32(buf, &pMsg->status);
  buf = taosDecodeFixedI32(buf, &pMsg->epoch);
  buf = taosDecodeFixedI64(buf, &pMsg->consumerId);
  int32_t sz;
  buf = taosDecodeFixedI32(buf, &sz);
  pMsg->pTopics = taosArrayInit(sz, sizeof(SMqHbTopicInfo));
  for (int i = 0; i < sz; i++) {
    SMqHbTopicInfo topicInfo;
    buf = taosDecodeSMqHbTopicInfoMsg(buf, &topicInfo);
    taosArrayPush(pMsg->pTopics, &topicInfo);
  }
  return buf;
}

typedef struct {
  int64_t leftForVer;
  int32_t vgId;
  int64_t consumerId;
  char    topicName[TSDB_TOPIC_FNAME_LEN];
  char    cgroup[TSDB_CONSUMER_GROUP_LEN];
  char*   sql;
  char*   logicalPlan;
  char*   physicalPlan;
  char*   qmsg;
} SMqSetCVgReq;

static FORCE_INLINE int32_t tEncodeSMqSetCVgReq(void** buf, const SMqSetCVgReq* pReq) {
  int32_t tlen = 0;
  tlen += taosEncodeFixedI64(buf, pReq->leftForVer);
  tlen += taosEncodeFixedI32(buf, pReq->vgId);
  tlen += taosEncodeFixedI64(buf, pReq->consumerId);
  tlen += taosEncodeString(buf, pReq->topicName);
  tlen += taosEncodeString(buf, pReq->cgroup);
  tlen += taosEncodeString(buf, pReq->sql);
  tlen += taosEncodeString(buf, pReq->logicalPlan);
  tlen += taosEncodeString(buf, pReq->physicalPlan);
  tlen += taosEncodeString(buf, pReq->qmsg);
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqSetCVgReq(void* buf, SMqSetCVgReq* pReq) {
  buf = taosDecodeFixedI64(buf, &pReq->leftForVer);
  buf = taosDecodeFixedI32(buf, &pReq->vgId);
  buf = taosDecodeFixedI64(buf, &pReq->consumerId);
  buf = taosDecodeStringTo(buf, pReq->topicName);
  buf = taosDecodeStringTo(buf, pReq->cgroup);
  buf = taosDecodeString(buf, &pReq->sql);
  buf = taosDecodeString(buf, &pReq->logicalPlan);
  buf = taosDecodeString(buf, &pReq->physicalPlan);
  buf = taosDecodeString(buf, &pReq->qmsg);
  return buf;
}

typedef struct {
  int64_t leftForVer;
  int32_t vgId;
  int64_t oldConsumerId;
  int64_t newConsumerId;
  // char     topicName[TSDB_TOPIC_FNAME_LEN];
  // char     cgroup[TSDB_CONSUMER_GROUP_LEN];
  // char*    sql;
  // char*    logicalPlan;
  // char*    physicalPlan;
  // char*    qmsg;
} SMqMVRebReq;

static FORCE_INLINE int32_t tEncodeSMqMVRebReq(void** buf, const SMqMVRebReq* pReq) {
  int32_t tlen = 0;
  tlen += taosEncodeFixedI64(buf, pReq->leftForVer);
  tlen += taosEncodeFixedI32(buf, pReq->vgId);
  tlen += taosEncodeFixedI64(buf, pReq->oldConsumerId);
  tlen += taosEncodeFixedI64(buf, pReq->newConsumerId);
  // tlen += taosEncodeString(buf, pReq->topicName);
  // tlen += taosEncodeString(buf, pReq->cgroup);
  // tlen += taosEncodeString(buf, pReq->sql);
  // tlen += taosEncodeString(buf, pReq->logicalPlan);
  // tlen += taosEncodeString(buf, pReq->physicalPlan);
  // tlen += taosEncodeString(buf, pReq->qmsg);
  // tlen += tEncodeSSubQueryMsg(buf, &pReq->msg);
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqMVRebReq(void* buf, SMqMVRebReq* pReq) {
  buf = taosDecodeFixedI64(buf, &pReq->leftForVer);
  buf = taosDecodeFixedI32(buf, &pReq->vgId);
  buf = taosDecodeFixedI64(buf, &pReq->oldConsumerId);
  buf = taosDecodeFixedI64(buf, &pReq->newConsumerId);
  // buf = taosDecodeStringTo(buf, pReq->topicName);
  // buf = taosDecodeStringTo(buf, pReq->cgroup);
  // buf = taosDecodeString(buf, &pReq->sql);
  // buf = taosDecodeString(buf, &pReq->logicalPlan);
  // buf = taosDecodeString(buf, &pReq->physicalPlan);
  // buf = taosDecodeString(buf, &pReq->qmsg);
  // buf = tDecodeSSubQueryMsg(buf, &pReq->msg);
  return buf;
}

typedef struct {
  SMsgHead header;
  int32_t  vgId;
  int64_t  consumerId;
  char     topicName[TSDB_TOPIC_FNAME_LEN];
  char     cGroup[TSDB_CONSUMER_GROUP_LEN];
} SMqSetCVgRsp;

typedef struct {
  SMsgHead header;
  int32_t  vgId;
  int64_t  consumerId;
  char     topicName[TSDB_TOPIC_FNAME_LEN];
  char     cGroup[TSDB_CONSUMER_GROUP_LEN];
} SMqMVRebRsp;

typedef struct {
  uint32_t nCols;
  SSchema* pSchema;
} SSchemaWrapper;

static FORCE_INLINE int32_t taosEncodeSSchema(void** buf, const SSchema* pSchema) {
  int32_t tlen = 0;
  tlen += taosEncodeFixedI8(buf, pSchema->type);
  tlen += taosEncodeFixedI32(buf, pSchema->bytes);
  tlen += taosEncodeFixedI32(buf, pSchema->colId);
  tlen += taosEncodeString(buf, pSchema->name);
  return tlen;
}

static FORCE_INLINE void* taosDecodeSSchema(void* buf, SSchema* pSchema) {
  buf = taosDecodeFixedI8(buf, &pSchema->type);
  buf = taosDecodeFixedI32(buf, &pSchema->bytes);
  buf = taosDecodeFixedI32(buf, &pSchema->colId);
  buf = taosDecodeStringTo(buf, pSchema->name);
  return buf;
}

static FORCE_INLINE int32_t tEncodeSSchema(SCoder* pEncoder, const SSchema* pSchema) {
  if (tEncodeI8(pEncoder, pSchema->type) < 0) return -1;
  if (tEncodeI32(pEncoder, pSchema->bytes) < 0) return -1;
  if (tEncodeI32(pEncoder, pSchema->colId) < 0) return -1;
  if (tEncodeCStr(pEncoder, pSchema->name) < 0) return -1;
  return 0;
}

static FORCE_INLINE int32_t tDecodeSSchema(SCoder* pDecoder, SSchema* pSchema) {
  if (tDecodeI8(pDecoder, &pSchema->type) < 0) return -1;
  if (tDecodeI32(pDecoder, &pSchema->bytes) < 0) return -1;
  if (tDecodeI32(pDecoder, &pSchema->colId) < 0) return -1;
  if (tDecodeCStrTo(pDecoder, pSchema->name) < 0) return -1;
  return 0;
}

static FORCE_INLINE int32_t tEncodeSSchemaWrapper(void** buf, const SSchemaWrapper* pSW) {
  int32_t tlen = 0;
  tlen += taosEncodeFixedU32(buf, pSW->nCols);
  for (int32_t i = 0; i < pSW->nCols; i++) {
    tlen += taosEncodeSSchema(buf, &pSW->pSchema[i]);
  }
  return tlen;
}

static FORCE_INLINE void* tDecodeSSchemaWrapper(void* buf, SSchemaWrapper* pSW) {
  buf = taosDecodeFixedU32(buf, &pSW->nCols);
  pSW->pSchema = (SSchema*)calloc(pSW->nCols, sizeof(SSchema));
  if (pSW->pSchema == NULL) {
    return NULL;
  }

  for (int32_t i = 0; i < pSW->nCols; i++) {
    buf = taosDecodeSSchema(buf, &pSW->pSchema[i]);
  }
  return buf;
}

typedef struct {
  int64_t uid;
  int32_t numOfRows;
  char*   colData;
} SMqTbData;

typedef struct {
  char       topicName[TSDB_TOPIC_FNAME_LEN];
  int64_t    committedOffset;
  int64_t    reqOffset;
  int64_t    rspOffset;
  int32_t    skipLogNum;
  int32_t    bodyLen;
  int32_t    numOfTb;
  SMqTbData* tbData;
} SMqTopicData;

typedef struct {
  int64_t         consumerId;
  SSchemaWrapper* schemas;
  int64_t         committedOffset;
  int64_t         reqOffset;
  int64_t         rspOffset;
  int32_t         skipLogNum;
  int32_t         numOfTopics;
  SArray*         pBlockData;  // SArray<SSDataBlock>
} SMqConsumeRsp;

// one req for one vg+topic
typedef struct {
  SMsgHead head;
  // 0: commit only, current offset
  // 1: consume only, poll next offset
  // 2: commit current and consume next offset
  int32_t reqType;

  int64_t reqId;
  int64_t consumerId;
  int64_t blockingTime;
  char    cgroup[TSDB_CONSUMER_GROUP_LEN];

  int64_t offset;
  char    topic[TSDB_TOPIC_FNAME_LEN];
} SMqConsumeReq;

typedef struct {
  int32_t vgId;
  SEpSet  epSet;
} SMqSubVgEp;

typedef struct {
  char    topic[TSDB_TOPIC_FNAME_LEN];
  SArray* vgs;  // SArray<SMqSubVgEp>
} SMqSubTopicEp;

typedef struct {
  int64_t consumerId;
  int32_t epoch;
  char    cgroup[TSDB_CONSUMER_GROUP_LEN];
  SArray* topics;  // SArray<SMqSubTopicEp>
} SMqCMGetSubEpRsp;

static FORCE_INLINE void tDeleteSMqSubTopicEp(SMqSubTopicEp* pSubTopicEp) { taosArrayDestroy(pSubTopicEp->vgs); }

static FORCE_INLINE int32_t tEncodeSMqSubVgEp(void** buf, const SMqSubVgEp* pVgEp) {
  int32_t tlen = 0;
  tlen += taosEncodeFixedI32(buf, pVgEp->vgId);
  tlen += taosEncodeSEpSet(buf, &pVgEp->epSet);
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqSubVgEp(void* buf, SMqSubVgEp* pVgEp) {
  buf = taosDecodeFixedI32(buf, &pVgEp->vgId);
  buf = taosDecodeSEpSet(buf, &pVgEp->epSet);
  return buf;
}

static FORCE_INLINE void tDeleteSMqCMGetSubEpRsp(SMqCMGetSubEpRsp* pRsp) {
  taosArrayDestroyEx(pRsp->topics, (void (*)(void*))tDeleteSMqSubTopicEp);
}

static FORCE_INLINE int32_t tEncodeSMqSubTopicEp(void** buf, const SMqSubTopicEp* pTopicEp) {
  int32_t tlen = 0;
  tlen += taosEncodeString(buf, pTopicEp->topic);
  int32_t sz = taosArrayGetSize(pTopicEp->vgs);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    SMqSubVgEp* pVgEp = (SMqSubVgEp*)taosArrayGet(pTopicEp->vgs, i);
    tlen += tEncodeSMqSubVgEp(buf, pVgEp);
  }
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqSubTopicEp(void* buf, SMqSubTopicEp* pTopicEp) {
  buf = taosDecodeStringTo(buf, pTopicEp->topic);
  int32_t sz;
  buf = taosDecodeFixedI32(buf, &sz);
  pTopicEp->vgs = taosArrayInit(sz, sizeof(SMqSubVgEp));
  if (pTopicEp->vgs == NULL) {
    return NULL;
  }
  for (int32_t i = 0; i < sz; i++) {
    SMqSubVgEp vgEp;
    buf = tDecodeSMqSubVgEp(buf, &vgEp);
    taosArrayPush(pTopicEp->vgs, &vgEp);
  }
  return buf;
}

static FORCE_INLINE int32_t tEncodeSMqCMGetSubEpRsp(void** buf, const SMqCMGetSubEpRsp* pRsp) {
  int32_t tlen = 0;
  tlen += taosEncodeFixedI64(buf, pRsp->consumerId);
  tlen += taosEncodeFixedI32(buf, pRsp->epoch);
  tlen += taosEncodeString(buf, pRsp->cgroup);
  int32_t sz = taosArrayGetSize(pRsp->topics);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    SMqSubTopicEp* pVgEp = (SMqSubTopicEp*)taosArrayGet(pRsp->topics, i);
    tlen += tEncodeSMqSubTopicEp(buf, pVgEp);
  }
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqCMGetSubEpRsp(void* buf, SMqCMGetSubEpRsp* pRsp) {
  buf = taosDecodeFixedI64(buf, &pRsp->consumerId);
  buf = taosDecodeFixedI32(buf, &pRsp->epoch);
  buf = taosDecodeStringTo(buf, pRsp->cgroup);
  int32_t sz;
  buf = taosDecodeFixedI32(buf, &sz);
  pRsp->topics = taosArrayInit(sz, sizeof(SMqSubTopicEp));
  if (pRsp->topics == NULL) {
    return NULL;
  }
  for (int32_t i = 0; i < sz; i++) {
    SMqSubTopicEp topicEp;
    buf = tDecodeSMqSubTopicEp(buf, &topicEp);
    taosArrayPush(pRsp->topics, &topicEp);
  }
  return buf;
}

#ifdef __cplusplus
}
#endif

#endif /*_TD_COMMON_TAOS_MSG_H_*/