tmsg.h 63.4 KB
Newer Older
D
dapan1121 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
/*
 * 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_

#include "taosdef.h"
#include "taoserror.h"
H
more  
Hongze Cheng 已提交
21
#include "tarray.h"
H
more  
Hongze Cheng 已提交
22
#include "tcoding.h"
S
Shengliang Guan 已提交
23
#include "tencode.h"
L
Liu Jicong 已提交
24
#include "thash.h"
H
Hongze Cheng 已提交
25
#include "tlist.h"
26
#include "trow.h"
L
Liu Jicong 已提交
27
#include "tuuid.h"
D
dapan1121 已提交
28

S
Shengliang Guan 已提交
29 30 31 32
#ifdef __cplusplus
extern "C" {
#endif

H
Hongze Cheng 已提交
33
/* ------------------------ MESSAGE DEFINITIONS ------------------------ */
H
Hongze Cheng 已提交
34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51
#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"

S
Shengliang Guan 已提交
52 53
extern char*   tMsgInfo[];
extern int32_t tMsgDict[];
H
Hongze Cheng 已提交
54 55

#define TMSG_SEG_CODE(TYPE) (((TYPE)&0xff00) >> 8)
L
Liu Jicong 已提交
56 57 58
#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))
D
dapan1121 已提交
59

H
Hongze Cheng 已提交
60 61
typedef uint16_t tmsg_t;

H
Hongze Cheng 已提交
62
/* ------------------------ OTHER DEFINITIONS ------------------------ */
D
dapan1121 已提交
63
// IE type
L
Liu Jicong 已提交
64 65 66 67
#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
D
dapan1121 已提交
68
#define TSDB_IE_TYPE_NEW_VERSION 5
L
Liu Jicong 已提交
69
#define TSDB_IE_TYPE_DNODE_EXT   6
D
dapan1121 已提交
70 71
#define TSDB_IE_TYPE_DNODE_STATE 7

L
Liu Jicong 已提交
72
typedef enum {
73
  HEARTBEAT_TYPE_MQ = 0,
L
Liu Jicong 已提交
74
  HEARTBEAT_TYPE_QUERY,
L
Liu Jicong 已提交
75 76 77 78 79
  // types can be added here
  //
  HEARTBEAT_TYPE_MAX
} EHbType;

D
dapan1121 已提交
80
enum {
D
dapan1121 已提交
81 82
  HEARTBEAT_KEY_DBINFO = 1,
  HEARTBEAT_KEY_STBINFO,
D
dapan1121 已提交
83
  HEARTBEAT_KEY_MQ_TMP,
D
dapan1121 已提交
84 85
};

S
Shengliang Guan 已提交
86 87
typedef enum _mgmt_table {
  TSDB_MGMT_TABLE_START,
D
dapan1121 已提交
88 89 90 91 92 93
  TSDB_MGMT_TABLE_ACCT,
  TSDB_MGMT_TABLE_USER,
  TSDB_MGMT_TABLE_DB,
  TSDB_MGMT_TABLE_TABLE,
  TSDB_MGMT_TABLE_DNODE,
  TSDB_MGMT_TABLE_MNODE,
S
Shengliang Guan 已提交
94 95 96
  TSDB_MGMT_TABLE_QNODE,
  TSDB_MGMT_TABLE_SNODE,
  TSDB_MGMT_TABLE_BNODE,
D
dapan1121 已提交
97
  TSDB_MGMT_TABLE_VGROUP,
S
Shengliang Guan 已提交
98
  TSDB_MGMT_TABLE_STB,
D
dapan1121 已提交
99 100 101 102 103
  TSDB_MGMT_TABLE_MODULE,
  TSDB_MGMT_TABLE_QUERIES,
  TSDB_MGMT_TABLE_STREAMS,
  TSDB_MGMT_TABLE_VARIABLES,
  TSDB_MGMT_TABLE_CONNS,
S
Shengliang Guan 已提交
104
  TSDB_MGMT_TABLE_TRANS,
D
dapan1121 已提交
105 106 107
  TSDB_MGMT_TABLE_GRANTS,
  TSDB_MGMT_TABLE_VNODES,
  TSDB_MGMT_TABLE_CLUSTER,
S
Shengliang Guan 已提交
108
  TSDB_MGMT_TABLE_STREAMTABLES,
D
dapan1121 已提交
109
  TSDB_MGMT_TABLE_TP,
S
Shengliang 已提交
110
  TSDB_MGMT_TABLE_FUNC,
D
dapan1121 已提交
111
  TSDB_MGMT_TABLE_MAX,
S
Shengliang Guan 已提交
112
} EShowType;
D
dapan1121 已提交
113

L
Liu Jicong 已提交
114 115
#define TSDB_ALTER_TABLE_ADD_TAG         1
#define TSDB_ALTER_TABLE_DROP_TAG        2
S
Shengliang Guan 已提交
116
#define TSDB_ALTER_TABLE_UPDATE_TAG_NAME 3
L
Liu Jicong 已提交
117
#define TSDB_ALTER_TABLE_UPDATE_TAG_VAL  4
D
dapan1121 已提交
118

L
Liu Jicong 已提交
119 120
#define TSDB_ALTER_TABLE_ADD_COLUMN          5
#define TSDB_ALTER_TABLE_DROP_COLUMN         6
S
Shengliang Guan 已提交
121
#define TSDB_ALTER_TABLE_UPDATE_COLUMN_BYTES 7
L
Liu Jicong 已提交
122
#define TSDB_ALTER_TABLE_UPDATE_TAG_BYTES    8
D
dapan1121 已提交
123

L
Liu Jicong 已提交
124 125
#define TSDB_FILL_NONE      0
#define TSDB_FILL_NULL      1
H
Hongze Cheng 已提交
126
#define TSDB_FILL_SET_VALUE 2
L
Liu Jicong 已提交
127 128 129 130 131 132 133 134 135 136
#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
S
Shengliang Guan 已提交
137
#define TSDB_ALTER_USER_REMOVE_WRITE_DB 0x7
L
Liu Jicong 已提交
138
#define TSDB_ALTER_USER_CLEAR_WRITE_DB  0x8
S
Shengliang Guan 已提交
139

D
dapan1121 已提交
140 141
#define TSDB_ALTER_USER_PRIVILEGES 0x2

H
Hongze Cheng 已提交
142
#define TSDB_KILL_MSG_LEN 30
D
dapan1121 已提交
143

D
dapan1121 已提交
144 145
#define TSDB_TABLE_NUM_UNIT 100000

L
Liu Jicong 已提交
146
#define TSDB_VN_READ_ACCCESS  ((char)0x1)
H
Hongze Cheng 已提交
147
#define TSDB_VN_WRITE_ACCCESS ((char)0x2)
L
Liu Jicong 已提交
148
#define TSDB_VN_ALL_ACCCESS   (TSDB_VN_READ_ACCCESS | TSDB_VN_WRITE_ACCCESS)
D
dapan1121 已提交
149

H
Hongze Cheng 已提交
150
#define TSDB_COL_NORMAL 0x0u  // the normal column of the table
L
Liu Jicong 已提交
151 152 153 154
#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
D
dapan1121 已提交
155

L
Liu Jicong 已提交
156
#define TSDB_COL_IS_TAG(f)        (((f & (~(TSDB_COL_NULL))) & TSDB_COL_TAG) != 0)
H
Hongze Cheng 已提交
157
#define TSDB_COL_IS_NORMAL_COL(f) ((f & (~(TSDB_COL_NULL))) == TSDB_COL_NORMAL)
L
Liu Jicong 已提交
158 159
#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)
S
Shengliang Guan 已提交
160

L
Liu Jicong 已提交
161 162
#define TD_SUPER_TABLE  TSDB_SUPER_TABLE
#define TD_CHILD_TABLE  TSDB_CHILD_TABLE
S
Shengliang Guan 已提交
163 164
#define TD_NORMAL_TABLE TSDB_NORMAL_TABLE

S
Shengliang Guan 已提交
165
typedef struct {
S
Shengliang Guan 已提交
166
  int32_t vgId;
D
dapan1121 已提交
167 168
  char*   dbFName;
  char*   tbName;
D
dapan1121 已提交
169 170
} SBuildTableMetaInput;

S
Shengliang Guan 已提交
171
typedef struct {
D
dapan1121 已提交
172
  char    db[TSDB_DB_FNAME_LEN];
D
dapan1121 已提交
173
  int64_t dbId;
174
  int32_t vgVersion;
L
Liu Jicong 已提交
175
  int32_t numOfTable;  // unit is TSDB_TABLE_NUM_UNIT
D
dapan1121 已提交
176 177
} SBuildUseDBInput;

S
Shengliang Guan 已提交
178 179 180 181 182 183
typedef struct SField {
  char    name[TSDB_COL_NAME_LEN];
  uint8_t type;
  int32_t bytes;
} SField;

D
dapan1121 已提交
184 185 186
#pragma pack(push, 1)

// null-terminated string instead of char array to avoid too many memory consumption in case of more than 1M tableMeta
H
Haojun Liao 已提交
187
typedef struct SEp {
D
dapan1121 已提交
188
  char     fqdn[TSDB_FQDN_LEN];
D
dapan1121 已提交
189
  uint16_t port;
H
Haojun Liao 已提交
190
} SEp;
D
dapan1121 已提交
191

S
Shengliang Guan 已提交
192
typedef struct {
D
dapan1121 已提交
193
  int32_t contLen;
L
Liu Jicong 已提交
194 195 196 197
  union {
    int32_t vgId;
    int32_t streamTaskId;
  };
D
dapan1121 已提交
198 199 200 201
} SMsgHead;

// Submit message for one table
typedef struct SSubmitBlk {
202 203 204 205 206 207 208 209
  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[];
D
dapan1121 已提交
210 211 212
} SSubmitBlk;

// Submit message for this TSDB
S
Shengliang Guan 已提交
213
typedef struct {
S
Shengliang Guan 已提交
214
  SMsgHead header;
H
Hongze Cheng 已提交
215
  int64_t  version;
S
Shengliang Guan 已提交
216 217 218
  int32_t  length;
  int32_t  numOfBlocks;
  char     blocks[];
S
Shengliang Guan 已提交
219
} SSubmitReq;
D
dapan1121 已提交
220

H
Hongze Cheng 已提交
221 222 223
typedef struct {
  int32_t totalLen;
  int32_t len;
224
  STSRow* row;
H
Hongze Cheng 已提交
225 226 227 228 229 230 231 232
} SSubmitBlkIter;

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

S
Shengliang Guan 已提交
233
int32_t tInitSubmitMsgIter(SSubmitReq* pMsg, SSubmitMsgIter* pIter);
S
Shengliang Guan 已提交
234 235
int32_t tGetSubmitMsgNext(SSubmitMsgIter* pIter, SSubmitBlk** pPBlock);
int32_t tInitSubmitBlkIter(SSubmitBlk* pBlock, SSubmitBlkIter* pIter);
C
Cary Xu 已提交
236
STSRow* tGetSubmitBlkNext(SSubmitBlkIter* pIter);
H
Hongze Cheng 已提交
237

D
dapan1121 已提交
238 239 240 241 242
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
S
Shengliang Guan 已提交
243
} SSubmitRspBlock;
D
dapan1121 已提交
244 245

typedef struct {
S
Shengliang Guan 已提交
246 247 248 249 250 251 252
  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;
  SSubmitRspBlock failedBlocks[];
} SSubmitRsp;
D
dapan1121 已提交
253 254

typedef struct SSchema {
S
Shengliang Guan 已提交
255
  int8_t  type;
S
Shengliang Guan 已提交
256
  int32_t colId;
257
  int32_t bytes;
S
Shengliang Guan 已提交
258
  char    name[TSDB_COL_NAME_LEN];
D
dapan1121 已提交
259 260 261
} SSchema;

typedef struct {
S
Shengliang Guan 已提交
262 263 264
  char    name[TSDB_TABLE_FNAME_LEN];
  int8_t  igExists;
  int32_t numOfColumns;
S
Shengliang Guan 已提交
265 266 267
  int32_t numOfTags;
  SArray* pColumns;
  SArray* pTags;
S
Shengliang Guan 已提交
268
  char    comment[TSDB_STB_COMMENT_LEN];
S
Shengliang Guan 已提交
269
} SMCreateStbReq;
D
dapan1121 已提交
270

S
Shengliang Guan 已提交
271 272
int32_t tSerializeSMCreateStbReq(void* buf, int32_t bufLen, SMCreateStbReq* pReq);
int32_t tDeserializeSMCreateStbReq(void* buf, int32_t bufLen, SMCreateStbReq* pReq);
S
Shengliang Guan 已提交
273
void    tFreeSMCreateStbReq(SMCreateStbReq* pReq);
S
Shengliang Guan 已提交
274

D
dapan1121 已提交
275 276 277
typedef struct {
  char   name[TSDB_TABLE_FNAME_LEN];
  int8_t igNotExists;
S
Shengliang Guan 已提交
278
} SMDropStbReq;
S
Shengliang Guan 已提交
279

S
Shengliang Guan 已提交
280 281
int32_t tSerializeSMDropStbReq(void* buf, int32_t bufLen, SMDropStbReq* pReq);
int32_t tDeserializeSMDropStbReq(void* buf, int32_t bufLen, SMDropStbReq* pReq);
S
Shengliang Guan 已提交
282

