mndDef.h 16.0 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 74 75
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,
} EOperType;

S
Shengliang Guan 已提交
76
typedef enum {
77 78 79 80 81 82 83 84
  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 已提交
85 86

typedef enum {
87 88 89 90 91 92
  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 已提交
93

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

S
Shengliang Guan 已提交
101
typedef enum {
102
  TRN_STAGE_PREPARE = 0,
103 104 105 106 107 108
  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 已提交
109 110
} ETrnStage;

S
Shengliang Guan 已提交
111 112 113 114
typedef enum {
  TRN_POLICY_ROLLBACK = 0,
  TRN_POLICY_RETRY = 1,
} ETrnPolicy;
S
Shengliang Guan 已提交
115

116
typedef enum {
117 118 119
  TRN_EXEC_PRARLLEL = 0,
  TRN_EXEC_SERIAL = 1,
} ETrnExec;
120

S
Shengliang Guan 已提交
121 122 123 124 125 126 127 128 129 130 131 132 133 134
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 已提交
135 136 137 138 139 140 141 142 143
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 已提交
144
typedef struct {
L
Liu Jicong 已提交
145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163
  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;
164 165
  tmsg_t      lastMsgType;
  tmsg_t      originRpcType;
166 167
  char        dbname[TSDB_TABLE_FNAME_LEN];
  char        stbname[TSDB_TABLE_FNAME_LEN];
L
Liu Jicong 已提交
168 169 170 171
  int32_t     startFunc;
  int32_t     stopFunc;
  int32_t     paramLen;
  void*       param;
172
  char        opername[TSDB_TRANS_OPER_LEN];
L
Liu Jicong 已提交
173
  SArray*     pRpcArray;
S
Shengliang Guan 已提交
174
} STrans;
S
Shengliang Guan 已提交
175

S
Shengliang Guan 已提交
176
typedef struct {
177
  int64_t id;
S
Shengliang Guan 已提交
178
  char    name[TSDB_CLUSTER_ID_LEN];
S
Shengliang Guan 已提交
179 180
  int64_t createdTime;
  int64_t updateTime;
181
  int32_t upTime;
S
Shengliang Guan 已提交
182 183
} SClusterObj;

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

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

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

typedef struct {
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
L
Liu Jicong 已提交
225
  SDnodeObj* pDnode;
S
Shengliang Guan 已提交
226 227
} SSnodeObj;

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

typedef struct {
  int32_t numOfUsers;
  int32_t numOfDbs;
  int32_t numOfTimeSeries;
  int32_t numOfStreams;
S
Shengliang Guan 已提交
248 249
  int64_t totalStorage;  // Total storage wrtten from this account
  int64_t compStorage;   // Compressed storage on disk
S
Shengliang Guan 已提交
250 251
} SAcctInfo;

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

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

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

S
Shengliang Guan 已提交
313
typedef struct {
314 315 316 317 318 319 320 321 322 323
  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;
S
Shengliang Guan 已提交
324 325 326 327
} SDbObj;

typedef struct {
  int32_t    dnodeId;
328 329
  ESyncState syncState;
  bool       syncRestore;
S
Shengliang Guan 已提交
330 331
} SVnodeGid;

S
Shengliang Guan 已提交
332
typedef struct {
S
Shengliang Guan 已提交
333
  int32_t   vgId;
S
Shengliang Guan 已提交
334 335
  int64_t   createdTime;
  int64_t   updateTime;
S
Shengliang Guan 已提交
336
  int32_t   version;
S
Shengliang Guan 已提交
337 338
  uint32_t  hashBegin;
  uint32_t  hashEnd;
339
  char      dbName[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
340
  int64_t   dbUid;
341
  int64_t   cacheUsage;
S
Shengliang Guan 已提交
342 343
  int64_t   numOfTables;
  int64_t   numOfTimeSeries;
S
Shengliang Guan 已提交
344 345 346
  int64_t   totalStorage;
  int64_t   compStorage;
  int64_t   pointsWritten;
S
Shengliang Guan 已提交
347
  int8_t    compact;
S
Shengliang Guan 已提交
348
  int8_t    isTsma;
S
Shengliang Guan 已提交
349 350
  int8_t    replica;
  SVnodeGid vnodeGid[TSDB_MAX_REPLICA];
C
Cary Xu 已提交
351
  void*     pTsma;
S
Shengliang Guan 已提交
352 353
} SVgObj;

S
Shengliang Guan 已提交
354
typedef struct {
C
Cary Xu 已提交
355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380
  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 已提交
381 382
} SSmaObj;

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

S
Shengliang Guan 已提交
412
typedef struct {
S
Shengliang Guan 已提交
413 414
  char    name[TSDB_FUNC_NAME_LEN];
  int64_t createdTime;
S
Shengliang Guan 已提交
415 416 417 418 419 420
  int8_t  funcType;
  int8_t  scriptType;
  int8_t  align;
  int8_t  outputType;
  int32_t outputLen;
  int32_t bufSize;
S
Shengliang 已提交
421
  int64_t signature;
S
Shengliang Guan 已提交
422 423
  int32_t commentSize;
  int32_t codeSize;
L
Liu Jicong 已提交
424 425
  char*   pComment;
  char*   pCode;
S
Shengliang Guan 已提交
426 427
} SFuncObj;

S
Shengliang Guan 已提交
428
typedef struct {
L
Liu Jicong 已提交
429 430 431 432 433 434 435
  int64_t        id;
  int8_t         type;
  int8_t         replica;
  int16_t        numOfColumns;
  int32_t        numOfRows;
  void*          pIter;
  SMnode*        pMnode;
436
  STableMetaRsp* pMeta;
L
Liu Jicong 已提交
437 438
  bool           sysDbRsp;
  char           db[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
439 440
} SShowObj;

441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457
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 已提交
458 459
typedef struct {
  char    key[TSDB_PARTITION_KEY_LEN];
L
Liu Jicong 已提交
460
  int64_t dbUid;
L
Liu Jicong 已提交
461 462 463
  int64_t offset;
} SMqOffsetObj;

S
Shengliang Guan 已提交
464 465
int32_t tEncodeSMqOffsetObj(void** buf, const SMqOffsetObj* pOffset);
void*   tDecodeSMqOffsetObj(void* buf, SMqOffsetObj* pOffset);
L
Liu Jicong 已提交
466

L
Liu Jicong 已提交
467
typedef struct {
L
Liu Jicong 已提交
468 469 470 471 472 473 474
  char           name[TSDB_TOPIC_FNAME_LEN];
  char           db[TSDB_DB_FNAME_LEN];
  int64_t        createTime;
  int64_t        updateTime;
  int64_t        uid;
  int64_t        dbUid;
  int32_t        version;
L
Liu Jicong 已提交
475 476
  int8_t         subType;   // column, db or stable
  int8_t         withMeta;  // TODO
L
Liu Jicong 已提交
477 478
  SRWLatch       lock;
  int32_t        sqlLen;
L
Liu Jicong 已提交
479
  int32_t        astLen;
L
Liu Jicong 已提交
480
  char*          sql;
L
Liu Jicong 已提交
481
  char*          ast;
L
Liu Jicong 已提交
482 483
  char*          physicalPlan;
  SSchemaWrapper schema;
L
Liu Jicong 已提交
484
  int64_t        stbUid;
L
Liu Jicong 已提交
485 486 487
  // forbid condition
  int64_t ntbUid;
  SArray* ntbColIds;
L
Liu Jicong 已提交
488
  int64_t ctbStbUid;
L
Liu Jicong 已提交
489 490
} SMqTopicObj;

L
Liu Jicong 已提交
491
typedef struct {
S
Shengliang Guan 已提交
492 493
  int64_t  consumerId;
  char     cgroup[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
494
  char     clientId[256];
S
Shengliang Guan 已提交
495 496 497 498 499
  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 已提交
500 501 502
  SArray*  currentTopics;     // SArray<char*>
  SArray*  rebNewTopics;      // SArray<char*>
  SArray*  rebRemovedTopics;  // SArray<char*>
L
Liu Jicong 已提交
503

L
Liu Jicong 已提交
504 505 506
  // subscribed by user
  SArray* assignedTopics;  // SArray<char*>

L
Liu Jicong 已提交
507 508 509 510 511 512
  // data for display
  int32_t pid;
  SEpSet  ep;
  int64_t upTime;
  int64_t subscribeTime;
  int64_t rebalanceTime;
L
Liu Jicong 已提交
513 514
} SMqConsumerObj;

515 516 517 518
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 已提交
519

520 521 522
typedef struct {
  int32_t vgId;
  char*   qmsg;
L
Liu Jicong 已提交
523
  SEpSet  epSet;
524 525 526 527 528 529 530 531 532 533
} 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*>
534
} SMqConsumerEp;
535

536
SMqConsumerEp* tCloneSMqConsumerEp(const SMqConsumerEp* pEp);
537
void           tDeleteSMqConsumerEp(void* pEp);
538 539
int32_t        tEncodeSMqConsumerEp(void** buf, const SMqConsumerEp* pEp);
void*          tDecodeSMqConsumerEp(const void* buf, SMqConsumerEp* pEp);
540 541

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

L
Liu Jicong 已提交
600
typedef struct {
L
Liu Jicong 已提交
601 602 603 604 605 606 607
  char name[TSDB_STREAM_FNAME_LEN];
  // ctl
  SRWLatch lock;
  // create info
  int64_t createTime;
  int64_t updateTime;
  int32_t version;
L
Liu Jicong 已提交
608
  int32_t totalLevel;
L
Liu Jicong 已提交
609 610 611 612 613
  int64_t smaId;  // 0 for unused
  // info
  int64_t uid;
  int8_t  status;
  // config
614
  int8_t  igExpired;
L
Liu Jicong 已提交
615 616 617 618 619 620 621 622 623 624 625
  int8_t  trigger;
  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 已提交
626 627 628
  // fixedSinkVg is not applicable for encode and decode
  SVgObj fixedSinkVg;

L
Liu Jicong 已提交
629
  // transformation
L
Liu Jicong 已提交
630
  char*          sql;
L
Liu Jicong 已提交
631
  char*          ast;
L
Liu Jicong 已提交
632 633 634
  char*          physicalPlan;
  SArray*        tasks;  // SArray<SArray<SStreamTask>>
  SSchemaWrapper outputSchema;
635
  SSchemaWrapper tagSchema;
L
Liu Jicong 已提交
636 637
} SStreamObj;

H
Hongze Cheng 已提交
638 639
int32_t tEncodeSStreamObj(SEncoder* pEncoder, const SStreamObj* pObj);
int32_t tDecodeSStreamObj(SDecoder* pDecoder, SStreamObj* pObj);
L
Liu Jicong 已提交
640
void    tFreeStreamObj(SStreamObj* pObj);
L
Liu Jicong 已提交
641

L
Liu Jicong 已提交
642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657
typedef struct {
  char    streamName[TSDB_STREAM_FNAME_LEN];
  int64_t uid;
  int64_t streamUid;
  SArray* childInfo;  // SArray<SStreamChildEpInfo>
} SStreamCheckpointObj;

#if 0
typedef struct {
  int64_t uid;
  int64_t streamId;
  int8_t  status;
  int8_t  stage;
} SStreamRecoverObj;
#endif

S
Shengliang Guan 已提交
658 659 660
#ifdef __cplusplus
}
#endif
S
Shengliang Guan 已提交
661

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