mndDef.h 16.4 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;
196
  int32_t    numOfOtherNodes;
S
Shengliang Guan 已提交
197
  int32_t    numOfSupportVnodes;
S
Shengliang Guan 已提交
198
  float      numOfCores;
199 200 201
  int64_t    memTotal;
  int64_t    memAvail;
  int64_t    memUsed;
S
Shengliang Guan 已提交
202 203 204 205
  EDndReason offlineReason;
  uint16_t   port;
  char       fqdn[TSDB_FQDN_LEN];
  char       ep[TSDB_EP_LEN];
S
Shengliang Guan 已提交
206 207
} SDnodeObj;

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

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

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

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

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

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

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

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

S
Shengliang Guan 已提交
319
typedef struct {
320 321 322 323 324 325 326 327 328 329
  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;
330
  int64_t  stateTs;
331
  int64_t  compactStartTime;
S
Shengliang Guan 已提交
332 333 334 335
} SDbObj;

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

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

S
Shengliang Guan 已提交
363
typedef struct {
C
Cary Xu 已提交
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 389
  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 已提交
390 391
} SSmaObj;

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

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

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

451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467
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 已提交
468 469
typedef struct {
  char    key[TSDB_PARTITION_KEY_LEN];
L
Liu Jicong 已提交
470
  int64_t dbUid;
L
Liu Jicong 已提交
471 472 473
  int64_t offset;
} SMqOffsetObj;

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

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

L
Liu Jicong 已提交
502
typedef struct {
S
Shengliang Guan 已提交
503 504
  int64_t  consumerId;
  char     cgroup[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
505
  char     clientId[256];
S
Shengliang Guan 已提交
506 507 508 509 510
  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 已提交
511 512 513
  SArray*  currentTopics;     // SArray<char*>
  SArray*  rebNewTopics;      // SArray<char*>
  SArray*  rebRemovedTopics;  // SArray<char*>
L
Liu Jicong 已提交
514

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

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

526 527 528 529
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 已提交
530

531 532 533
typedef struct {
  int32_t vgId;
  char*   qmsg;
L
Liu Jicong 已提交
534
  SEpSet  epSet;
535 536 537 538 539 540 541 542 543 544
} 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*>
545
} SMqConsumerEp;
546

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

typedef struct {
L
Liu Jicong 已提交
553 554 555 556 557
  char      key[TSDB_SUBSCRIBE_KEY_LEN];
  SRWLatch  lock;
  int64_t   dbUid;
  int32_t   vgNum;
  int8_t    subType;
L
Liu Jicong 已提交
558
  int8_t    withMeta;
L
Liu Jicong 已提交
559
  int64_t   stbUid;
560 561
  SHashObj* consumerHash;   // consumerId -> SMqConsumerEp
  SArray*   unassignedVgs;  // SArray<SMqVgEp*>
L
Liu Jicong 已提交
562
  char      dbName[TSDB_DB_FNAME_LEN];
563 564 565 566 567 568 569 570 571 572
} 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;
573
  SArray* consumers;  // SArray<SMqConsumerEp*>
574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591
} 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 已提交
592 593
  int32_t           oldConsumerNum;
  const SMqRebInfo* pRebInfo;
594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609
} 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 已提交
610

L
Liu Jicong 已提交
611
typedef struct {
L
Liu Jicong 已提交
612 613 614 615 616 617 618
  char name[TSDB_STREAM_FNAME_LEN];
  // ctl
  SRWLatch lock;
  // create info
  int64_t createTime;
  int64_t updateTime;
  int32_t version;
L
Liu Jicong 已提交
619
  int32_t totalLevel;
L
Liu Jicong 已提交
620 621 622 623 624
  int64_t smaId;  // 0 for unused
  // info
  int64_t uid;
  int8_t  status;
  // config
625
  int8_t  igExpired;
L
Liu Jicong 已提交
626
  int8_t  trigger;
L
Liu Jicong 已提交
627
  int8_t  fillHistory;
L
Liu Jicong 已提交
628 629 630 631 632 633 634 635 636 637
  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 已提交
638 639 640
  // fixedSinkVg is not applicable for encode and decode
  SVgObj fixedSinkVg;

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

  // 3.0.20
  int64_t checkpointFreq;  // ms
  int64_t currentTick;     // do not serialize
652
  int64_t deleteMark;
653
  int8_t  igCheckUpdate;
L
Liu Jicong 已提交
654 655
} SStreamObj;

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

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

S
Shengliang Guan 已提交
667 668 669
#ifdef __cplusplus
}
#endif
S
Shengliang Guan 已提交
670

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