S
Shengliang Guan 已提交
283 284 285
typedef struct {
  char    name[TSDB_TABLE_FNAME_LEN];
  int8_t  alterType;
S
Shengliang Guan 已提交
286 287
  int32_t numOfFields;
  SArray* pFields;
S
Shengliang Guan 已提交
288
} SMAltertbReq;
D
dapan1121 已提交
289

S
Shengliang Guan 已提交
290 291 292
int32_t tSerializeSMAlterStbReq(void* buf, int32_t bufLen, SMAltertbReq* pReq);
int32_t tDeserializeSMAlterStbReq(void* buf, int32_t bufLen, SMAltertbReq* pReq);
void    tFreeSMAltertbReq(SMAltertbReq* pReq);
D
dapan1121 已提交
293 294

typedef struct SEpSet {
295 296 297
  int8_t inUse;
  int8_t numOfEps;
  SEp    eps[TSDB_MAX_REPLICA];
D
dapan1121 已提交
298 299
} SEpSet;

S
Shengliang Guan 已提交
300 301 302 303
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);
L
Liu Jicong 已提交
304

D
dapan1121 已提交
305 306
typedef struct {
  int32_t pid;
S
Shengliang Guan 已提交
307 308
  char    app[TSDB_APP_NAME_LEN];
  char    db[TSDB_DB_NAME_LEN];
309
  int64_t startTime;
S
Shengliang Guan 已提交
310
} SConnectReq;
L
Liu Jicong 已提交
311

S
Shengliang Guan 已提交
312 313
int32_t tSerializeSConnectReq(void* buf, int32_t bufLen, SConnectReq* pReq);
int32_t tDeserializeSConnectReq(void* buf, int32_t bufLen, SConnectReq* pReq);
314

L
Liu Jicong 已提交
315 316 317 318 319 320
typedef struct {
  int32_t acctId;
  int64_t clusterId;
  int32_t connId;
  int8_t  superUser;
  SEpSet  epSet;
321
  char    sVersion[128];
L
Liu Jicong 已提交
322
} SConnectRsp;
L
Liu Jicong 已提交
323

S
Shengliang Guan 已提交
324 325 326
int32_t tSerializeSConnectRsp(void* buf, int32_t bufLen, SConnectRsp* pRsp);
int32_t tDeserializeSConnectRsp(void* buf, int32_t bufLen, SConnectRsp* pRsp);

L
Liu Jicong 已提交
327 328 329 330 331 332 333 334 335 336
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;
L
Liu Jicong 已提交
337

S
Shengliang Guan 已提交
338 339
int32_t tSerializeSCreateAcctReq(void* buf, int32_t bufLen, SCreateAcctReq* pReq);
int32_t tDeserializeSCreateAcctReq(void* buf, int32_t bufLen, SCreateAcctReq* pReq);
S
Shengliang Guan 已提交
340

L
Liu Jicong 已提交
341 342 343
typedef struct {
  char user[TSDB_USER_LEN];
} SDropUserReq, SDropAcctReq;
D
dapan1121 已提交
344

S
Shengliang Guan 已提交
345 346
int32_t tSerializeSDropUserReq(void* buf, int32_t bufLen, SDropUserReq* pReq);
int32_t tDeserializeSDropUserReq(void* buf, int32_t bufLen, SDropUserReq* pReq);
S
Shengliang Guan 已提交
347

D
dapan1121 已提交
348
typedef struct {
S
Shengliang Guan 已提交
349 350
  int8_t createType;
  int8_t superUser;  // denote if it is a super user or not
S
Shengliang Guan 已提交
351 352
  char   user[TSDB_USER_LEN];
  char   pass[TSDB_PASSWORD_LEN];
S
Shengliang Guan 已提交
353 354
} SCreateUserReq;

S
Shengliang Guan 已提交
355 356
int32_t tSerializeSCreateUserReq(void* buf, int32_t bufLen, SCreateUserReq* pReq);
int32_t tDeserializeSCreateUserReq(void* buf, int32_t bufLen, SCreateUserReq* pReq);
S
Shengliang Guan 已提交
357 358 359

typedef struct {
  int8_t alterType;
S
Shengliang Guan 已提交
360
  int8_t superUser;
S
Shengliang Guan 已提交
361 362 363 364 365
  char   user[TSDB_USER_LEN];
  char   pass[TSDB_PASSWORD_LEN];
  char   dbname[TSDB_DB_FNAME_LEN];
} SAlterUserReq;

S
Shengliang Guan 已提交
366 367
int32_t tSerializeSAlterUserReq(void* buf, int32_t bufLen, SAlterUserReq* pReq);
int32_t tDeserializeSAlterUserReq(void* buf, int32_t bufLen, SAlterUserReq* pReq);
D
dapan1121 已提交
368

S
Shengliang Guan 已提交
369 370 371 372
typedef struct {
  char user[TSDB_USER_LEN];
} SGetUserAuthReq;

S
Shengliang Guan 已提交
373 374
int32_t tSerializeSGetUserAuthReq(void* buf, int32_t bufLen, SGetUserAuthReq* pReq);
int32_t tDeserializeSGetUserAuthReq(void* buf, int32_t bufLen, SGetUserAuthReq* pReq);
S
Shengliang Guan 已提交
375 376 377 378 379 380 381 382

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

S
Shengliang Guan 已提交
383 384
int32_t tSerializeSGetUserAuthRsp(void* buf, int32_t bufLen, SGetUserAuthRsp* pRsp);
int32_t tDeserializeSGetUserAuthRsp(void* buf, int32_t bufLen, SGetUserAuthRsp* pRsp);
S
Shengliang Guan 已提交
385

D
dapan1121 已提交
386
typedef struct {
S
Shengliang Guan 已提交
387 388 389
  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
390
  char    name[TSDB_DB_FNAME_LEN];
D
dapan1121 已提交
391 392
} SColIndex;

