mndDef.h 16.3 KB
Newer Older
H
refact  
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * 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/>.
 */

S
Shengliang Guan 已提交
16 17
#ifndef _TD_MND_DEF_H_
#define _TD_MND_DEF_H_
S
Shengliang Guan 已提交
18

19
#include "os.h"
S
Shengliang Guan 已提交
20 21

#include "cJSON.h"
L
Liu Jicong 已提交
22
#include "scheduler.h"
S
Shengliang Guan 已提交
23 24
#include "sync.h"
#include "thash.h"
L
Liu Jicong 已提交
25
#include "tlist.h"
S
Shengliang Guan 已提交
26
#include "tlog.h"
L
Liu Jicong 已提交
27
#include "tmsg.h"
S
Shengliang Guan 已提交
28
#include "trpc.h"
L
Liu Jicong 已提交
29
#include "tstream.h"
S
Shengliang Guan 已提交
30
#include "ttimer.h"
S
Shengliang Guan 已提交
31

S
Shengliang Guan 已提交
32
#include "mnode.h"
S
Shengliang Guan 已提交
33

S
Shengliang Guan 已提交
34 35 36
#ifdef __cplusplus
extern "C" {
#endif
S
Shengliang Guan 已提交
37

38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73
typedef enum {
  MND_OPER_CONNECT = 1,
  MND_OPER_CREATE_ACCT,
  MND_OPER_DROP_ACCT,
  MND_OPER_ALTER_ACCT,
  MND_OPER_CREATE_USER,
  MND_OPER_DROP_USER,
  MND_OPER_ALTER_USER,
  MND_OPER_CREATE_DNODE,
  MND_OPER_DROP_DNODE,
  MND_OPER_CONFIG_DNODE,
  MND_OPER_CREATE_MNODE,
  MND_OPER_DROP_MNODE,
  MND_OPER_CREATE_QNODE,
  MND_OPER_DROP_QNODE,
  MND_OPER_CREATE_SNODE,
  MND_OPER_DROP_SNODE,
  MND_OPER_REDISTRIBUTE_VGROUP,
  MND_OPER_MERGE_VGROUP,
  MND_OPER_SPLIT_VGROUP,
  MND_OPER_BALANCE_VGROUP,
  MND_OPER_CREATE_FUNC,
  MND_OPER_DROP_FUNC,
  MND_OPER_KILL_TRANS,
  MND_OPER_KILL_CONN,
  MND_OPER_KILL_QUERY,
  MND_OPER_CREATE_DB,
  MND_OPER_ALTER_DB,
  MND_OPER_DROP_DB,
  MND_OPER_COMPACT_DB,
  MND_OPER_TRIM_DB,
  MND_OPER_USE_DB,
  MND_OPER_WRITE_DB,
  MND_OPER_READ_DB,
  MND_OPER_READ_OR_WRITE_DB,
  MND_OPER_SHOW_VARIBALES,
74
  MND_OPER_SUBSCRIBE,
S
Shengliang Guan 已提交
75 76
  MND_OPER_CREATE_TOPIC,
  MND_OPER_DROP_TOPIC,
77 78
} EOperType;

S
Shengliang Guan 已提交
79
typedef enum {
80 81 82 83 84 85 86 87
  MND_AUTH_ACCT_START = 0,
  MND_AUTH_ACCT_USER,
  MND_AUTH_ACCT_DNODE,
  MND_AUTH_ACCT_MNODE,
  MND_AUTH_ACCT_DB,
  MND_AUTH_ACCT_TABLE,
  MND_AUTH_ACCT_MAX
} EAuthAcct;
S
Shengliang Guan 已提交
88 89

typedef enum {
90 91 92 93 94 95
  MND_AUTH_OP_START = 0,
  MND_AUTH_OP_CREATE_USER,
  MND_AUTH_OP_ALTER_USER,
  MND_AUTH_OP_DROP_USER,
  MND_AUTH_MAX
} EAuthOp;
S
Shengliang Guan 已提交
96

S
Shengliang Guan 已提交
97
typedef enum {
98 99 100
  TRN_CONFLICT_NOTHING = 0,
  TRN_CONFLICT_GLOBAL = 1,
  TRN_CONFLICT_DB = 2,
101
  TRN_CONFLICT_DB_INSIDE = 3,
102
} ETrnConflct;
S
Shengliang Guan 已提交
103

S
Shengliang Guan 已提交
104
typedef enum {
105
  TRN_STAGE_PREPARE = 0,
106 107 108 109 110 111
  TRN_STAGE_REDO_ACTION = 1,
  TRN_STAGE_ROLLBACK = 2,
  TRN_STAGE_UNDO_ACTION = 3,
  TRN_STAGE_COMMIT = 4,
  TRN_STAGE_COMMIT_ACTION = 5,
  TRN_STAGE_FINISHED = 6
S
Shengliang Guan 已提交
112 113
} ETrnStage;

S
Shengliang Guan 已提交
114 115 116 117
typedef enum {
  TRN_POLICY_ROLLBACK = 0,
  TRN_POLICY_RETRY = 1,
} ETrnPolicy;
S
Shengliang Guan 已提交
118

119
typedef enum {
120 121 122
  TRN_EXEC_PRARLLEL = 0,
  TRN_EXEC_SERIAL = 1,
} ETrnExec;
123

S
Shengliang Guan 已提交
124 125 126 127 128 129 130 131 132 133 134 135 136 137
typedef enum {
  DND_REASON_ONLINE = 0,
  DND_REASON_STATUS_MSG_TIMEOUT,
  DND_REASON_STATUS_NOT_RECEIVED,
  DND_REASON_VERSION_NOT_MATCH,
  DND_REASON_DNODE_ID_NOT_MATCH,
  DND_REASON_CLUSTER_ID_NOT_MATCH,
  DND_REASON_STATUS_INTERVAL_NOT_MATCH,
  DND_REASON_TIME_ZONE_NOT_MATCH,
  DND_REASON_LOCALE_NOT_MATCH,
  DND_REASON_CHARSET_NOT_MATCH,
  DND_REASON_OTHERS
} EDndReason;

S
Shengliang Guan 已提交
138 139 140 141 142 143 144 145 146
typedef enum {
  CONSUMER_UPDATE__TOUCH = 1,
  CONSUMER_UPDATE__ADD,
  CONSUMER_UPDATE__REMOVE,
  CONSUMER_UPDATE__LOST,
  CONSUMER_UPDATE__RECOVER,
  CONSUMER_UPDATE__MODIFY,
} ECsmUpdateType;

S
Shengliang Guan 已提交
147
typedef struct {
L
Liu Jicong 已提交
148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166
  int32_t     id;
  ETrnStage   stage;
  ETrnPolicy  policy;
  ETrnConflct conflict;
  ETrnExec    exec;
  EOperType   oper;
  int32_t     code;
  int32_t     failedTimes;
  void*       rpcRsp;
  int32_t     rpcRspLen;
  int32_t     redoActionPos;
  SArray*     redoActions;
  SArray*     undoActions;
  SArray*     commitActions;
  int64_t     createdTime;
  int64_t     lastExecTime;
  int32_t     lastAction;
  int32_t     lastErrorNo;
  SEpSet      lastEpset;
167 168
  tmsg_t      lastMsgType;
  tmsg_t      originRpcType;
169 170
  char        dbname[TSDB_TABLE_FNAME_LEN];
  char        stbname[TSDB_TABLE_FNAME_LEN];
L
Liu Jicong 已提交
171 172 173 174
  int32_t     startFunc;
  int32_t     stopFunc;
  int32_t     paramLen;
  void*       param;
175
  char        opername[TSDB_TRANS_OPER_LEN];
L
Liu Jicong 已提交
176
  SArray*     pRpcArray;
177
  SRWLatch    lockRpcArray;
S
Shengliang Guan 已提交
178
} STrans;
S
Shengliang Guan 已提交
179

S
Shengliang Guan 已提交
180
typedef struct {
181
  int64_t id;
S
Shengliang Guan 已提交
182
  char    name[TSDB_CLUSTER_ID_LEN];
S
Shengliang Guan 已提交
183 184
  int64_t createdTime;
  int64_t updateTime;
185
  int32_t upTime;
S
Shengliang Guan 已提交
186 187
} SClusterObj;

S
Shengliang Guan 已提交
188
typedef struct {
S
Shengliang Guan 已提交
189 190 191 192
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
  int64_t    rebootTime;
193
  int64_t    lastAccessTime;
S
Shengliang Guan 已提交
194
  int32_t    accessTimes;
S
Shengliang Guan 已提交
195
  int32_t    numOfVnodes;
S
Shengliang Guan 已提交
196
  int32_t    numOfSupportVnodes;
S
Shengliang Guan 已提交
197
  float      numOfCores;
198 199 200
  int64_t    memTotal;
  int64_t    memAvail;
  int64_t    memUsed;
S
Shengliang Guan 已提交
201 202 203 204
  EDndReason offlineReason;
  uint16_t   port;
  char       fqdn[TSDB_FQDN_LEN];
  char       ep[TSDB_EP_LEN];
S
Shengliang Guan 已提交
205 206
} SDnodeObj;

S
Shengliang Guan 已提交
207
typedef struct {
S
Shengliang Guan 已提交
208
  int32_t    id;
S
Shengliang Guan 已提交
209 210
  int64_t    createdTime;
  int64_t    updateTime;
211 212
  ESyncState syncState;
  bool       syncRestore;
213
  int64_t    stateStartTime;
L
Liu Jicong 已提交
214
  SDnodeObj* pDnode;
S
Shengliang Guan 已提交
215 216
} SMnodeObj;

S
Shengliang Guan 已提交
217 218 219 220
typedef struct {
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
L
Liu Jicong 已提交
221
  SDnodeObj* pDnode;
L
Liu Jicong 已提交
222
  SQnodeLoad load;
S
Shengliang Guan 已提交
223 224 225 226 227 228
} SQnodeObj;

typedef struct {
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
L
Liu Jicong 已提交
229
  SDnodeObj* pDnode;
S
Shengliang Guan 已提交
230 231
} SSnodeObj;

S
Shengliang Guan 已提交
232 233 234
typedef struct {
  int32_t maxUsers;
  int32_t maxDbs;
S
Shengliang Guan 已提交
235 236
  int32_t maxStbs;
  int32_t maxTbs;
S
Shengliang Guan 已提交
237 238
  int32_t maxTimeSeries;
  int32_t maxStreams;
S
Shengliang Guan 已提交
239 240 241 242
  int32_t maxFuncs;
  int32_t maxConsumers;
  int32_t maxConns;
  int32_t maxTopics;
243
  int64_t maxStorage;
S
Shengliang Guan 已提交
244
  int32_t accessState;  // Configured only by command
S
Shengliang Guan 已提交
245 246 247 248 249 250 251
} SAcctCfg;

typedef struct {
  int32_t numOfUsers;
  int32_t numOfDbs;
  int32_t numOfTimeSeries;
  int32_t numOfStreams;
S
Shengliang Guan 已提交
252 253
  int64_t totalStorage;  // Total storage wrtten from this account
  int64_t compStorage;   // Compressed storage on disk
S
Shengliang Guan 已提交
254 255
} SAcctInfo;

S
Shengliang Guan 已提交
256
typedef struct {
S
Shengliang Guan 已提交
257 258 259 260
  char      acct[TSDB_USER_LEN];
  int64_t   createdTime;
  int64_t   updateTime;
  int32_t   acctId;
S
Shengliang Guan 已提交
261
  int32_t   status;
S
Shengliang Guan 已提交
262 263 264 265
  SAcctCfg  cfg;
  SAcctInfo info;
} SAcctObj;

S
Shengliang Guan 已提交
266
typedef struct {
S
Shengliang Guan 已提交
267
  char      user[TSDB_USER_LEN];
268
  char      pass[TSDB_PASSWORD_LEN];
S
Shengliang Guan 已提交
269 270 271
  char      acct[TSDB_USER_LEN];
  int64_t   createdTime;
  int64_t   updateTime;
272
  int8_t    superUser;
273 274 275
  int8_t    sysInfo;
  int8_t    enable;
  int8_t    reserve;
S
Shengliang Guan 已提交
276
  int32_t   acctId;
D
dapan 已提交
277
  int32_t   authVersion;
278 279
  SHashObj* readDbs;
  SHashObj* writeDbs;
280
  SHashObj* topics;
S
Shengliang Guan 已提交
281
  SRWLatch  lock;
S
Shengliang Guan 已提交
282 283 284
} SUserObj;

typedef struct {
285
  int32_t numOfVgroups;
S
Shengliang Guan 已提交
286 287 288 289
  int32_t numOfStables;
  int32_t buffer;
  int32_t pageSize;
  int32_t pages;
290
  int32_t cacheLastSize;
S
Shengliang Guan 已提交
291 292 293 294
  int32_t daysPerFile;
  int32_t daysToKeep0;
  int32_t daysToKeep1;
  int32_t daysToKeep2;
S
Shengliang Guan 已提交
295 296
  int32_t minRows;
  int32_t maxRows;
297
  int32_t walFsyncPeriod;
S
Shengliang Guan 已提交
298
  int8_t  walLevel;
S
Shengliang Guan 已提交
299 300 301
  int8_t  precision;
  int8_t  compression;
  int8_t  replications;
S
Shengliang Guan 已提交
302 303
  int8_t  strict;
  int8_t  hashMethod;  // default is 1
304
  int8_t  cacheLast;
S
Shengliang Guan 已提交
305
  int8_t  schemaless;
306 307 308
  int16_t hashPrefix;
  int16_t hashSuffix;
  int16_t sstTrigger;
309
  int32_t tsdbPageSize;
S
sma  
Shengliang Guan 已提交
310 311
  int32_t numOfRetensions;
  SArray* pRetensions;
S
Shengliang Guan 已提交
312 313
  int32_t walRetentionPeriod;
  int32_t walRollPeriod;
314
  int64_t walRetentionSize;
L
Liu Jicong 已提交
315
  int64_t walSegmentSize;
S
Shengliang Guan 已提交
316 317
} SDbCfg;

S
Shengliang Guan 已提交
318
typedef struct {
319 320 321 322 323 324 325 326 327 328
  char     name[TSDB_DB_FNAME_LEN];
  char     acct[TSDB_USER_LEN];
  char     createUser[TSDB_USER_LEN];
  int64_t  createdTime;
  int64_t  updateTime;
  int64_t  uid;
  int32_t  cfgVersion;
  int32_t  vgVersion;
  SDbCfg   cfg;
  SRWLatch lock;
329
  int64_t  stateTs;
330
  int64_t  compactStartTime;
S
Shengliang Guan 已提交
331 332 333 334
} SDbObj;

typedef struct {
  int32_t    dnodeId;
335 336
  ESyncState syncState;
  bool       syncRestore;
337
  bool       syncCanRead;
S
Shengliang Guan 已提交
338 339
} SVnodeGid;

S
Shengliang Guan 已提交
340
typedef struct {
S
Shengliang Guan 已提交
341
  int32_t   vgId;
S
Shengliang Guan 已提交
342 343
  int64_t   createdTime;
  int64_t   updateTime;
S
Shengliang Guan 已提交
344
  int32_t   version;
S
Shengliang Guan 已提交
345 346
  uint32_t  hashBegin;
  uint32_t  hashEnd;
347
  char      dbName[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
348
  int64_t   dbUid;
349
  int64_t   cacheUsage;
S
Shengliang Guan 已提交
350 351
  int64_t   numOfTables;
  int64_t   numOfTimeSeries;
S
Shengliang Guan 已提交
352 353 354
  int64_t   totalStorage;
  int64_t   compStorage;
  int64_t   pointsWritten;
S
Shengliang Guan 已提交
355
  int8_t    compact;
S
Shengliang Guan 已提交
356
  int8_t    isTsma;
S
Shengliang Guan 已提交
357 358
  int8_t    replica;
  SVnodeGid vnodeGid[TSDB_MAX_REPLICA];
C
Cary Xu 已提交
359
  void*     pTsma;
S
Shengliang Guan 已提交
360 361
} SVgObj;

S
Shengliang Guan 已提交
362
typedef struct {
C
Cary Xu 已提交
363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388
  char           name[TSDB_TABLE_FNAME_LEN];
  char           stb[TSDB_TABLE_FNAME_LEN];
  char           db[TSDB_DB_FNAME_LEN];
  char           dstTbName[TSDB_TABLE_FNAME_LEN];
  int64_t        createdTime;
  int64_t        uid;
  int64_t        stbUid;
  int64_t        dbUid;
  int64_t        dstTbUid;
  int8_t         intervalUnit;
  int8_t         slidingUnit;
  int8_t         timezone;
  int32_t        dstVgId;  // for stream
  int64_t        interval;
  int64_t        offset;
  int64_t        sliding;
  int32_t        exprLen;  // strlen + 1
  int32_t        tagsFilterLen;
  int32_t        sqlLen;
  int32_t        astLen;
  char*          expr;
  char*          tagsFilter;
  char*          sql;
  char*          ast;
  SSchemaWrapper schemaRow;  // for dstVgroup
  SSchemaWrapper schemaTag;  // for dstVgroup
S
Shengliang Guan 已提交
389 390
} SSmaObj;

S
Shengliang Guan 已提交
391
typedef struct {
S
Shengliang Guan 已提交
392
  char     name[TSDB_TABLE_FNAME_LEN];
393
  char     db[TSDB_DB_FNAME_LEN];
394 395
  int64_t  createdTime;
  int64_t  updateTime;
L
Liu Jicong 已提交
396 397
  int64_t  uid;
  int64_t  dbUid;
398 399
  int32_t  tagVer;
  int32_t  colVer;
D
dapan1121 已提交
400
  int32_t  smaVer;
S
Shengliang Guan 已提交
401
  int32_t  nextColId;
402
  int64_t  maxdelay[2];
S
Shengliang Guan 已提交
403
  int64_t  watermark[2];
S
sma  
Shengliang Guan 已提交
404
  int32_t  ttl;
S
Shengliang Guan 已提交
405 406
  int32_t  numOfColumns;
  int32_t  numOfTags;
S
Shengliang Guan 已提交
407
  int32_t  numOfFuncs;
S
sma  
Shengliang Guan 已提交
408
  int32_t  commentLen;
409 410
  int32_t  ast1Len;
  int32_t  ast2Len;
D
dapan1121 已提交
411
  SArray*  pFuncs;
S
Shengliang Guan 已提交
412
  SSchema* pColumns;
S
Shengliang Guan 已提交
413
  SSchema* pTags;
S
sma  
Shengliang Guan 已提交
414
  char*    comment;
415 416
  char*    pAst1;
  char*    pAst2;
S
Shengliang Guan 已提交
417
  SRWLatch lock;
S
Shengliang Guan 已提交
418
} SStbObj;
S
Shengliang Guan 已提交
419

S
Shengliang Guan 已提交
420
typedef struct {
S
Shengliang Guan 已提交
421 422
  char    name[TSDB_FUNC_NAME_LEN];
  int64_t createdTime;
S
Shengliang Guan 已提交
423 424 425 426 427 428
  int8_t  funcType;
  int8_t  scriptType;
  int8_t  align;
  int8_t  outputType;
  int32_t outputLen;
  int32_t bufSize;
S
Shengliang 已提交
429
  int64_t signature;
S
Shengliang Guan 已提交
430 431
  int32_t commentSize;
  int32_t codeSize;
L
Liu Jicong 已提交
432 433
  char*   pComment;
  char*   pCode;
S
Shengliang Guan 已提交
434 435
} SFuncObj;

S
Shengliang Guan 已提交
436
typedef struct {
L
Liu Jicong 已提交
437 438 439 440 441 442 443
  int64_t        id;
  int8_t         type;
  int8_t         replica;
  int16_t        numOfColumns;
  int32_t        numOfRows;
  void*          pIter;
  SMnode*        pMnode;
444
  STableMetaRsp* pMeta;
L
Liu Jicong 已提交
445 446
  bool           sysDbRsp;
  char           db[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
447 448
} SShowObj;

449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465
typedef struct {
  int64_t id;
  int8_t  type;
  int8_t  replica;
  int16_t numOfColumns;
  int32_t rowSize;
  int32_t numOfRows;
  int32_t numOfReads;
  int32_t payloadLen;
  void*   pIter;
  SMnode* pMnode;
  char    db[TSDB_DB_FNAME_LEN];
  int16_t offset[TSDB_MAX_COLUMNS];
  int32_t bytes[TSDB_MAX_COLUMNS];
  char    payload[];
} SSysTableRetrieveObj;

L
Liu Jicong 已提交
466 467
typedef struct {
  char    key[TSDB_PARTITION_KEY_LEN];
L
Liu Jicong 已提交
468
  int64_t dbUid;
L
Liu Jicong 已提交
469 470 471
  int64_t offset;
} SMqOffsetObj;

S
Shengliang Guan 已提交
472 473
int32_t tEncodeSMqOffsetObj(void** buf, const SMqOffsetObj* pOffset);
void*   tDecodeSMqOffsetObj(void* buf, SMqOffsetObj* pOffset);
L
Liu Jicong 已提交
474

L
Liu Jicong 已提交
475
typedef struct {
L
Liu Jicong 已提交
476 477
  char           name[TSDB_TOPIC_FNAME_LEN];
  char           db[TSDB_DB_FNAME_LEN];
478
  char           createUser[TSDB_USER_LEN];
L
Liu Jicong 已提交
479 480 481 482 483
  int64_t        createTime;
  int64_t        updateTime;
  int64_t        uid;
  int64_t        dbUid;
  int32_t        version;
L
Liu Jicong 已提交
484 485
  int8_t         subType;   // column, db or stable
  int8_t         withMeta;  // TODO
L
Liu Jicong 已提交
486 487
  SRWLatch       lock;
  int32_t        sqlLen;
L
Liu Jicong 已提交
488
  int32_t        astLen;
L
Liu Jicong 已提交
489
  char*          sql;
L
Liu Jicong 已提交
490
  char*          ast;
L
Liu Jicong 已提交
491 492
  char*          physicalPlan;
  SSchemaWrapper schema;
L
Liu Jicong 已提交
493
  int64_t        stbUid;
L
Liu Jicong 已提交
494 495 496
  // forbid condition
  int64_t ntbUid;
  SArray* ntbColIds;
L
Liu Jicong 已提交
497
  int64_t ctbStbUid;
L
Liu Jicong 已提交
498 499
} SMqTopicObj;

L
Liu Jicong 已提交
500
typedef struct {
S
Shengliang Guan 已提交
501 502
  int64_t  consumerId;
  char     cgroup[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
503
  char     clientId[256];
S
Shengliang Guan 已提交
504 505 506 507 508
  int8_t   updateType;  // used only for update
  int32_t  epoch;
  int32_t  status;
  int32_t  hbStatus;          // hbStatus is not applicable to serialization
  SRWLatch lock;              // lock is used for topics update
L
Liu Jicong 已提交
509 510 511
  SArray*  currentTopics;     // SArray<char*>
  SArray*  rebNewTopics;      // SArray<char*>
  SArray*  rebRemovedTopics;  // SArray<char*>
L
Liu Jicong 已提交
512

L
Liu Jicong 已提交
513 514 515
  // subscribed by user
  SArray* assignedTopics;  // SArray<char*>

L
Liu Jicong 已提交
516 517 518 519 520 521
  // data for display
  int32_t pid;
  SEpSet  ep;
  int64_t upTime;
  int64_t subscribeTime;
  int64_t rebalanceTime;
L
Liu Jicong 已提交
522 523
} SMqConsumerObj;

524 525 526 527
SMqConsumerObj* tNewSMqConsumerObj(int64_t consumerId, char cgroup[TSDB_CGROUP_LEN]);
void            tDeleteSMqConsumerObj(SMqConsumerObj* pConsumer);
int32_t         tEncodeSMqConsumerObj(void** buf, const SMqConsumerObj* pConsumer);
void*           tDecodeSMqConsumerObj(const void* buf, SMqConsumerObj* pConsumer);
L
Liu Jicong 已提交
528

529 530 531
typedef struct {
  int32_t vgId;
  char*   qmsg;
L
Liu Jicong 已提交
532
  SEpSet  epSet;
533 534 535 536 537 538 539 540 541 542
} SMqVgEp;

SMqVgEp* tCloneSMqVgEp(const SMqVgEp* pVgEp);
void     tDeleteSMqVgEp(SMqVgEp* pVgEp);
int32_t  tEncodeSMqVgEp(void** buf, const SMqVgEp* pVgEp);
void*    tDecodeSMqVgEp(const void* buf, SMqVgEp* pVgEp);

typedef struct {
  int64_t consumerId;  // -1 for unassigned
  SArray* vgs;         // SArray<SMqVgEp*>
543
} SMqConsumerEp;
544

545
SMqConsumerEp* tCloneSMqConsumerEp(const SMqConsumerEp* pEp);
546
void           tDeleteSMqConsumerEp(void* pEp);
547 548
int32_t        tEncodeSMqConsumerEp(void** buf, const SMqConsumerEp* pEp);
void*          tDecodeSMqConsumerEp(const void* buf, SMqConsumerEp* pEp);
549 550

typedef struct {
L
Liu Jicong 已提交
551 552 553 554 555
  char      key[TSDB_SUBSCRIBE_KEY_LEN];
  SRWLatch  lock;
  int64_t   dbUid;
  int32_t   vgNum;
  int8_t    subType;
L
Liu Jicong 已提交
556
  int8_t    withMeta;
L
Liu Jicong 已提交
557
  int64_t   stbUid;
558 559
  SHashObj* consumerHash;   // consumerId -> SMqConsumerEp
  SArray*   unassignedVgs;  // SArray<SMqVgEp*>
L
Liu Jicong 已提交
560
  char      dbName[TSDB_DB_FNAME_LEN];
561 562 563 564 565 566 567 568 569 570
} SMqSubscribeObj;

SMqSubscribeObj* tNewSubscribeObj(const char key[TSDB_SUBSCRIBE_KEY_LEN]);
SMqSubscribeObj* tCloneSubscribeObj(const SMqSubscribeObj* pSub);
void             tDeleteSubscribeObj(SMqSubscribeObj* pSub);
int32_t          tEncodeSubscribeObj(void** buf, const SMqSubscribeObj* pSub);
void*            tDecodeSubscribeObj(const void* buf, SMqSubscribeObj* pSub);

typedef struct {
  int32_t epoch;
571
  SArray* consumers;  // SArray<SMqConsumerEp*>
572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589
} SMqSubActionLogEntry;

SMqSubActionLogEntry* tCloneSMqSubActionLogEntry(SMqSubActionLogEntry* pEntry);
void                  tDeleteSMqSubActionLogEntry(SMqSubActionLogEntry* pEntry);
int32_t               tEncodeSMqSubActionLogEntry(void** buf, const SMqSubActionLogEntry* pEntry);
void*                 tDecodeSMqSubActionLogEntry(const void* buf, SMqSubActionLogEntry* pEntry);

typedef struct {
  char    key[TSDB_SUBSCRIBE_KEY_LEN];
  SArray* logs;  // SArray<SMqSubActionLogEntry*>
} SMqSubActionLogObj;

SMqSubActionLogObj* tCloneSMqSubActionLogObj(SMqSubActionLogObj* pLog);
void                tDeleteSMqSubActionLogObj(SMqSubActionLogObj* pLog);
int32_t             tEncodeSMqSubActionLogObj(void** buf, const SMqSubActionLogObj* pLog);
void*               tDecodeSMqSubActionLogObj(const void* buf, SMqSubActionLogObj* pLog);

typedef struct {
L
Liu Jicong 已提交
590 591
  int32_t           oldConsumerNum;
  const SMqRebInfo* pRebInfo;
592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607
} SMqRebInputObj;

typedef struct {
  int64_t  oldConsumerId;
  int64_t  newConsumerId;
  SMqVgEp* pVgEp;
} SMqRebOutputVg;

typedef struct {
  SArray*               rebVgs;            // SArray<SMqRebOutputVg>
  SArray*               newConsumers;      // SArray<int64_t>
  SArray*               removedConsumers;  // SArray<int64_t>
  SArray*               touchedConsumers;  // SArray<int64_t>
  SMqSubscribeObj*      pSub;
  SMqSubActionLogEntry* pLogEntry;
} SMqRebOutputObj;
L
Liu Jicong 已提交
608

L
Liu Jicong 已提交
609
typedef struct {
L
Liu Jicong 已提交
610 611 612 613 614 615 616
  char name[TSDB_STREAM_FNAME_LEN];
  // ctl
  SRWLatch lock;
  // create info
  int64_t createTime;
  int64_t updateTime;
  int32_t version;
L
Liu Jicong 已提交
617
  int32_t totalLevel;
L
Liu Jicong 已提交
618 619 620 621 622
  int64_t smaId;  // 0 for unused
  // info
  int64_t uid;
  int8_t  status;
  // config
623
  int8_t  igExpired;
L
Liu Jicong 已提交
624
  int8_t  trigger;
L
Liu Jicong 已提交
625
  int8_t  fillHistory;
L
Liu Jicong 已提交
626 627 628 629 630 631 632 633 634 635
  int64_t triggerParam;
  int64_t watermark;
  // source and target
  int64_t sourceDbUid;
  int64_t targetDbUid;
  char    sourceDb[TSDB_DB_FNAME_LEN];
  char    targetDb[TSDB_DB_FNAME_LEN];
  char    targetSTbName[TSDB_TABLE_FNAME_LEN];
  int64_t targetStbUid;
  int32_t fixedSinkVgId;  // 0 for shuffle
L
Liu Jicong 已提交
636 637 638
  // fixedSinkVg is not applicable for encode and decode
  SVgObj fixedSinkVg;

L
Liu Jicong 已提交
639
  // transformation
L
Liu Jicong 已提交
640
  char*          sql;
L
Liu Jicong 已提交
641
  char*          ast;
L
Liu Jicong 已提交
642 643 644
  char*          physicalPlan;
  SArray*        tasks;  // SArray<SArray<SStreamTask>>
  SSchemaWrapper outputSchema;
645
  SSchemaWrapper tagSchema;
646 647 648 649

  // 3.0.20
  int64_t checkpointFreq;  // ms
  int64_t currentTick;     // do not serialize
650
  int64_t deleteMark;
L
Liu Jicong 已提交
651 652
} SStreamObj;

H
Hongze Cheng 已提交
653
int32_t tEncodeSStreamObj(SEncoder* pEncoder, const SStreamObj* pObj);
654
int32_t tDecodeSStreamObj(SDecoder* pDecoder, SStreamObj* pObj, int32_t sver);
L
Liu Jicong 已提交
655
void    tFreeStreamObj(SStreamObj* pObj);
L
Liu Jicong 已提交
656

L
Liu Jicong 已提交
657 658 659 660 661 662 663
typedef struct {
  char    streamName[TSDB_STREAM_FNAME_LEN];
  int64_t uid;
  int64_t streamUid;
  SArray* childInfo;  // SArray<SStreamChildEpInfo>
} SStreamCheckpointObj;

S
Shengliang Guan 已提交
664 665 666
#ifdef __cplusplus
}
#endif
S
Shengliang Guan 已提交
667

S
Shengliang Guan 已提交
668
#endif /*_TD_MND_DEF_H_*/