S
Shengliang Guan 已提交
393
typedef struct {
D
dapan1121 已提交
394 395
  int16_t lowerRelOptr;
  int16_t upperRelOptr;
S
Shengliang Guan 已提交
396
  int16_t filterstr;  // denote if current column is char(binary/nchar)
D
dapan1121 已提交
397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413

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

S
Shengliang Guan 已提交
414
typedef struct {
S
Shengliang Guan 已提交
415 416 417
  int16_t numOfFilters;
  union {
    int64_t            placeholder;
H
Hongze Cheng 已提交
418
    SColumnFilterInfo* filterInfo;
D
dapan1121 已提交
419 420 421
  };
} SColumnFilterList;
/*
L
Liu Jicong 已提交
422
 * for client side struct, only column id, type, bytes are necessary
D
dapan1121 已提交
423 424
 * But for data in vnode side, we need all the following information.
 */
S
Shengliang Guan 已提交
425
typedef struct {
H
Haojun Liao 已提交
426 427 428 429 430
  union {
    int16_t colId;
    int16_t slotId;
  };

L
Liu Jicong 已提交
431 432 433 434
  int16_t type;
  int32_t bytes;
  uint8_t precision;
  uint8_t scale;
D
dapan1121 已提交
435 436
} SColumnInfo;

S
Shengliang Guan 已提交
437
typedef struct {
L
Liu Jicong 已提交
438 439
  int64_t uid;
  TSKEY   key;  // last accessed ts, for subscription
D
dapan1121 已提交
440 441 442 443 444 445 446 447
} STableIdInfo;

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

typedef struct {
S
Shengliang Guan 已提交
448 449 450 451
  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
D
dapan1121 已提交
452 453
} STsBufInfo;

S
Shengliang Guan 已提交
454
typedef struct {
S
Shengliang Guan 已提交
455
  int32_t tz;  // query client timezone
D
dapan1121 已提交
456 457 458 459 460 461 462 463 464
  char    intervalUnit;
  char    slidingUnit;
  char    offsetUnit;
  int64_t interval;
  int64_t sliding;
  int64_t offset;
} SInterval;

typedef struct {
S
Shengliang Guan 已提交
465
  int32_t code;
D
dapan1121 已提交
466 467 468
} SQueryTableRsp;

typedef struct {
469
  char    db[TSDB_DB_FNAME_LEN];
470
  int32_t numOfVgroups;
S
Shengliang Guan 已提交
471 472 473 474 475 476
  int32_t cacheBlockSize;  // MB
  int32_t totalBlocks;
  int32_t daysPerFile;
  int32_t daysToKeep0;
  int32_t daysToKeep1;
  int32_t daysToKeep2;
S
Shengliang Guan 已提交
477 478
  int32_t minRows;
  int32_t maxRows;
S
Shengliang Guan 已提交
479 480
  int32_t commitTime;
  int32_t fsyncPeriod;
S
Shengliang Guan 已提交
481
  int8_t  walLevel;
S
Shengliang Guan 已提交
482 483 484 485 486 487
  int8_t  precision;  // time resolution
  int8_t  compression;
  int8_t  replications;
  int8_t  quorum;
  int8_t  update;
  int8_t  cacheLastRow;
S
Shengliang Guan 已提交
488
  int8_t  ignoreExist;
X
Xiaoyu Wang 已提交
489
  int8_t  streamMode;
S
Shengliang Guan 已提交
490
} SCreateDbReq;
S
Shengliang Guan 已提交
491

S
Shengliang Guan 已提交
492 493 494
int32_t tSerializeSCreateDbReq(void* buf, int32_t bufLen, SCreateDbReq* pReq);
int32_t tDeserializeSCreateDbReq(void* buf, int32_t bufLen, SCreateDbReq* pReq);

S
Shengliang Guan 已提交
495
typedef struct {
496
  char    db[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
497 498 499 500 501 502 503 504
  int32_t totalBlocks;
  int32_t daysToKeep0;
  int32_t daysToKeep1;
  int32_t daysToKeep2;
  int32_t fsyncPeriod;
  int8_t  walLevel;
  int8_t  quorum;
  int8_t  cacheLastRow;
S
Shengliang Guan 已提交
505
} SAlterDbReq;
S
Shengliang Guan 已提交
506

S
Shengliang Guan 已提交
507 508 509
int32_t tSerializeSAlterDbReq(void* buf, int32_t bufLen, SAlterDbReq* pReq);
int32_t tDeserializeSAlterDbReq(void* buf, int32_t bufLen, SAlterDbReq* pReq);

S
Shengliang Guan 已提交
510
typedef struct {
S
Shengliang Guan 已提交
511
  char   db[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
512
  int8_t ignoreNotExists;
S
Shengliang Guan 已提交
513
} SDropDbReq;
S
Shengliang Guan 已提交
514

S
Shengliang Guan 已提交
515 516 517
int32_t tSerializeSDropDbReq(void* buf, int32_t bufLen, SDropDbReq* pReq);
int32_t tDeserializeSDropDbReq(void* buf, int32_t bufLen, SDropDbReq* pReq);

S
Shengliang Guan 已提交
518
typedef struct {
L
Liu Jicong 已提交
519 520
  char    db[TSDB_DB_FNAME_LEN];
  int64_t uid;
S
Shengliang Guan 已提交
521 522
} SDropDbRsp;

S
Shengliang Guan 已提交
523 524 525
int32_t tSerializeSDropDbRsp(void* buf, int32_t bufLen, SDropDbRsp* pRsp);
int32_t tDeserializeSDropDbRsp(void* buf, int32_t bufLen, SDropDbRsp* pRsp);

S
Shengliang Guan 已提交
526 527
typedef struct {
  char    db[TSDB_DB_FNAME_LEN];
D
dapan1121 已提交
528
  int64_t dbId;
529
  int32_t vgVersion;
L
Liu Jicong 已提交
530
  int32_t numOfTable;  // unit is TSDB_TABLE_NUM_UNIT
S
Shengliang Guan 已提交
531
} SUseDbReq;
S
Shengliang Guan 已提交
532

S
Shengliang Guan 已提交
533 534
int32_t tSerializeSUseDbReq(void* buf, int32_t bufLen, SUseDbReq* pReq);
int32_t tDeserializeSUseDbReq(void* buf, int32_t bufLen, SUseDbReq* pReq);
S
Shengliang Guan 已提交
535 536

typedef struct {
L
Liu Jicong 已提交
537 538 539 540 541 542
  char    db[TSDB_DB_FNAME_LEN];
  int64_t uid;
  int32_t vgVersion;
  int32_t vgNum;
  int8_t  hashMethod;
  SArray* pVgroupInfos;  // Array of SVgroupInfo
S
Shengliang Guan 已提交
543 544 545 546 547 548
} 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);

D
dapan1121 已提交
549 550 551 552 553 554 555 556
typedef struct {
  int32_t rowNum;
} SQnodeListReq;

int32_t tSerializeSQnodeListReq(void* buf, int32_t bufLen, SQnodeListReq* pReq);
int32_t tDeserializeSQnodeListReq(void* buf, int32_t bufLen, SQnodeListReq* pReq);

typedef struct {
L
Liu Jicong 已提交
557
  SArray* epSetList;  // SArray<SEpSet>
D
dapan1121 已提交
558 559 560 561 562 563
} SQnodeListRsp;

int32_t tSerializeSQnodeListRsp(void* buf, int32_t bufLen, SQnodeListRsp* pRsp);
int32_t tDeserializeSQnodeListRsp(void* buf, int32_t bufLen, SQnodeListRsp* pRsp);
void    tFreeSQnodeListRsp(SQnodeListRsp* pRsp);

S
Shengliang Guan 已提交
564 565 566 567 568 569 570
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);
S
Shengliang Guan 已提交
571 572

typedef struct {
S
Shengliang Guan 已提交
573
  char db[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
574 575 576 577
} SSyncDbReq, SCompactDbReq;

int32_t tSerializeSSyncDbReq(void* buf, int32_t bufLen, SSyncDbReq* pReq);
int32_t tDeserializeSSyncDbReq(void* buf, int32_t bufLen, SSyncDbReq* pReq);
D
dapan1121 已提交
578 579

typedef struct {
S
Shengliang Guan 已提交
580
  char    name[TSDB_FUNC_NAME_LEN];
S
Shengliang 已提交
581
  int8_t  igExists;
S
Shengliang Guan 已提交
582 583
  int8_t  funcType;
  int8_t  scriptType;
S
Shengliang Guan 已提交
584
  int8_t  outputType;
S
Shengliang Guan 已提交
585
  int32_t outputLen;
S
Shengliang Guan 已提交
586
  int32_t bufSize;
S
Shengliang 已提交
587
  int64_t signature;
S
Shengliang Guan 已提交
588 589
  int32_t commentSize;
  int32_t codeSize;
S
Shengliang Guan 已提交
590 591
  char    pComment[TSDB_FUNC_COMMENT_LEN];
  char    pCode[TSDB_FUNC_CODE_LEN];
S
Shengliang Guan 已提交
592
} SCreateFuncReq;
D
dapan1121 已提交
593

S
Shengliang Guan 已提交
594 595 596
int32_t tSerializeSCreateFuncReq(void* buf, int32_t bufLen, SCreateFuncReq* pReq);
int32_t tDeserializeSCreateFuncReq(void* buf, int32_t bufLen, SCreateFuncReq* pReq);

D
dapan1121 已提交
597
typedef struct {
S
Shengliang 已提交
598 599
  char   name[TSDB_FUNC_NAME_LEN];
  int8_t igNotExists;
S
Shengliang Guan 已提交
600
} SDropFuncReq;
S
Shengliang Guan 已提交
601

S
Shengliang Guan 已提交
602 603 604
int32_t tSerializeSDropFuncReq(void* buf, int32_t bufLen, SDropFuncReq* pReq);
int32_t tDeserializeSDropFuncReq(void* buf, int32_t bufLen, SDropFuncReq* pReq);

S
Shengliang Guan 已提交
605 606
typedef struct {
  int32_t numOfFuncs;
S
Shengliang Guan 已提交
607
  SArray* pFuncNames;
S
Shengliang Guan 已提交
608
} SRetrieveFuncReq;
D
dapan1121 已提交
609

S
Shengliang Guan 已提交
610 611 612
int32_t tSerializeSRetrieveFuncReq(void* buf, int32_t bufLen, SRetrieveFuncReq* pReq);
int32_t tDeserializeSRetrieveFuncReq(void* buf, int32_t bufLen, SRetrieveFuncReq* pReq);

D
dapan1121 已提交
613 614
typedef struct {
  char    name[TSDB_FUNC_NAME_LEN];
S
Shengliang Guan 已提交
615 616 617 618
  int8_t  funcType;
  int8_t  scriptType;
  int8_t  outputType;
  int32_t outputLen;
D
dapan1121 已提交
619
  int32_t bufSize;
S
Shengliang 已提交
620
  int64_t signature;
S
Shengliang Guan 已提交
621 622
  int32_t commentSize;
  int32_t codeSize;
S
Shengliang Guan 已提交
623 624
  char    pComment[TSDB_FUNC_COMMENT_LEN];
  char    pCode[TSDB_FUNC_CODE_LEN];
S
Shengliang Guan 已提交
625
} SFuncInfo;
D
dapan1121 已提交
626 627

typedef struct {
S
Shengliang Guan 已提交
628
  int32_t numOfFuncs;
S
Shengliang Guan 已提交
629
  SArray* pFuncInfos;
S
Shengliang Guan 已提交
630
} SRetrieveFuncRsp;
D
dapan1121 已提交
631

S
Shengliang Guan 已提交
632 633 634
int32_t tSerializeSRetrieveFuncRsp(void* buf, int32_t bufLen, SRetrieveFuncRsp* pRsp);
int32_t tDeserializeSRetrieveFuncRsp(void* buf, int32_t bufLen, SRetrieveFuncRsp* pRsp);

D
dapan1121 已提交
635
typedef struct {
S
Shengliang Guan 已提交
636
  int32_t statusInterval;
S
Shengliang Guan 已提交
637
  int64_t checkTime;                  // 1970-01-01 00:00:00.000
S
Shengliang Guan 已提交
638 639 640
  char    timezone[TD_TIMEZONE_LEN];  // tsTimezone
  char    locale[TD_LOCALE_LEN];      // tsLocale
  char    charset[TD_LOCALE_LEN];     // tsCharset
S
Shengliang Guan 已提交
641
} SClusterCfg;
D
dapan1121 已提交
642

S
Shengliang Guan 已提交
643 644 645
typedef struct {
  int32_t vgId;
  int8_t  role;
S
Shengliang Guan 已提交
646 647
  int64_t numOfTables;
  int64_t numOfTimeSeries;
S
Shengliang Guan 已提交
648 649 650
  int64_t totalStorage;
  int64_t compStorage;
  int64_t pointsWritten;
S
Shengliang Guan 已提交
651 652 653 654 655
  int64_t numOfSelectReqs;
  int64_t numOfInsertReqs;
  int64_t numOfInsertSuccessReqs;
  int64_t numOfBatchInsertReqs;
  int64_t numOfBatchInsertSuccessReqs;
S
Shengliang Guan 已提交
656 657 658
} SVnodeLoad;

typedef struct {
S
Shengliang Guan 已提交
659 660
  int32_t     sver;  // software version
  int64_t     dver;  // dnode table version in sdb
S
Shengliang Guan 已提交
661
  int32_t     dnodeId;
662
  int64_t     clusterId;
S
Shengliang Guan 已提交
663 664
  int64_t     rebootTime;
  int64_t     updateTime;
S
Shengliang Guan 已提交
665 666
  int32_t     numOfCores;
  int32_t     numOfSupportVnodes;
S
Shengliang Guan 已提交
667 668
  char        dnodeEp[TSDB_EP_LEN];
  SClusterCfg clusterCfg;
S
Shengliang Guan 已提交
669
  SArray*     pVloads;  // array of SVnodeLoad
S
Shengliang Guan 已提交
670
} SStatusReq;
D
dapan1121 已提交
671

S
Shengliang Guan 已提交
672 673
int32_t tSerializeSStatusReq(void* buf, int32_t bufLen, SStatusReq* pReq);
int32_t tDeserializeSStatusReq(void* buf, int32_t bufLen, SStatusReq* pReq);
S
Shengliang Guan 已提交
674

D
dapan1121 已提交
675
typedef struct {
S
Shengliang Guan 已提交
676
  int32_t dnodeId;
677
  int64_t clusterId;
D
dapan1121 已提交
678 679 680
} SDnodeCfg;

typedef struct {
681 682 683
  int32_t id;
  int8_t  isMnode;
  SEp     ep;
D
dapan1121 已提交
684 685 686
} SDnodeEp;

typedef struct {
687
  int64_t   dver;
S
Shengliang Guan 已提交
688
  SDnodeCfg dnodeCfg;
S
Shengliang Guan 已提交
689
  SArray*   pDnodeEps;  // Array of SDnodeEp
D
dapan1121 已提交
690 691
} SStatusRsp;

S
Shengliang Guan 已提交
692 693
int32_t tSerializeSStatusRsp(void* buf, int32_t bufLen, SStatusRsp* pRsp);
int32_t tDeserializeSStatusRsp(void* buf, int32_t bufLen, SStatusRsp* pRsp);
S
Shengliang Guan 已提交
694 695

typedef struct {
S
Shengliang Guan 已提交
696 697 698 699 700
  int32_t reserved;
} SMTimerReq;

int32_t tSerializeSMTimerMsg(void* buf, int32_t bufLen, SMTimerReq* pReq);
int32_t tDeserializeSMTimerMsg(void* buf, int32_t bufLen, SMTimerReq* pReq);
S
Shengliang Guan 已提交
701

D
dapan1121 已提交
702
typedef struct {
S
Shengliang Guan 已提交
703 704 705
  int32_t  id;
  uint16_t port;                 // node sync Port
  char     fqdn[TSDB_FQDN_LEN];  // node FQDN
S
Shengliang Guan 已提交
706 707 708
} SReplica;

typedef struct {
S
Shengliang Guan 已提交
709
  int32_t  vgId;
S
Shengliang Guan 已提交
710
  int32_t  dnodeId;
711
  char     db[TSDB_DB_FNAME_LEN];
L
Liu Jicong 已提交
712
  int64_t  dbUid;
713
  int32_t  vgVersion;
S
Shengliang Guan 已提交
714 715 716 717 718 719
  int32_t  cacheBlockSize;
  int32_t  totalBlocks;
  int32_t  daysPerFile;
  int32_t  daysToKeep0;
  int32_t  daysToKeep1;
  int32_t  daysToKeep2;
S
Shengliang Guan 已提交
720 721 722
  int32_t  minRows;
  int32_t  maxRows;
  int32_t  commitTime;
S
Shengliang Guan 已提交
723
  int32_t  fsyncPeriod;
D
dapan1121 已提交
724 725
  uint32_t hashBegin;
  uint32_t hashEnd;
L
Liu Jicong 已提交
726
  int8_t   hashMethod;
S
Shengliang Guan 已提交
727
  int8_t   walLevel;
S
Shengliang Guan 已提交
728 729 730
  int8_t   precision;
  int8_t   compression;
  int8_t   quorum;
S
Shengliang Guan 已提交
731 732
  int8_t   update;
  int8_t   cacheLastRow;
S
Shengliang Guan 已提交
733
  int8_t   replica;
S
Shengliang Guan 已提交
734
  int8_t   selfIndex;
L
Liu Jicong 已提交
735
  int8_t   streamMode;
S
Shengliang Guan 已提交
736
  SReplica replicas[TSDB_MAX_REPLICA];
L
Liu Jicong 已提交
737

S
Shengliang Guan 已提交
738
} SCreateVnodeReq, SAlterVnodeReq;
D
dapan1121 已提交
739

S
Shengliang Guan 已提交
740 741 742
int32_t tSerializeSCreateVnodeReq(void* buf, int32_t bufLen, SCreateVnodeReq* pReq);
int32_t tDeserializeSCreateVnodeReq(void* buf, int32_t bufLen, SCreateVnodeReq* pReq);

S
Shengliang Guan 已提交
743
typedef struct {
L
Liu Jicong 已提交
744 745 746 747
  int32_t vgId;
  int32_t dnodeId;
  int64_t dbUid;
  char    db[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
748
} SDropVnodeReq, SSyncVnodeReq, SCompactVnodeReq;
S
Shengliang Guan 已提交
749

S
Shengliang Guan 已提交
750 751
int32_t tSerializeSDropVnodeReq(void* buf, int32_t bufLen, SDropVnodeReq* pReq);
int32_t tDeserializeSDropVnodeReq(void* buf, int32_t bufLen, SDropVnodeReq* pReq);
S
Shengliang Guan 已提交
752

D
dapan1121 已提交
753
typedef struct {
D
dapan1121 已提交
754
  SMsgHead header;
D
dapan1121 已提交
755 756
  char     dbFName[TSDB_DB_FNAME_LEN];
  char     tbName[TSDB_TABLE_NAME_LEN];
S
Shengliang Guan 已提交
757
} STableInfoReq;
D
dapan1121 已提交
758

S
Shengliang Guan 已提交
759 760 761
int32_t tSerializeSTableInfoReq(void* buf, int32_t bufLen, STableInfoReq* pReq);
int32_t tDeserializeSTableInfoReq(void* buf, int32_t bufLen, STableInfoReq* pReq);

D
dapan1121 已提交
762
typedef struct {
S
Shengliang Guan 已提交
763
  int8_t  metaClone;  // create local clone of the cached table meta
D
dapan1121 已提交
764 765 766 767
  int32_t numOfVgroups;
  int32_t numOfTables;
  int32_t numOfUdfs;
  char    tableNames[];
S
Shengliang Guan 已提交
768
} SMultiTableInfoReq;
D
dapan1121 已提交
769

H
Haojun Liao 已提交
770
// todo refactor
D
dapan1121 已提交
771
typedef struct SVgroupInfo {
772 773 774
  int32_t  vgId;
  uint32_t hashBegin;
  uint32_t hashEnd;
L
Liu Jicong 已提交
775
  SEpSet   epSet;
D
dapan1121 已提交
776
  int32_t  numOfTable;  // unit is TSDB_TABLE_NUM_UNIT
D
dapan1121 已提交
777 778
} SVgroupInfo;

D
dapan1121 已提交
779
typedef struct {
H
Haojun Liao 已提交
780 781
  int32_t     numOfVgroups;
  SVgroupInfo vgroups[];
S
Shengliang Guan 已提交
782
} SVgroupsInfo;
D
dapan1121 已提交
783

S
Shengliang Guan 已提交
784
typedef struct {
D
dapan1121 已提交
785 786 787
  char     tbName[TSDB_TABLE_NAME_LEN];
  char     stbName[TSDB_TABLE_NAME_LEN];
  char     dbFName[TSDB_DB_FNAME_LEN];
L
Liu Jicong 已提交
788
  int64_t  dbId;
S
Shengliang Guan 已提交
789 790 791 792 793 794 795
  int32_t  numOfTags;
  int32_t  numOfColumns;
  int8_t   precision;
  int8_t   tableType;
  int8_t   update;
  int32_t  sversion;
  int32_t  tversion;
L
Liu Jicong 已提交
796 797
  uint64_t suid;
  uint64_t tuid;
S
Shengliang Guan 已提交
798
  int32_t  vgId;
S
Shengliang Guan 已提交
799
  SSchema* pSchemas;
S
Shengliang Guan 已提交
800
} STableMetaRsp;
D
dapan1121 已提交
801

S
Shengliang Guan 已提交
802 803 804 805 806 807 808 809 810 811 812 813
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);

S
Shengliang Guan 已提交
814
typedef struct {
S
Shengliang Guan 已提交
815 816 817 818 819 820 821 822
  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[];
D
dapan1121 已提交
823 824 825 826 827
} SMultiTableMeta;

typedef struct {
  int32_t dataLen;
  char    name[TSDB_TABLE_FNAME_LEN];
H
Hongze Cheng 已提交
828
  char*   data;
D
dapan1121 已提交
829 830 831 832 833 834 835 836
} STagData;

/*
 * sql: show tables like '%a_%'
 * payload is the query condition, e.g., '%a_%'
 * payloadLen is the length of payload
 */
typedef struct {
L
Liu Jicong 已提交
837
  int32_t type;
838
  char    db[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
839 840
  int32_t payloadLen;
  char*   payload;
S
Shengliang Guan 已提交
841
} SShowReq;
D
dapan1121 已提交
842

S
Shengliang Guan 已提交
843 844 845 846
int32_t tSerializeSShowReq(void* buf, int32_t bufLen, SShowReq* pReq);
int32_t tDeserializeSShowReq(void* buf, int32_t bufLen, SShowReq* pReq);
void    tFreeSShowReq(SShowReq* pReq);

S
Shengliang Guan 已提交
847
typedef struct {
848
  int64_t       showId;
S
Shengliang Guan 已提交
849
  STableMetaRsp tableMeta;
S
Shengliang Guan 已提交
850
} SShowRsp, SVShowTablesRsp;
D
dapan1121 已提交
851

S
Shengliang Guan 已提交
852 853 854
int32_t tSerializeSShowRsp(void* buf, int32_t bufLen, SShowRsp* pRsp);
int32_t tDeserializeSShowRsp(void* buf, int32_t bufLen, SShowRsp* pRsp);
void    tFreeSShowRsp(SShowRsp* pRsp);
D
dapan1121 已提交
855

856
typedef struct {
857 858
  int32_t type;
  char    db[TSDB_DB_FNAME_LEN];
859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875
  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;

D
dapan1121 已提交
876
typedef struct {
S
Shengliang Guan 已提交
877
  char    fqdn[TSDB_FQDN_LEN];  // end point, hostname:port
878
  int32_t port;
S
Shengliang Guan 已提交
879
} SCreateDnodeReq;
S
Shengliang Guan 已提交
880

S
Shengliang Guan 已提交
881 882 883
int32_t tSerializeSCreateDnodeReq(void* buf, int32_t bufLen, SCreateDnodeReq* pReq);
int32_t tDeserializeSCreateDnodeReq(void* buf, int32_t bufLen, SCreateDnodeReq* pReq);

S
Shengliang Guan 已提交
884 885
typedef struct {
  int32_t dnodeId;
886 887
  char    fqdn[TSDB_FQDN_LEN];
  int32_t port;
S
Shengliang Guan 已提交
888
} SDropDnodeReq;
S
Shengliang Guan 已提交
889

S
Shengliang Guan 已提交
890 891 892
int32_t tSerializeSDropDnodeReq(void* buf, int32_t bufLen, SDropDnodeReq* pReq);
int32_t tDeserializeSDropDnodeReq(void* buf, int32_t bufLen, SDropDnodeReq* pReq);

S
Shengliang Guan 已提交
893 894
typedef struct {
  int32_t dnodeId;
895
  char    config[TSDB_DNODE_CONFIG_LEN];
S
Shengliang Guan 已提交
896
  char    value[TSDB_DNODE_VALUE_LEN];
S
Shengliang Guan 已提交
897
} SMCfgDnodeReq, SDCfgDnodeReq;
D
dapan1121 已提交
898

S
Shengliang Guan 已提交
899 900 901
int32_t tSerializeSMCfgDnodeReq(void* buf, int32_t bufLen, SMCfgDnodeReq* pReq);
int32_t tDeserializeSMCfgDnodeReq(void* buf, int32_t bufLen, SMCfgDnodeReq* pReq);

S
Shengliang Guan 已提交
902 903
typedef struct {
  int32_t dnodeId;
S
Shengliang Guan 已提交
904
} SMCreateMnodeReq, SMDropMnodeReq, SDDropMnodeReq;
S
Shengliang Guan 已提交
905

S
Shengliang Guan 已提交
906 907 908
int32_t tSerializeSMCreateDropMnodeReq(void* buf, int32_t bufLen, SMCreateMnodeReq* pReq);
int32_t tDeserializeSMCreateDropMnodeReq(void* buf, int32_t bufLen, SMCreateMnodeReq* pReq);

D
dapan1121 已提交
909
typedef struct {
S
Shengliang Guan 已提交
910 911
  int32_t  dnodeId;
  int8_t   replica;
S
Shengliang Guan 已提交
912
  SReplica replicas[TSDB_MAX_REPLICA];
S
Shengliang Guan 已提交
913
} SDCreateMnodeReq, SDAlterMnodeReq;
D
dapan1121 已提交
914

S
Shengliang Guan 已提交
915 916 917
int32_t tSerializeSDCreateMnodeReq(void* buf, int32_t bufLen, SDCreateMnodeReq* pReq);
int32_t tDeserializeSDCreateMnodeReq(void* buf, int32_t bufLen, SDCreateMnodeReq* pReq);

S
Shengliang Guan 已提交
918 919
typedef struct {
  int32_t dnodeId;
S
Shengliang Guan 已提交
920 921
} SMCreateQnodeReq, SMDropQnodeReq, SDCreateQnodeReq, SDDropQnodeReq, SMCreateSnodeReq, SMDropSnodeReq,
    SDCreateSnodeReq, SDDropSnodeReq, SMCreateBnodeReq, SMDropBnodeReq, SDCreateBnodeReq, SDDropBnodeReq;
S
Shengliang Guan 已提交
922

S
Shengliang Guan 已提交
923 924
int32_t tSerializeSMCreateDropQSBNodeReq(void* buf, int32_t bufLen, SMCreateQnodeReq* pReq);
int32_t tDeserializeSMCreateDropQSBNodeReq(void* buf, int32_t bufLen, SMCreateQnodeReq* pReq);
S
Shengliang Guan 已提交
925

D
dapan1121 已提交
926
typedef struct {
S
Shengliang Guan 已提交
927 928 929 930 931 932 933 934 935 936 937
  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)
D
dapan1121 已提交
938 939 940
} SQueryDesc;

typedef struct {
S
Shengliang Guan 已提交
941 942 943 944 945 946
  int32_t connId;
  int32_t pid;
  int32_t numOfQueries;
  int32_t numOfStreams;
  char    app[TSDB_APP_NAME_LEN];
  char    pData[];
S
Shengliang Guan 已提交
947
} SHeartBeatReq;
D
dapan1121 已提交
948 949

typedef struct {
S
Shengliang Guan 已提交
950 951 952 953 954 955
  int32_t connId;
  int32_t queryId;
  int32_t streamId;
  int32_t totalDnodes;
  int32_t onlineDnodes;
  int8_t  killConnection;
S
Shengliang Guan 已提交
956
  int8_t  align[3];
S
Shengliang Guan 已提交
957
  SEpSet  epSet;
D
dapan1121 已提交
958 959
} SHeartBeatRsp;

S
Shengliang Guan 已提交
960 961 962
typedef struct {
  int32_t connId;
  int32_t queryId;
S
Shengliang Guan 已提交
963
} SKillQueryReq;
S
Shengliang Guan 已提交
964

S
Shengliang Guan 已提交
965 966 967
int32_t tSerializeSKillQueryReq(void* buf, int32_t bufLen, SKillQueryReq* pReq);
int32_t tDeserializeSKillQueryReq(void* buf, int32_t bufLen, SKillQueryReq* pReq);

S
Shengliang Guan 已提交
968 969
typedef struct {
  int32_t connId;
S
Shengliang Guan 已提交
970
} SKillConnReq;
D
dapan1121 已提交
971

S
Shengliang Guan 已提交
972 973 974
int32_t tSerializeSKillConnReq(void* buf, int32_t bufLen, SKillConnReq* pReq);
int32_t tDeserializeSKillConnReq(void* buf, int32_t bufLen, SKillConnReq* pReq);

S
Shengliang Guan 已提交
975 976 977 978 979 980 981
typedef struct {
  int32_t transId;
} SKillTransReq;

int32_t tSerializeSKillTransReq(void* buf, int32_t bufLen, SKillTransReq* pReq);
int32_t tDeserializeSKillTransReq(void* buf, int32_t bufLen, SKillTransReq* pReq);

D
dapan1121 已提交
982 983 984 985
typedef struct {
  char user[TSDB_USER_LEN];
  char spi;
  char encrypt;
986 987
  char secret[TSDB_PASSWORD_LEN];
  char ckey[TSDB_PASSWORD_LEN];
S
Shengliang Guan 已提交
988
} SAuthReq, SAuthRsp;
D
dapan1121 已提交
989

S
Shengliang Guan 已提交
990 991 992
int32_t tSerializeSAuthReq(void* buf, int32_t bufLen, SAuthReq* pReq);
int32_t tDeserializeSAuthReq(void* buf, int32_t bufLen, SAuthReq* pReq);

D
dapan1121 已提交
993
typedef struct {
S
Shengliang Guan 已提交
994 995 996
  int8_t finished;
  char   name[TSDB_STEP_NAME_LEN];
  char   desc[TSDB_STEP_DESC_LEN];
S
Shengliang Guan 已提交
997
} SStartupReq;
D
dapan1121 已提交
998

999 1000 1001 1002 1003 1004 1005
/**
 * The layout of the query message payload is as following:
 * +--------------------+---------------------------------+
 * |Sql statement       | Physical plan                   |
 * |(denoted by sqlLen) |(In JSON, denoted by contentLen) |
 * +--------------------+---------------------------------+
 */
L
Liu Jicong 已提交
1006
typedef struct SSubQueryMsg {
D
dapan1121 已提交
1007
  SMsgHead header;
D
dapan1121 已提交
1008
  uint64_t sId;
H
Hongze Cheng 已提交
1009 1010
  uint64_t queryId;
  uint64_t taskId;
D
dapan1121 已提交
1011
  int64_t  refId;
D
dapan1121 已提交
1012
  int8_t   taskType;
1013
  uint32_t sqlLen;  // the query sql,
1014
  uint32_t phyLen;
H
Hongze Cheng 已提交
1015
  char     msg[];
D
dapan1121 已提交
1016
} SSubQueryMsg;
D
dapan 已提交
1017

1018 1019 1020 1021 1022 1023 1024
typedef struct {
  SMsgHead header;
  uint64_t sId;
  uint64_t queryId;
  uint64_t taskId;
} SSinkDataReq;

D
dapan1121 已提交
1025 1026 1027 1028 1029 1030 1031
typedef struct {
  SMsgHead header;
  uint64_t sId;
  uint64_t queryId;
  uint64_t taskId;
} SQueryContinueReq;

S
Shengliang Guan 已提交
1032
typedef struct {
D
dapan1121 已提交
1033
  SMsgHead header;
D
dapan1121 已提交
1034
  uint64_t sId;
H
Hongze Cheng 已提交
1035 1036
  uint64_t queryId;
  uint64_t taskId;
S
Shengliang Guan 已提交
1037
} SResReadyReq;
D
dapan 已提交
1038

S
Shengliang Guan 已提交
1039
typedef struct {
D
dapan1121 已提交
1040 1041 1042
  int32_t code;
} SResReadyRsp;

S
Shengliang Guan 已提交
1043
typedef struct {
D
dapan1121 已提交
1044
  SMsgHead header;
D
dapan1121 已提交
1045
  uint64_t sId;
H
Hongze Cheng 已提交
1046 1047
  uint64_t queryId;
  uint64_t taskId;
S
Shengliang Guan 已提交
1048
} SResFetchReq;
D
dapan 已提交
1049

S
Shengliang Guan 已提交
1050
typedef struct {
D
dapan1121 已提交
1051
  SMsgHead header;
D
dapan1121 已提交
1052
  uint64_t sId;
S
Shengliang Guan 已提交
1053
} SSchTasksStatusReq;
D
dapan1121 已提交
1054

S
Shengliang Guan 已提交
1055
typedef struct {
H
Hongze Cheng 已提交
1056 1057
  uint64_t queryId;
  uint64_t taskId;
D
dapan1121 已提交
1058
  int64_t  refId;
H
Hongze Cheng 已提交
1059
  int8_t   status;
D
dapan1121 已提交
1060 1061
} STaskStatus;

S
Shengliang Guan 已提交
1062
typedef struct {
L
Liu Jicong 已提交
1063 1064
  int64_t refId;
  SArray* taskStatus;  // SArray<STaskStatus>
D
dapan1121 已提交
1065 1066
} SSchedulerStatusRsp;

D
dapan1121 已提交
1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081
typedef struct {
  uint64_t queryId;
  uint64_t taskId;
  int8_t   action;
} STaskAction;

typedef struct SQueryNodeEpId {
  int32_t nodeId;  // vgId or qnodeId
  SEp     ep;
} SQueryNodeEpId;

typedef struct {
  SMsgHead       header;
  uint64_t       sId;
  SQueryNodeEpId epId;
L
Liu Jicong 已提交
1082
  SArray*        taskAction;  // SArray<STaskAction>
D
dapan1121 已提交
1083 1084
} SSchedulerHbReq;

L
Liu Jicong 已提交
1085 1086 1087
int32_t tSerializeSSchedulerHbReq(void* buf, int32_t bufLen, SSchedulerHbReq* pReq);
int32_t tDeserializeSSchedulerHbReq(void* buf, int32_t bufLen, SSchedulerHbReq* pReq);
void    tFreeSSchedulerHbReq(SSchedulerHbReq* pReq);
D
dapan1121 已提交
1088 1089 1090 1091

typedef struct {
  uint64_t       seqId;
  SQueryNodeEpId epId;
L
Liu Jicong 已提交
1092
  SArray*        taskStatus;  // SArray<STaskStatus>
D
dapan1121 已提交
1093 1094
} SSchedulerHbRsp;

L
Liu Jicong 已提交
1095 1096 1097
int32_t tSerializeSSchedulerHbRsp(void* buf, int32_t bufLen, SSchedulerHbRsp* pRsp);
int32_t tDeserializeSSchedulerHbRsp(void* buf, int32_t bufLen, SSchedulerHbRsp* pRsp);
void    tFreeSSchedulerHbRsp(SSchedulerHbRsp* pRsp);
D
dapan1121 已提交
1098

S
Shengliang Guan 已提交
1099
typedef struct {
D
dapan1121 已提交
1100
  SMsgHead header;
D
dapan1121 已提交
1101
  uint64_t sId;
H
Hongze Cheng 已提交
1102 1103
  uint64_t queryId;
  uint64_t taskId;
D
dapan1121 已提交
1104
  int64_t  refId;
S
Shengliang Guan 已提交
1105
} STaskCancelReq;
D
dapan1121 已提交
1106

S
Shengliang Guan 已提交
1107
typedef struct {
D
dapan1121 已提交
1108 1109 1110
  int32_t code;
} STaskCancelRsp;

S
Shengliang Guan 已提交
1111
typedef struct {
D
dapan1121 已提交
1112
  SMsgHead header;
D
dapan1121 已提交
1113
  uint64_t sId;
H
Hongze Cheng 已提交
1114 1115
  uint64_t queryId;
  uint64_t taskId;
D
dapan1121 已提交
1116
  int64_t  refId;
S
Shengliang Guan 已提交
1117
} STaskDropReq;
D
dapan1121 已提交
1118

S
Shengliang Guan 已提交
1119
typedef struct {
D
dapan1121 已提交
1120 1121 1122
  int32_t code;
} STaskDropRsp;

L
Liu Jicong 已提交
1123
typedef struct {
L
Liu Jicong 已提交
1124 1125 1126 1127 1128
  char   name[TSDB_TOPIC_FNAME_LEN];
  int8_t igExists;
  char*  sql;
  char*  physicalPlan;
  char*  logicalPlan;
L
Liu Jicong 已提交
1129
} SCMCreateStreamReq;
L
Liu Jicong 已提交
1130

L
Liu Jicong 已提交
1131 1132 1133 1134 1135 1136 1137 1138
typedef struct {
  int64_t streamId;
} SCMCreateStreamRsp;

int32_t tSerializeSCMCreateStreamReq(void* buf, int32_t bufLen, const SCMCreateStreamReq* pReq);
int32_t tDeserializeSCMCreateStreamReq(void* buf, int32_t bufLen, SCMCreateStreamReq* pReq);
void    tFreeSCMCreateStreamReq(SCMCreateStreamReq* pReq);

L
Liu Jicong 已提交
1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149
typedef struct {
  char    name[TSDB_TOPIC_FNAME_LEN];
  int64_t streamId;
  char*   sql;
  char*   executorMsg;
} SMVCreateStreamReq, SMSCreateStreamReq;

typedef struct {
  int64_t streamId;
} SMVCreateStreamRsp, SMSCreateStreamRsp;

L
Liu Jicong 已提交
1150 1151 1152 1153
typedef struct {
  char   name[TSDB_TOPIC_FNAME_LEN];
  int8_t igExists;
  char*  sql;
1154 1155
  char*  ast;
  char   subscribeDbName[TSDB_DB_NAME_LEN];
L
Liu Jicong 已提交
1156 1157 1158 1159 1160
} SCMCreateTopicReq;

int32_t tSerializeSCMCreateTopicReq(void* buf, int32_t bufLen, const SCMCreateTopicReq* pReq);
int32_t tDeserializeSCMCreateTopicReq(void* buf, int32_t bufLen, SCMCreateTopicReq* pReq);
void    tFreeSCMCreateTopicReq(SCMCreateTopicReq* pReq);
L
Liu Jicong 已提交
1161 1162 1163

typedef struct {
  int64_t topicId;
L
Liu Jicong 已提交
1164
} SCMCreateTopicRsp;
L
Liu Jicong 已提交
1165

L
Liu Jicong 已提交
1166 1167
int32_t tSerializeSCMCreateTopicRsp(void* buf, int32_t bufLen, const SCMCreateTopicRsp* pRsp);
int32_t tDeserializeSCMCreateTopicRsp(void* buf, int32_t bufLen, SCMCreateTopicRsp* pRsp);
L
Liu Jicong 已提交
1168 1169

typedef struct {
L
Liu Jicong 已提交
1170
  int32_t topicNum;
L
Liu Jicong 已提交
1171
  int64_t consumerId;
L
Liu Jicong 已提交
1172
  char*   consumerGroup;
L
Liu Jicong 已提交
1173
  SArray* topicNames;  // SArray<char*>
L
Liu Jicong 已提交
1174 1175
} SCMSubscribeReq;

S
Shengliang Guan 已提交
1176 1177
static FORCE_INLINE int32_t tSerializeSCMSubscribeReq(void** buf, const SCMSubscribeReq* pReq) {
  int32_t tlen = 0;
L
Liu Jicong 已提交
1178
  tlen += taosEncodeFixedI32(buf, pReq->topicNum);
L
Liu Jicong 已提交
1179
  tlen += taosEncodeFixedI64(buf, pReq->consumerId);
L
Liu Jicong 已提交
1180
  tlen += taosEncodeString(buf, pReq->consumerGroup);
L
Liu Jicong 已提交
1181

S
Shengliang Guan 已提交
1182
  for (int32_t i = 0; i < pReq->topicNum; i++) {
L
Liu Jicong 已提交
1183
    tlen += taosEncodeString(buf, (char*)taosArrayGetP(pReq->topicNames, i));
L
Liu Jicong 已提交
1184
  }
L
Liu Jicong 已提交
1185 1186 1187 1188
  return tlen;
}

static FORCE_INLINE void* tDeserializeSCMSubscribeReq(void* buf, SCMSubscribeReq* pReq) {
L
Liu Jicong 已提交
1189
  buf = taosDecodeFixedI32(buf, &pReq->topicNum);
L
Liu Jicong 已提交
1190
  buf = taosDecodeFixedI64(buf, &pReq->consumerId);
L
Liu Jicong 已提交
1191
  buf = taosDecodeString(buf, &pReq->consumerGroup);
L
Liu Jicong 已提交
1192
  pReq->topicNames = taosArrayInit(pReq->topicNum, sizeof(void*));
S
Shengliang Guan 已提交
1193
  for (int32_t i = 0; i < pReq->topicNum; i++) {
L
Liu Jicong 已提交
1194
    char* name;
L
Liu Jicong 已提交
1195 1196
    buf = taosDecodeString(buf, &name);
    taosArrayPush(pReq->topicNames, &name);
L
Liu Jicong 已提交
1197
  }
L
Liu Jicong 已提交
1198 1199 1200
  return buf;
}

L
Liu Jicong 已提交
1201
typedef struct SMqSubTopic {
L
Liu Jicong 已提交
1202
  int32_t vgId;
L
Liu Jicong 已提交
1203 1204 1205 1206 1207
  int64_t topicId;
  SEpSet  epSet;
} SMqSubTopic;

typedef struct {
L
Liu Jicong 已提交
1208
  int32_t     topicNum;
L
Liu Jicong 已提交
1209
  SMqSubTopic topics[];
L
Liu Jicong 已提交
1210 1211
} SCMSubscribeRsp;

S
Shengliang Guan 已提交
1212 1213
static FORCE_INLINE int32_t tSerializeSCMSubscribeRsp(void** buf, const SCMSubscribeRsp* pRsp) {
  int32_t tlen = 0;
L
Liu Jicong 已提交
1214
  tlen += taosEncodeFixedI32(buf, pRsp->topicNum);
S
Shengliang Guan 已提交
1215
  for (int32_t i = 0; i < pRsp->topicNum; i++) {
L
Liu Jicong 已提交
1216 1217 1218 1219
    tlen += taosEncodeFixedI32(buf, pRsp->topics[i].vgId);
    tlen += taosEncodeFixedI64(buf, pRsp->topics[i].topicId);
    tlen += taosEncodeSEpSet(buf, &pRsp->topics[i].epSet);
  }
L
Liu Jicong 已提交
1220 1221 1222 1223
  return tlen;
}

static FORCE_INLINE void* tDeserializeSCMSubscribeRsp(void* buf, SCMSubscribeRsp* pRsp) {
L
Liu Jicong 已提交
1224
  buf = taosDecodeFixedI32(buf, &pRsp->topicNum);
S
Shengliang Guan 已提交
1225
  for (int32_t i = 0; i < pRsp->topicNum; i++) {
L
Liu Jicong 已提交
1226 1227 1228 1229
    buf = taosDecodeFixedI32(buf, &pRsp->topics[i].vgId);
    buf = taosDecodeFixedI64(buf, &pRsp->topics[i].topicId);
    buf = taosDecodeSEpSet(buf, &pRsp->topics[i].epSet);
  }
L
Liu Jicong 已提交
1230 1231 1232 1233 1234 1235 1236 1237
  return buf;
}

typedef struct {
  int64_t topicId;
  int64_t consumerId;
  int64_t consumerGroupId;
  int64_t offset;
L
Liu Jicong 已提交
1238 1239 1240
  char*   sql;
  char*   logicalPlan;
  char*   physicalPlan;
L
Liu Jicong 已提交
1241 1242
} SMVSubscribeReq;

S
Shengliang Guan 已提交
1243 1244
static FORCE_INLINE int32_t tSerializeSMVSubscribeReq(void** buf, SMVSubscribeReq* pReq) {
  int32_t tlen = 0;
L
Liu Jicong 已提交
1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265
  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;
}

L
Liu Jicong 已提交
1266
typedef struct {
1267 1268 1269 1270
  const char* key;
  SArray*     lostConsumers;     // SArray<int64_t>
  SArray*     removedConsumers;  // SArray<int64_t>
  SArray*     newConsumers;      // SArray<int64_t>
L
Liu Jicong 已提交
1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299
} 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;
}

L
Liu Jicong 已提交
1300
// this message is sent from mnode to mnode(read thread to write thread), so there is no need for serialization or
1301
// deserialization
L
Liu Jicong 已提交
1302
typedef struct {
1303
  SHashObj* rebSubHash;  // SHashObj<key, SMqRebSubscribe>
L
Liu Jicong 已提交
1304 1305
} SMqDoRebalanceMsg;

L
Liu Jicong 已提交
1306
typedef struct {
L
Liu Jicong 已提交
1307
  int64_t status;
L
Liu Jicong 已提交
1308 1309
} SMVSubscribeRsp;

L
Liu Jicong 已提交
1310 1311 1312
typedef struct {
  char   name[TSDB_TABLE_FNAME_LEN];
  int8_t igNotExists;
S
Shengliang Guan 已提交
1313 1314
} SMDropTopicReq;

S
Shengliang Guan 已提交
1315
int32_t tSerializeSMDropTopicReq(void* buf, int32_t bufLen, SMDropTopicReq* pReq);
S
Shengliang Guan 已提交
1316
int32_t tDeserializeSMDropTopicReq(void* buf, int32_t bufLen, SMDropTopicReq* pReq);
L
Liu Jicong 已提交
1317 1318 1319 1320 1321

typedef struct {
  char    name[TSDB_TABLE_FNAME_LEN];
  int8_t  alterType;
  SSchema schema;
S
Shengliang Guan 已提交
1322
} SAlterTopicReq;
L
Liu Jicong 已提交
1323 1324 1325 1326

typedef struct {
  SMsgHead head;
  char     name[TSDB_TABLE_FNAME_LEN];
L
Liu Jicong 已提交
1327
  int64_t  tuid;
L
Liu Jicong 已提交
1328 1329 1330 1331 1332
  int32_t  sverson;
  int32_t  execLen;
  char*    executor;
  int32_t  sqlLen;
  char*    sql;
S
Shengliang Guan 已提交
1333
} SDCreateTopicReq;
L
Liu Jicong 已提交
1334 1335 1336 1337

typedef struct {
  SMsgHead head;
  char     name[TSDB_TABLE_FNAME_LEN];
L
Liu Jicong 已提交
1338
  int64_t  tuid;
S
Shengliang Guan 已提交
1339
} SDDropTopicReq;
S
Shengliang Guan 已提交
1340

H
more  
Hongze Cheng 已提交
1341
typedef struct SVCreateTbReq {
L
Liu Jicong 已提交
1342
  int64_t  ver;  // use a general definition
H
more  
Hongze Cheng 已提交
1343 1344 1345
  char*    name;
  uint32_t ttl;
  uint32_t keep;
S
Shengliang Guan 已提交
1346
  uint8_t  type;
H
more  
Hongze Cheng 已提交
1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363
  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;
  };
S
Shengliang Guan 已提交
1364
} SVCreateTbReq, SVUpdateTbReq;
H
more  
Hongze Cheng 已提交
1365

H
more  
Hongze Cheng 已提交
1366
typedef struct {
L
Liu Jicong 已提交
1367
  int tmp;  // TODO: to avoid compile error
S
Shengliang Guan 已提交
1368
} SVCreateTbRsp, SVUpdateTbRsp;
H
more  
Hongze Cheng 已提交
1369

S
Shengliang Guan 已提交
1370 1371
int32_t tSerializeSVCreateTbReq(void** buf, SVCreateTbReq* pReq);
void*   tDeserializeSVCreateTbReq(void* buf, SVCreateTbReq* pReq);
H
more  
Hongze Cheng 已提交
1372

H
more  
Hongze Cheng 已提交
1373
typedef struct {
L
Liu Jicong 已提交
1374 1375
  int64_t ver;  // use a general definition
  SArray* pArray;
H
more  
Hongze Cheng 已提交
1376
} SVCreateTbBatchReq;
H
more  
Hongze Cheng 已提交
1377

S
Shengliang Guan 已提交
1378
typedef struct {
L
Liu Jicong 已提交
1379
  int tmp;  // TODO: to avoid compile error
S
Shengliang Guan 已提交
1380
} SVCreateTbBatchRsp;
H
more  
Hongze Cheng 已提交
1381

S
Shengliang Guan 已提交
1382 1383
int32_t tSerializeSVCreateTbBatchReq(void** buf, SVCreateTbBatchReq* pReq);
void*   tDeserializeSVCreateTbBatchReq(void* buf, SVCreateTbBatchReq* pReq);
S
Shengliang Guan 已提交
1384 1385

typedef struct {
L
Liu Jicong 已提交
1386
  int64_t  ver;
S
Shengliang Guan 已提交
1387 1388 1389
  char*    name;
  uint8_t  type;
  tb_uid_t suid;
S
Shengliang Guan 已提交
1390 1391 1392
} SVDropTbReq;

typedef struct {
L
Liu Jicong 已提交
1393
  int tmp;  // TODO: to avoid compile error
S
Shengliang Guan 已提交
1394 1395
} SVDropTbRsp;

S
Shengliang Guan 已提交
1396 1397 1398
int32_t tSerializeSVDropTbReq(void** buf, SVDropTbReq* pReq);
void*   tDeserializeSVDropTbReq(void* buf, SVDropTbReq* pReq);

S
Shengliang Guan 已提交
1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417
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 {
H
Hongze Cheng 已提交
1418 1419 1420
  SMsgHead head;
} SVShowTablesReq;

S
Shengliang Guan 已提交
1421
typedef struct {
H
Hongze Cheng 已提交
1422
  SMsgHead head;
H
Haojun Liao 已提交
1423
  int32_t  id;
H
Hongze Cheng 已提交
1424 1425
} SVShowTablesFetchReq;

S
Shengliang Guan 已提交
1426
typedef struct {
H
Hongze Cheng 已提交
1427 1428 1429 1430 1431 1432 1433 1434 1435
  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;

L
Liu Jicong 已提交
1436 1437 1438
typedef struct SMqCMGetSubEpReq {
  int64_t consumerId;
  int32_t epoch;
L
Liu Jicong 已提交
1439
  char    cgroup[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
1440 1441
} SMqCMGetSubEpReq;

L
Liu Jicong 已提交
1442 1443 1444 1445 1446 1447 1448 1449
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 {
1450
  int8_t status;  // idle or not
L
Liu Jicong 已提交
1451
  int8_t vnodeChanged;
1452
  int8_t epChanged;  // should use new epset
L
Liu Jicong 已提交
1453 1454 1455 1456
  int8_t reserved;
  SEpSet epSet;
} SMqHbRsp;

S
Shengliang Guan 已提交
1457 1458
static FORCE_INLINE int32_t taosEncodeSMqHbRsp(void** buf, const SMqHbRsp* pRsp) {
  int32_t tlen = 0;
L
Liu Jicong 已提交
1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474
  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 {
1475
  char    topicName[TSDB_TOPIC_FNAME_LEN];
L
Liu Jicong 已提交
1476 1477 1478
  SArray* rsps;  // SArray<SMqHbRsp>
} SMqHbOneTopicBatchRsp;

S
Shengliang Guan 已提交
1479 1480
static FORCE_INLINE int32_t taosEncodeSMqHbOneTopicBatchRsp(void** buf, const SMqHbOneTopicBatchRsp* pBatchRsp) {
  int32_t tlen = 0;
L
Liu Jicong 已提交
1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504
  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 {
1505 1506
  int64_t consumerId;
  SArray* batchRsps;  // SArray<SMqHbOneTopicBatchRsp>
L
Liu Jicong 已提交
1507 1508
} SMqHbBatchRsp;

S
Shengliang Guan 已提交
1509 1510
static FORCE_INLINE int32_t taosEncodeSMqHbBatchRsp(void** buf, const SMqHbBatchRsp* pBatchRsp) {
  int32_t tlen = 0;
L
Liu Jicong 已提交
1511 1512 1513 1514
  tlen += taosEncodeFixedI64(buf, pBatchRsp->consumerId);
  int32_t sz;
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
1515
    SMqHbOneTopicBatchRsp* pRsp = (SMqHbOneTopicBatchRsp*)taosArrayGet(pBatchRsp->batchRsps, i);
L
Liu Jicong 已提交
1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526
    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++) {
1527
    SMqHbOneTopicBatchRsp rsp;
L
Liu Jicong 已提交
1528 1529 1530 1531 1532 1533 1534
    buf = taosDecodeSMqHbOneTopicBatchRsp(buf, &rsp);
    buf = taosArrayPush(pBatchRsp->batchRsps, &rsp);
  }
  return buf;
}

typedef struct {
D
dapan1121 已提交
1535
  int32_t key;
L
Liu Jicong 已提交
1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557
  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;
D
dapan1121 已提交
1558
  SArray*      info;  // Array<Skv>
L
Liu Jicong 已提交
1559 1560 1561 1562 1563 1564 1565 1566
} SClientHbRsp;

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

1567
static FORCE_INLINE uint32_t hbKeyHashFunc(const char* key, uint32_t keyLen) { return taosIntHash_64(key, keyLen); }
L
Liu Jicong 已提交
1568

1569 1570
static FORCE_INLINE void tFreeReqKvHash(SHashObj* info) {
  void* pIter = taosHashIterate(info, NULL);
D
dapan1121 已提交
1571 1572 1573 1574 1575 1576 1577
  while (pIter != NULL) {
    SKv* kv = (SKv*)pIter;
    tfree(kv->value);
    pIter = taosHashIterate(info, pIter);
  }
}

1578
static FORCE_INLINE void tFreeClientHbReq(void* pReq) {
L
Liu Jicong 已提交
1579
  SClientHbReq* req = (SClientHbReq*)pReq;
D
dapan1121 已提交
1580 1581 1582 1583
  if (req->info) {
    tFreeReqKvHash(req->info);
    taosHashCleanup(req->info);
  }
L
Liu Jicong 已提交
1584 1585
}

S
Shengliang Guan 已提交
1586 1587
int32_t tSerializeSClientHbBatchReq(void* buf, int32_t bufLen, const SClientHbBatchReq* pReq);
int32_t tDeserializeSClientHbBatchReq(void* buf, int32_t bufLen, SClientHbBatchReq* pReq);
L
Liu Jicong 已提交
1588 1589

static FORCE_INLINE void tFreeClientHbBatchReq(void* pReq, bool deep) {
1590
  SClientHbBatchReq* req = (SClientHbBatchReq*)pReq;
L
Liu Jicong 已提交
1591 1592 1593 1594 1595 1596 1597 1598
  if (deep) {
    taosArrayDestroyEx(req->reqs, tFreeClientHbReq);
  } else {
    taosArrayDestroy(req->reqs);
  }
  free(pReq);
}

1599 1600
static FORCE_INLINE void tFreeClientKv(void* pKv) {
  SKv* kv = (SKv*)pKv;
D
dapan1121 已提交
1601 1602 1603 1604 1605
  if (kv) {
    tfree(kv->value);
  }
}

1606
static FORCE_INLINE void tFreeClientHbRsp(void* pRsp) {
D
dapan1121 已提交
1607 1608 1609 1610 1611
  SClientHbRsp* rsp = (SClientHbRsp*)pRsp;
  if (rsp->info) taosArrayDestroyEx(rsp->info, tFreeClientKv);
}

static FORCE_INLINE void tFreeClientHbBatchRsp(void* pRsp) {
1612
  SClientHbBatchRsp* rsp = (SClientHbBatchRsp*)pRsp;
D
dapan1121 已提交
1613 1614 1615
  taosArrayDestroyEx(rsp->rsps, tFreeClientHbRsp);
}

S
Shengliang Guan 已提交
1616 1617
int32_t tSerializeSClientHbBatchRsp(void* buf, int32_t bufLen, const SClientHbBatchRsp* pBatchRsp);
int32_t tDeserializeSClientHbBatchRsp(void* buf, int32_t bufLen, SClientHbBatchRsp* pBatchRsp);
L
Liu Jicong 已提交
1618

S
Shengliang Guan 已提交
1619 1620 1621
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;
S
Shengliang Guan 已提交
1622
  if (tEncodeBinary(pEncoder, (const char*)pKv->value, pKv->valueLen) < 0) return -1;
S
Shengliang Guan 已提交
1623
  return 0;
L
Liu Jicong 已提交
1624 1625
}

S
Shengliang Guan 已提交
1626 1627 1628 1629 1630 1631 1632
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;
L
Liu Jicong 已提交
1633 1634
}

S
Shengliang Guan 已提交
1635 1636 1637 1638
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;
L
Liu Jicong 已提交
1639 1640
}

S
Shengliang Guan 已提交
1641 1642 1643 1644
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;
L
Liu Jicong 已提交
1645 1646 1647 1648 1649 1650
}

typedef struct SMqHbVgInfo {
  int32_t vgId;
} SMqHbVgInfo;

S
Shengliang Guan 已提交
1651 1652
static FORCE_INLINE int32_t taosEncodeSMqVgInfo(void** buf, const SMqHbVgInfo* pVgInfo) {
  int32_t tlen = 0;
L
Liu Jicong 已提交
1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668
  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;

S
Shengliang Guan 已提交
1669 1670
static FORCE_INLINE int32_t taosEncodeSMqHbTopicInfoMsg(void** buf, const SMqHbTopicInfo* pTopicInfo) {
  int32_t tlen = 0;
L
Liu Jicong 已提交
1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698
  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 {
1699 1700 1701 1702
  int32_t status;  // ask hb endpoint
  int32_t epoch;
  int64_t consumerId;
  SArray* pTopics;  // SArray<SMqHbTopicInfo>
L
Liu Jicong 已提交
1703 1704
} SMqHbMsg;

S
Shengliang Guan 已提交
1705 1706
static FORCE_INLINE int32_t taosEncodeSMqMsg(void** buf, const SMqHbMsg* pMsg) {
  int32_t tlen = 0;
L
Liu Jicong 已提交
1707 1708 1709 1710 1711
  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);
S
Shengliang Guan 已提交
1712
  for (int32_t i = 0; i < sz; i++) {
L
Liu Jicong 已提交
1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725
    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));
S
Shengliang Guan 已提交
1726
  for (int32_t i = 0; i < sz; i++) {
L
Liu Jicong 已提交
1727 1728 1729 1730 1731 1732 1733
    SMqHbTopicInfo topicInfo;
    buf = taosDecodeSMqHbTopicInfoMsg(buf, &topicInfo);
    taosArrayPush(pMsg->pTopics, &topicInfo);
  }
  return buf;
}

L
Liu Jicong 已提交
1734
typedef struct {
1735 1736 1737 1738
  int64_t leftForVer;
  int32_t vgId;
  int64_t consumerId;
  char    topicName[TSDB_TOPIC_FNAME_LEN];
L
Liu Jicong 已提交
1739
  char    cgroup[TSDB_CGROUP_LEN];
1740 1741 1742 1743
  char*   sql;
  char*   logicalPlan;
  char*   physicalPlan;
  char*   qmsg;
L
Liu Jicong 已提交
1744 1745 1746 1747
} SMqSetCVgReq;

static FORCE_INLINE int32_t tEncodeSMqSetCVgReq(void** buf, const SMqSetCVgReq* pReq) {
  int32_t tlen = 0;
L
Liu Jicong 已提交
1748
  tlen += taosEncodeFixedI64(buf, pReq->leftForVer);
L
Liu Jicong 已提交
1749
  tlen += taosEncodeFixedI32(buf, pReq->vgId);
L
Liu Jicong 已提交
1750
  tlen += taosEncodeFixedI64(buf, pReq->consumerId);
L
Liu Jicong 已提交
1751
  tlen += taosEncodeString(buf, pReq->topicName);
L
Liu Jicong 已提交
1752
  tlen += taosEncodeString(buf, pReq->cgroup);
L
Liu Jicong 已提交
1753 1754 1755
  tlen += taosEncodeString(buf, pReq->sql);
  tlen += taosEncodeString(buf, pReq->logicalPlan);
  tlen += taosEncodeString(buf, pReq->physicalPlan);
L
Liu Jicong 已提交
1756
  tlen += taosEncodeString(buf, pReq->qmsg);
L
Liu Jicong 已提交
1757 1758 1759 1760
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqSetCVgReq(void* buf, SMqSetCVgReq* pReq) {
L
Liu Jicong 已提交
1761
  buf = taosDecodeFixedI64(buf, &pReq->leftForVer);
L
Liu Jicong 已提交
1762
  buf = taosDecodeFixedI32(buf, &pReq->vgId);
L
Liu Jicong 已提交
1763
  buf = taosDecodeFixedI64(buf, &pReq->consumerId);
L
Liu Jicong 已提交
1764
  buf = taosDecodeStringTo(buf, pReq->topicName);
L
Liu Jicong 已提交
1765
  buf = taosDecodeStringTo(buf, pReq->cgroup);
L
Liu Jicong 已提交
1766 1767 1768
  buf = taosDecodeString(buf, &pReq->sql);
  buf = taosDecodeString(buf, &pReq->logicalPlan);
  buf = taosDecodeString(buf, &pReq->physicalPlan);
L
Liu Jicong 已提交
1769
  buf = taosDecodeString(buf, &pReq->qmsg);
L
Liu Jicong 已提交
1770 1771 1772
  return buf;
}

L
Liu Jicong 已提交
1773
typedef struct {
1774 1775 1776 1777
  int64_t leftForVer;
  int32_t vgId;
  int64_t oldConsumerId;
  int64_t newConsumerId;
L
Liu Jicong 已提交
1778 1779 1780 1781 1782 1783 1784 1785 1786 1787 1788 1789 1790 1791 1792 1793 1794 1795 1796 1797
} 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);
  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);
  return buf;
}

typedef struct {
L
Liu Jicong 已提交
1798 1799 1800 1801
  SMsgHead header;
  int32_t  vgId;
  int64_t  consumerId;
  char     topicName[TSDB_TOPIC_FNAME_LEN];
L
Liu Jicong 已提交
1802
  char     cgroup[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
1803 1804
} SMqSetCVgRsp;

L
Liu Jicong 已提交
1805 1806 1807 1808 1809
typedef struct {
  SMsgHead header;
  int32_t  vgId;
  int64_t  consumerId;
  char     topicName[TSDB_TOPIC_FNAME_LEN];
L
Liu Jicong 已提交
1810
  char     cgroup[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
1811 1812
} SMqMVRebRsp;

L
Liu Jicong 已提交
1813 1814 1815 1816
typedef struct {
  int32_t vgId;
  int64_t offset;
  char    topicName[TSDB_TOPIC_FNAME_LEN];
L
Liu Jicong 已提交
1817
  char    cgroup[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
1818 1819 1820 1821 1822
} SMqOffset;

typedef struct {
  int32_t    num;
  SMqOffset* offsets;
L
Liu Jicong 已提交
1823
} SMqCMCommitOffsetReq;
L
Liu Jicong 已提交
1824 1825 1826

typedef struct {
  int32_t reserved;
L
Liu Jicong 已提交
1827
} SMqCMCommitOffsetRsp;
L
Liu Jicong 已提交
1828 1829 1830

int32_t tEncodeSMqOffset(SCoder* encoder, const SMqOffset* pOffset);
int32_t tDecodeSMqOffset(SCoder* decoder, SMqOffset* pOffset);
L
Liu Jicong 已提交
1831 1832
int32_t tEncodeSMqCMCommitOffsetReq(SCoder* encoder, const SMqCMCommitOffsetReq* pReq);
int32_t tDecodeSMqCMCommitOffsetReq(SCoder* decoder, SMqCMCommitOffsetReq* pReq);
L
Liu Jicong 已提交
1833

L
Liu Jicong 已提交
1834 1835
typedef struct {
  uint32_t nCols;
1836
  SSchema* pSchema;
L
Liu Jicong 已提交
1837 1838
} SSchemaWrapper;

S
Shengliang Guan 已提交
1839
static FORCE_INLINE int32_t taosEncodeSSchema(void** buf, const SSchema* pSchema) {
L
Liu Jicong 已提交
1840 1841 1842 1843 1844 1845 1846 1847
  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;
}

S
Shengliang Guan 已提交
1848
static FORCE_INLINE void* taosDecodeSSchema(void* buf, SSchema* pSchema) {
L
Liu Jicong 已提交
1849 1850 1851 1852 1853 1854 1855
  buf = taosDecodeFixedI8(buf, &pSchema->type);
  buf = taosDecodeFixedI32(buf, &pSchema->bytes);
  buf = taosDecodeFixedI32(buf, &pSchema->colId);
  buf = taosDecodeStringTo(buf, pSchema->name);
  return buf;
}

S
Shengliang Guan 已提交
1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871
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;
}

L
Liu Jicong 已提交
1872 1873 1874
static FORCE_INLINE int32_t tEncodeSSchemaWrapper(void** buf, const SSchemaWrapper* pSW) {
  int32_t tlen = 0;
  tlen += taosEncodeFixedU32(buf, pSW->nCols);
1875
  for (int32_t i = 0; i < pSW->nCols; i++) {
S
Shengliang Guan 已提交
1876
    tlen += taosEncodeSSchema(buf, &pSW->pSchema[i]);
L
Liu Jicong 已提交
1877 1878 1879 1880 1881 1882
  }
  return tlen;
}

static FORCE_INLINE void* tDecodeSSchemaWrapper(void* buf, SSchemaWrapper* pSW) {
  buf = taosDecodeFixedU32(buf, &pSW->nCols);
1883
  pSW->pSchema = (SSchema*)calloc(pSW->nCols, sizeof(SSchema));
L
Liu Jicong 已提交
1884 1885 1886
  if (pSW->pSchema == NULL) {
    return NULL;
  }
1887

1888
  for (int32_t i = 0; i < pSW->nCols; i++) {
S
Shengliang Guan 已提交
1889
    buf = taosDecodeSSchema(buf, &pSW->pSchema[i]);
L
Liu Jicong 已提交
1890 1891 1892
  }
  return buf;
}
C
Cary Xu 已提交
1893
typedef struct {
C
Cary Xu 已提交
1894
  int8_t   version;  // for compatibility(default 0)
C
Cary Xu 已提交
1895 1896
  int8_t   intervalUnit; // MACRO: TIME_UNIT_XXX
  int8_t   slidingUnit; // MACRO: TIME_UNIT_XXX
C
Cary Xu 已提交
1897
  char     indexName[TSDB_INDEX_NAME_LEN];
C
Cary Xu 已提交
1898
  char     timezone[TD_TIMEZONE_LEN];  // sma data expired if timezone changes.
X
Xiaoyu Wang 已提交
1899 1900
  int32_t  exprLen;
  int32_t  tagsFilterLen;
C
Cary Xu 已提交
1901 1902 1903
  int64_t  indexUid;
  tb_uid_t tableUid;  // super/child/common table uid
  int64_t  interval;
C
Cary Xu 已提交
1904
  int64_t  offset; // use unit by precision of DB
C
Cary Xu 已提交
1905 1906 1907
  int64_t  sliding;
  char*    expr;  // sma expression
  char*    tagsFilter;
C
Cary Xu 已提交
1908
} STSma;  // Time-range-wise SMA
C
Cary Xu 已提交
1909

C
Cary Xu 已提交
1910
typedef struct {
C
Cary Xu 已提交
1911 1912 1913
  int64_t ver;  // use a general definition
  STSma   tSma;
} SVCreateTSmaReq;
C
Cary Xu 已提交
1914 1915

typedef struct {
L
Liu Jicong 已提交
1916
  int8_t      type;                            // 0 status report, 1 update data
C
Cary Xu 已提交
1917
  char        indexName[TSDB_INDEX_NAME_LEN];  //
C
Cary Xu 已提交
1918 1919 1920 1921 1922
  STimeWindow windows;
} STSmaMsg;

typedef struct {
  int64_t ver;  // use a general definition
C
Cary Xu 已提交
1923
  char    indexName[TSDB_INDEX_NAME_LEN];
C
Cary Xu 已提交
1924
} SVDropTSmaReq;
1925

C
Cary Xu 已提交
1926
typedef struct {
L
Liu Jicong 已提交
1927
  int tmp;  // TODO: to avoid compile error
C
Cary Xu 已提交
1928 1929 1930 1931 1932 1933
} SVCreateTSmaRsp, SVDropTSmaRsp;

int32_t tSerializeSVCreateTSmaReq(void** buf, SVCreateTSmaReq* pReq);
void*   tDeserializeSVCreateTSmaReq(void* buf, SVCreateTSmaReq* pReq);
int32_t tSerializeSVDropTSmaReq(void** buf, SVDropTSmaReq* pReq);
void*   tDeserializeSVDropTSmaReq(void* buf, SVDropTSmaReq* pReq);
C
Cary Xu 已提交
1934 1935

typedef struct {
C
Cary Xu 已提交
1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948
  col_id_t colId;
  uint16_t blockSize;  // sma data block size
  char     data[];
} STSmaColData;

typedef struct {
  tb_uid_t tableUid;  // super/child/normal table uid
  int32_t  dataLen;   // not including head
  char     data[];
} STSmaTbData;

typedef struct {
  int64_t indexUid;
C
Cary Xu 已提交
1949
  TSKEY   skey;  // startKey of one interval/sliding window
C
Cary Xu 已提交
1950 1951 1952 1953 1954 1955 1956 1957 1958 1959
  int64_t interval;
  int32_t dataLen;  // not including head
  int8_t  intervalUnit;
  char    data[];
} STSmaDataWrapper;  // sma data for a interval/sliding window

// interval/sliding => window

// => window->table->colId
// => 当一个window下所有的表均计算完成时,流计算告知tsdb清除window的过期标记
C
Cary Xu 已提交
1960

C
Cary Xu 已提交
1961
// RSma: Rollup SMA
C
Cary Xu 已提交
1962 1963 1964 1965 1966 1967 1968 1969 1970 1971 1972 1973 1974
typedef struct {
  int64_t  interval;
  int32_t  retention;  // unit: day
  uint16_t days;       // unit: day
  int8_t   intervalUnit;
} SSmaParams;

typedef struct {
  STSma   tsma;
  float   xFilesFactor;
  SArray* smaParams;  // SSmaParams
} SRSma;

C
Cary Xu 已提交
1975 1976 1977 1978 1979
typedef struct {
  uint32_t number;
  STSma*   tSma;
} STSmaWrapper;

C
Cary Xu 已提交
1980
static FORCE_INLINE void tdDestroyTSma(STSma* pSma) {
C
Cary Xu 已提交
1981
  if (pSma) {
C
Cary Xu 已提交
1982
    tfree(pSma->expr);
C
Cary Xu 已提交
1983
    tfree(pSma->tagsFilter);
C
Cary Xu 已提交
1984 1985 1986
  }
}

C
Cary Xu 已提交
1987
static FORCE_INLINE void tdDestroyTSmaWrapper(STSmaWrapper* pSW) {
C
Cary Xu 已提交
1988 1989 1990
  if (pSW) {
    if (pSW->tSma) {
      for (uint32_t i = 0; i < pSW->number; ++i) {
C
Cary Xu 已提交
1991
        tdDestroyTSma(pSW->tSma + i);
C
Cary Xu 已提交
1992 1993 1994
      }
      tfree(pSW->tSma);
    }
C
Cary Xu 已提交
1995 1996 1997 1998 1999 2000
  }
}

static FORCE_INLINE int32_t tEncodeTSma(void** buf, const STSma* pSma) {
  int32_t tlen = 0;

C
Cary Xu 已提交
2001 2002 2003
  tlen += taosEncodeFixedI8(buf, pSma->version);
  tlen += taosEncodeFixedI8(buf, pSma->intervalUnit);
  tlen += taosEncodeFixedI8(buf, pSma->slidingUnit);
C
Cary Xu 已提交
2004
  tlen += taosEncodeString(buf, pSma->indexName);
C
Cary Xu 已提交
2005
  tlen += taosEncodeString(buf, pSma->timezone);
X
Xiaoyu Wang 已提交
2006 2007
  tlen += taosEncodeFixedI32(buf, pSma->exprLen);
  tlen += taosEncodeFixedI32(buf, pSma->tagsFilterLen);
C
Cary Xu 已提交
2008
  tlen += taosEncodeFixedI64(buf, pSma->indexUid);
C
Cary Xu 已提交
2009
  tlen += taosEncodeFixedI64(buf, pSma->tableUid);
C
Cary Xu 已提交
2010
  tlen += taosEncodeFixedI64(buf, pSma->interval);
X
Xiaoyu Wang 已提交
2011
  tlen += taosEncodeFixedI64(buf, pSma->offset);
C
Cary Xu 已提交
2012
  tlen += taosEncodeFixedI64(buf, pSma->sliding);
L
Liu Jicong 已提交
2013

C
Cary Xu 已提交
2014 2015
  if (pSma->exprLen > 0) {
    tlen += taosEncodeString(buf, pSma->expr);
C
Cary Xu 已提交
2016 2017
  }

C
Cary Xu 已提交
2018 2019
  if (pSma->tagsFilterLen > 0) {
    tlen += taosEncodeString(buf, pSma->tagsFilter);
C
Cary Xu 已提交
2020 2021 2022 2023 2024 2025 2026 2027 2028 2029 2030 2031 2032 2033 2034 2035
  }

  return tlen;
}

static FORCE_INLINE int32_t tEncodeTSmaWrapper(void** buf, const STSmaWrapper* pSW) {
  int32_t tlen = 0;

  tlen += taosEncodeFixedU32(buf, pSW->number);
  for (uint32_t i = 0; i < pSW->number; ++i) {
    tlen += tEncodeTSma(buf, pSW->tSma + i);
  }
  return tlen;
}

static FORCE_INLINE void* tDecodeTSma(void* buf, STSma* pSma) {
C
Cary Xu 已提交
2036 2037 2038
  buf = taosDecodeFixedI8(buf, &pSma->version);
  buf = taosDecodeFixedI8(buf, &pSma->intervalUnit);
  buf = taosDecodeFixedI8(buf, &pSma->slidingUnit);
C
Cary Xu 已提交
2039
  buf = taosDecodeStringTo(buf, pSma->indexName);
C
Cary Xu 已提交
2040
  buf = taosDecodeStringTo(buf, pSma->timezone);
X
Xiaoyu Wang 已提交
2041 2042
  buf = taosDecodeFixedI32(buf, &pSma->exprLen);
  buf = taosDecodeFixedI32(buf, &pSma->tagsFilterLen);
C
Cary Xu 已提交
2043
  buf = taosDecodeFixedI64(buf, &pSma->indexUid);
C
Cary Xu 已提交
2044
  buf = taosDecodeFixedI64(buf, &pSma->tableUid);
C
Cary Xu 已提交
2045
  buf = taosDecodeFixedI64(buf, &pSma->interval);
X
Xiaoyu Wang 已提交
2046
  buf = taosDecodeFixedI64(buf, &pSma->offset);
C
Cary Xu 已提交
2047 2048
  buf = taosDecodeFixedI64(buf, &pSma->sliding);

C
Cary Xu 已提交
2049 2050 2051 2052 2053
  if (pSma->exprLen > 0) {
    pSma->expr = (char*)calloc(pSma->exprLen, 1);
    if (pSma->expr != NULL) {
      buf = taosDecodeStringTo(buf, pSma->expr);
    } else {
C
Cary Xu 已提交
2054
      tdDestroyTSma(pSma);
C
Cary Xu 已提交
2055 2056
      return NULL;
    }
C
Cary Xu 已提交
2057

C
Cary Xu 已提交
2058
  } else {
C
Cary Xu 已提交
2059
    pSma->expr = NULL;
C
Cary Xu 已提交
2060 2061
  }

C
Cary Xu 已提交
2062 2063 2064 2065 2066 2067
  if (pSma->tagsFilterLen > 0) {
    pSma->tagsFilter = (char*)calloc(pSma->tagsFilterLen, 1);
    if (pSma->tagsFilter != NULL) {
      buf = taosDecodeStringTo(buf, pSma->tagsFilter);
    } else {
      tdDestroyTSma(pSma);
C
Cary Xu 已提交
2068 2069
      return NULL;
    }
C
Cary Xu 已提交
2070

C
Cary Xu 已提交
2071
  } else {
C
Cary Xu 已提交
2072
    pSma->tagsFilter = NULL;
C
Cary Xu 已提交
2073 2074 2075 2076 2077 2078 2079 2080 2081 2082 2083 2084 2085 2086 2087 2088
  }

  return buf;
}

static FORCE_INLINE void* tDecodeTSmaWrapper(void* buf, STSmaWrapper* pSW) {
  buf = taosDecodeFixedU32(buf, &pSW->number);

  pSW->tSma = (STSma*)calloc(pSW->number, sizeof(STSma));
  if (pSW->tSma == NULL) {
    return NULL;
  }

  for (uint32_t i = 0; i < pSW->number; ++i) {
    if ((buf = tDecodeTSma(buf, pSW->tSma + i)) == NULL) {
      for (uint32_t j = i; j >= 0; --i) {
C
Cary Xu 已提交
2089
        tdDestroyTSma(pSW->tSma + j);
C
Cary Xu 已提交
2090 2091 2092 2093 2094 2095 2096
      }
      free(pSW->tSma);
      return NULL;
    }
  }
  return buf;
}
L
Liu Jicong 已提交
2097

L
Liu Jicong 已提交
2098
typedef struct {
2099 2100 2101
  int64_t uid;
  int32_t numOfRows;
  char*   colData;
L
Liu Jicong 已提交
2102 2103
} SMqTbData;

L
Liu Jicong 已提交
2104
typedef struct {
L
Liu Jicong 已提交
2105 2106 2107 2108 2109 2110 2111 2112
  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;
L
Liu Jicong 已提交
2113 2114
} SMqTopicData;

L
Liu Jicong 已提交
2115 2116 2117 2118 2119 2120
typedef struct {
  int8_t  mqMsgType;
  int32_t code;
  int32_t epoch;
} SMqRspHead;

L
Liu Jicong 已提交
2121
typedef struct {
L
Liu Jicong 已提交
2122 2123
  int64_t         consumerId;
  SSchemaWrapper* schemas;
L
Liu Jicong 已提交
2124 2125 2126
  int64_t         reqOffset;
  int64_t         rspOffset;
  int32_t         skipLogNum;
L
Liu Jicong 已提交
2127
  int32_t         numOfTopics;
2128
  SArray*         pBlockData;  // SArray<SSDataBlock>
L
Liu Jicong 已提交
2129
} SMqPollRsp;
L
Liu Jicong 已提交
2130

L
Liu Jicong 已提交
2131
// one req for one vg+topic
L
Liu Jicong 已提交
2132
typedef struct {
2133
  SMsgHead head;
L
Liu Jicong 已提交
2134

2135 2136
  int64_t consumerId;
  int64_t blockingTime;
L
Liu Jicong 已提交
2137
  int32_t epoch;
L
Liu Jicong 已提交
2138
  char    cgroup[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
2139

L
Liu Jicong 已提交
2140
  int64_t currentOffset;
2141
  char    topic[TSDB_TOPIC_FNAME_LEN];
L
Liu Jicong 已提交
2142
} SMqPollReq;
L
Liu Jicong 已提交
2143

L
Liu Jicong 已提交
2144
typedef struct {
L
Liu Jicong 已提交
2145
  int32_t vgId;
L
Liu Jicong 已提交
2146
  int64_t offset;
L
Liu Jicong 已提交
2147 2148 2149
  SEpSet  epSet;
} SMqSubVgEp;

L
Liu Jicong 已提交
2150
typedef struct {
L
Liu Jicong 已提交
2151
  char    topic[TSDB_TOPIC_FNAME_LEN];
2152
  SArray* vgs;  // SArray<SMqSubVgEp>
L
Liu Jicong 已提交
2153 2154
} SMqSubTopicEp;

L
Liu Jicong 已提交
2155
typedef struct {
L
Liu Jicong 已提交
2156
  int64_t consumerId;
L
Liu Jicong 已提交
2157
  char    cgroup[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
2158 2159 2160
  SArray* topics;  // SArray<SMqSubTopicEp>
} SMqCMGetSubEpRsp;

L
Liu Jicong 已提交
2161
typedef struct {
L
Liu Jicong 已提交
2162 2163
  SMqRspHead head;
  union {
L
Liu Jicong 已提交
2164
    SMqPollRsp       consumeRsp;
L
Liu Jicong 已提交
2165 2166
    SMqCMGetSubEpRsp getEpRsp;
  };
L
Liu Jicong 已提交
2167
  void* extra;
L
Liu Jicong 已提交
2168 2169 2170 2171 2172 2173 2174 2175 2176 2177 2178 2179 2180 2181 2182 2183 2184 2185 2186 2187 2188 2189 2190
} SMqMsgWrapper;

typedef struct {
  int32_t curBlock;
  int32_t curRow;
  void**  uData;
} SMqRowIter;

struct tmq_message_t_v1 {
  SMqPollRsp rsp;
  SMqRowIter iter;
};

struct tmq_message_t {
  SMqRspHead head;
  union {
    SMqPollRsp       consumeRsp;
    SMqCMGetSubEpRsp getEpRsp;
  };
  void*   extra;
  int32_t curBlock;
  int32_t curRow;
  void**  uData;
L
Liu Jicong 已提交
2191 2192
};

2193
static FORCE_INLINE void tDeleteSMqSubTopicEp(SMqSubTopicEp* pSubTopicEp) { taosArrayDestroy(pSubTopicEp->vgs); }
L
Liu Jicong 已提交
2194

L
Liu Jicong 已提交
2195 2196
static FORCE_INLINE int32_t tEncodeSMqSubVgEp(void** buf, const SMqSubVgEp* pVgEp) {
  int32_t tlen = 0;
L
Liu Jicong 已提交
2197
  tlen += taosEncodeFixedI32(buf, pVgEp->vgId);
L
Liu Jicong 已提交
2198
  tlen += taosEncodeFixedI64(buf, pVgEp->offset);
L
Liu Jicong 已提交
2199 2200 2201 2202 2203 2204
  tlen += taosEncodeSEpSet(buf, &pVgEp->epSet);
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqSubVgEp(void* buf, SMqSubVgEp* pVgEp) {
  buf = taosDecodeFixedI32(buf, &pVgEp->vgId);
L
Liu Jicong 已提交
2205
  buf = taosDecodeFixedI64(buf, &pVgEp->offset);
L
Liu Jicong 已提交
2206 2207 2208 2209
  buf = taosDecodeSEpSet(buf, &pVgEp->epSet);
  return buf;
}

L
Liu Jicong 已提交
2210
static FORCE_INLINE void tDeleteSMqCMGetSubEpRsp(SMqCMGetSubEpRsp* pRsp) {
2211
  taosArrayDestroyEx(pRsp->topics, (void (*)(void*))tDeleteSMqSubTopicEp);
L
Liu Jicong 已提交
2212 2213
}

L
Liu Jicong 已提交
2214 2215 2216 2217 2218 2219 2220 2221 2222 2223 2224 2225 2226 2227 2228 2229 2230 2231 2232 2233 2234 2235 2236 2237 2238 2239 2240 2241 2242 2243 2244 2245 2246 2247 2248 2249 2250 2251 2252 2253 2254 2255 2256 2257 2258 2259 2260 2261 2262 2263 2264 2265 2266 2267 2268 2269 2270
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 += 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 = 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;
}
L
Liu Jicong 已提交
2271

L
Liu Jicong 已提交
2272 2273 2274 2275 2276 2277 2278 2279 2280 2281 2282 2283 2284 2285 2286 2287 2288 2289 2290 2291 2292 2293 2294 2295 2296 2297 2298 2299 2300 2301 2302 2303 2304 2305 2306 2307 2308 2309 2310 2311 2312 2313 2314 2315 2316
enum {
  STREAM_TASK_STATUS__RUNNING = 1,
  STREAM_TASK_STATUS__STOP,
};

typedef struct {
  int64_t streamId;
  int32_t taskId;
  int32_t level;
  int8_t  status;
  char*   qmsg;
  void*   executor;
  // void*   stateStore;
  //  storage handle
} SStreamTask;

static FORCE_INLINE SStreamTask* streamTaskNew(int64_t streamId, int32_t level) {
  SStreamTask* pTask = (SStreamTask*)calloc(1, sizeof(SStreamTask));
  if (pTask == NULL) {
    return NULL;
  }
  pTask->taskId = tGenIdPI32();
  pTask->status = STREAM_TASK_STATUS__RUNNING;
  pTask->qmsg = NULL;
  return pTask;
}

int32_t tEncodeSStreamTask(SCoder* pEncoder, const SStreamTask* pTask);
int32_t tDecodeSStreamTask(SCoder* pDecoder, SStreamTask* pTask);
void    tFreeSStreamTask(SStreamTask* pTask);

typedef struct {
  SMsgHead     head;
  SStreamTask* task;
} SStreamTaskDeployReq;

typedef struct {
  int32_t reserved;
} SStreamTaskDeployRsp;

typedef struct {
  SMsgHead head;
  // TODO: other info needed by task
} SStreamTaskExecReq;

L
Liu Jicong 已提交
2317 2318 2319 2320
typedef struct {
  int32_t reserved;
} SStreamTaskExecRsp;

L
Liu Jicong 已提交
2321 2322
#pragma pack(pop)

D
dapan1121 已提交
2323 2324 2325 2326 2327
#ifdef __cplusplus
}
#endif

#endif /*_TD_COMMON_TAOS_MSG_H_*/