mndDef.h 14.9 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

S
Shengliang Guan 已提交
38
typedef enum {
39 40 41 42 43 44 45 46
  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 已提交
47 48

typedef enum {
49 50 51 52 53 54
  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 已提交
55

S
Shengliang Guan 已提交
56
typedef enum {
57
  TRN_STAGE_PREPARE = 0,
S
Shengliang Guan 已提交
58 59
  TRN_STAGE_REDO_LOG = 1,
  TRN_STAGE_REDO_ACTION = 2,
60 61 62 63 64
  TRN_STAGE_ROLLBACK = 3,
  TRN_STAGE_UNDO_ACTION = 4,
  TRN_STAGE_UNDO_LOG = 5,
  TRN_STAGE_COMMIT = 6,
  TRN_STAGE_COMMIT_LOG = 7,
S
Shengliang Guan 已提交
65
  TRN_STAGE_FINISHED = 8
S
Shengliang Guan 已提交
66 67
} ETrnStage;

S
Shengliang Guan 已提交
68
typedef enum {
S
Shengliang Guan 已提交
69 70 71 72
  TRN_TYPE_BASIC_SCOPE = 1000,
  TRN_TYPE_CREATE_USER = 1001,
  TRN_TYPE_ALTER_USER = 1002,
  TRN_TYPE_DROP_USER = 1003,
S
Shengliang Guan 已提交
73 74
  TRN_TYPE_CREATE_FUNC = 1004,
  TRN_TYPE_DROP_FUNC = 1005,
L
Liu Jicong 已提交
75

S
Shengliang Guan 已提交
76 77 78 79 80 81 82 83 84 85 86 87
  TRN_TYPE_CREATE_SNODE = 1006,
  TRN_TYPE_DROP_SNODE = 1007,
  TRN_TYPE_CREATE_QNODE = 1008,
  TRN_TYPE_DROP_QNODE = 1009,
  TRN_TYPE_CREATE_BNODE = 1010,
  TRN_TYPE_DROP_BNODE = 1011,
  TRN_TYPE_CREATE_MNODE = 1012,
  TRN_TYPE_DROP_MNODE = 1013,
  TRN_TYPE_CREATE_TOPIC = 1014,
  TRN_TYPE_DROP_TOPIC = 1015,
  TRN_TYPE_SUBSCRIBE = 1016,
  TRN_TYPE_REBALANCE = 1017,
L
Liu Jicong 已提交
88
  TRN_TYPE_COMMIT_OFFSET = 1018,
L
Liu Jicong 已提交
89
  TRN_TYPE_CREATE_STREAM = 1019,
L
Liu Jicong 已提交
90 91
  TRN_TYPE_DROP_STREAM = 1020,
  TRN_TYPE_ALTER_STREAM = 1021,
92
  TRN_TYPE_CONSUMER_LOST = 1022,
L
Liu Jicong 已提交
93
  TRN_TYPE_CONSUMER_RECOVER = 1023,
S
Shengliang Guan 已提交
94
  TRN_TYPE_BASIC_SCOPE_END,
95

S
Shengliang Guan 已提交
96 97 98 99
  TRN_TYPE_GLOBAL_SCOPE = 2000,
  TRN_TYPE_CREATE_DNODE = 2001,
  TRN_TYPE_DROP_DNODE = 2002,
  TRN_TYPE_GLOBAL_SCOPE_END,
100

S
Shengliang Guan 已提交
101 102 103 104
  TRN_TYPE_DB_SCOPE = 3000,
  TRN_TYPE_CREATE_DB = 3001,
  TRN_TYPE_ALTER_DB = 3002,
  TRN_TYPE_DROP_DB = 3003,
S
Shengliang Guan 已提交
105 106
  TRN_TYPE_SPLIT_VGROUP = 3004,
  TRN_TYPE_MERGE_VGROUP = 3015,
S
Shengliang Guan 已提交
107
  TRN_TYPE_DB_SCOPE_END,
108

S
Shengliang Guan 已提交
109 110 111 112
  TRN_TYPE_STB_SCOPE = 4000,
  TRN_TYPE_CREATE_STB = 4001,
  TRN_TYPE_ALTER_STB = 4002,
  TRN_TYPE_DROP_STB = 4003,
S
Shengliang Guan 已提交
113 114
  TRN_TYPE_CREATE_SMA = 4004,
  TRN_TYPE_DROP_SMA = 4005,
S
Shengliang Guan 已提交
115
  TRN_TYPE_STB_SCOPE_END,
S
Shengliang Guan 已提交
116 117
} ETrnType;

S
Shengliang Guan 已提交
118 119 120 121
typedef enum {
  TRN_POLICY_ROLLBACK = 0,
  TRN_POLICY_RETRY = 1,
} ETrnPolicy;
S
Shengliang Guan 已提交
122

S
Shengliang Guan 已提交
123 124 125 126 127 128 129 130 131 132 133 134 135 136
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 已提交
137 138 139 140 141 142 143 144 145
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 已提交
146
typedef struct {
S
Shengliang Guan 已提交
147 148 149
  int32_t    id;
  ETrnStage  stage;
  ETrnPolicy policy;
150
  ETrnType   type;
S
Shengliang Guan 已提交
151 152
  int32_t    code;
  int32_t    failedTimes;
L
Liu Jicong 已提交
153 154
  void*      rpcHandle;
  void*      rpcAHandle;
dengyihao's avatar
dengyihao 已提交
155
  int64_t    rpcRefId;
S
Shengliang Guan 已提交
156 157
  void*      rpcRsp;
  int32_t    rpcRspLen;
L
Liu Jicong 已提交
158 159 160 161 162
  SArray*    redoLogs;
  SArray*    undoLogs;
  SArray*    commitLogs;
  SArray*    redoActions;
  SArray*    undoActions;
S
Shengliang Guan 已提交
163 164
  int64_t    createdTime;
  int64_t    lastExecTime;
L
Liu Jicong 已提交
165
  int64_t    dbUid;
S
Shengliang Guan 已提交
166
  char       dbname[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
167
  char       lastError[TSDB_TRANS_ERROR_LEN];
168 169 170 171
  int32_t    startFunc;
  int32_t    stopFunc;
  int32_t    paramLen;
  void*      param;
S
Shengliang Guan 已提交
172
} STrans;
S
Shengliang Guan 已提交
173

S
Shengliang Guan 已提交
174
typedef struct {
175
  int64_t id;
S
Shengliang Guan 已提交
176
  char    name[TSDB_CLUSTER_ID_LEN];
S
Shengliang Guan 已提交
177 178 179 180
  int64_t createdTime;
  int64_t updateTime;
} SClusterObj;

S
Shengliang Guan 已提交
181
typedef struct {
S
Shengliang Guan 已提交
182 183 184 185
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
  int64_t    rebootTime;
186
  int64_t    lastAccessTime;
S
Shengliang Guan 已提交
187
  int32_t    accessTimes;
S
Shengliang Guan 已提交
188
  int32_t    numOfVnodes;
S
Shengliang Guan 已提交
189 190
  int32_t    numOfSupportVnodes;
  int32_t    numOfCores;
S
Shengliang Guan 已提交
191 192 193 194
  EDndReason offlineReason;
  uint16_t   port;
  char       fqdn[TSDB_FQDN_LEN];
  char       ep[TSDB_EP_LEN];
S
Shengliang Guan 已提交
195 196
} SDnodeObj;

S
Shengliang Guan 已提交
197
typedef struct {
S
Shengliang Guan 已提交
198
  int32_t    id;
S
Shengliang Guan 已提交
199 200
  int64_t    createdTime;
  int64_t    updateTime;
S
Shengliang Guan 已提交
201
  ESyncState role;
S
Shengliang Guan 已提交
202 203
  int32_t    roleTerm;
  int64_t    roleTime;
L
Liu Jicong 已提交
204
  SDnodeObj* pDnode;
S
Shengliang Guan 已提交
205 206
} SMnodeObj;

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

typedef struct {
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
L
Liu Jicong 已提交
218
  SDnodeObj* pDnode;
S
Shengliang Guan 已提交
219 220 221 222 223 224
} SSnodeObj;

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

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;
S
Shengliang Guan 已提交
269
  int32_t   acctId;
D
dapan 已提交
270
  int32_t   authVersion;
271 272
  SHashObj* readDbs;
  SHashObj* writeDbs;
S
Shengliang Guan 已提交
273
  SRWLatch  lock;
S
Shengliang Guan 已提交
274 275 276
} SUserObj;

typedef struct {
277
  int32_t numOfVgroups;
S
Shengliang Guan 已提交
278 279 280 281
  int32_t numOfStables;
  int32_t buffer;
  int32_t pageSize;
  int32_t pages;
S
Shengliang Guan 已提交
282 283 284 285
  int32_t daysPerFile;
  int32_t daysToKeep0;
  int32_t daysToKeep1;
  int32_t daysToKeep2;
S
Shengliang Guan 已提交
286 287
  int32_t minRows;
  int32_t maxRows;
S
Shengliang Guan 已提交
288
  int32_t fsyncPeriod;
S
Shengliang Guan 已提交
289
  int8_t  walLevel;
S
Shengliang Guan 已提交
290 291 292
  int8_t  precision;
  int8_t  compression;
  int8_t  replications;
S
Shengliang Guan 已提交
293
  int8_t  strict;
S
Shengliang Guan 已提交
294
  int8_t  cacheLastRow;
S
Shengliang Guan 已提交
295
  int8_t  hashMethod;  // default is 1
S
sma  
Shengliang Guan 已提交
296 297
  int32_t numOfRetensions;
  SArray* pRetensions;
S
Shengliang Guan 已提交
298 299
} SDbCfg;

S
Shengliang Guan 已提交
300
typedef struct {
301 302 303 304 305 306 307 308 309 310
  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 已提交
311 312 313 314
} SDbObj;

typedef struct {
  int32_t    dnodeId;
S
Shengliang Guan 已提交
315
  ESyncState role;
S
Shengliang Guan 已提交
316 317
} SVnodeGid;

S
Shengliang Guan 已提交
318
typedef struct {
S
Shengliang Guan 已提交
319
  int32_t   vgId;
S
Shengliang Guan 已提交
320 321
  int64_t   createdTime;
  int64_t   updateTime;
S
Shengliang Guan 已提交
322
  int32_t   version;
S
Shengliang Guan 已提交
323 324
  uint32_t  hashBegin;
  uint32_t  hashEnd;
325
  char      dbName[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
326
  int64_t   dbUid;
S
Shengliang Guan 已提交
327 328
  int64_t   numOfTables;
  int64_t   numOfTimeSeries;
S
Shengliang Guan 已提交
329 330 331
  int64_t   totalStorage;
  int64_t   compStorage;
  int64_t   pointsWritten;
S
Shengliang Guan 已提交
332 333 334
  int8_t    compact;
  int8_t    replica;
  SVnodeGid vnodeGid[TSDB_MAX_REPLICA];
S
Shengliang Guan 已提交
335 336
} SVgObj;

S
Shengliang Guan 已提交
337
typedef struct {
S
Shengliang Guan 已提交
338
  char    name[TSDB_TABLE_FNAME_LEN];
S
sma  
Shengliang Guan 已提交
339 340
  char    stb[TSDB_TABLE_FNAME_LEN];
  char    db[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
341 342 343
  int64_t createdTime;
  int64_t uid;
  int64_t stbUid;
S
sma  
Shengliang Guan 已提交
344
  int64_t dbUid;
S
Shengliang Guan 已提交
345 346 347
  int8_t  intervalUnit;
  int8_t  slidingUnit;
  int8_t  timezone;
S
sma  
Shengliang Guan 已提交
348
  int32_t dstVgId;  // for stream
S
Shengliang Guan 已提交
349 350 351
  int64_t interval;
  int64_t offset;
  int64_t sliding;
S
sma  
Shengliang Guan 已提交
352
  int32_t exprLen;  // strlen + 1
S
Shengliang Guan 已提交
353
  int32_t tagsFilterLen;
S
sma  
Shengliang Guan 已提交
354 355
  int32_t sqlLen;
  int32_t astLen;
S
Shengliang Guan 已提交
356 357
  char*   expr;
  char*   tagsFilter;
S
sma  
Shengliang Guan 已提交
358 359
  char*   sql;
  char*   ast;
S
Shengliang Guan 已提交
360 361
} SSmaObj;

S
Shengliang Guan 已提交
362
typedef struct {
S
Shengliang Guan 已提交
363
  char     name[TSDB_TABLE_FNAME_LEN];
364
  char     db[TSDB_DB_FNAME_LEN];
365 366
  int64_t  createdTime;
  int64_t  updateTime;
L
Liu Jicong 已提交
367 368
  int64_t  uid;
  int64_t  dbUid;
S
Shengliang Guan 已提交
369
  int32_t  version;
S
Shengliang Guan 已提交
370
  int32_t  nextColId;
S
sma  
Shengliang Guan 已提交
371 372
  float    xFilesFactor;
  int32_t  delay;
S
sma  
Shengliang Guan 已提交
373
  int32_t  ttl;
S
Shengliang Guan 已提交
374 375
  int32_t  numOfColumns;
  int32_t  numOfTags;
S
sma  
Shengliang Guan 已提交
376
  int32_t  commentLen;
377 378
  int32_t  ast1Len;
  int32_t  ast2Len;
S
Shengliang Guan 已提交
379
  SSchema* pColumns;
S
Shengliang Guan 已提交
380
  SSchema* pTags;
S
sma  
Shengliang Guan 已提交
381
  char*    comment;
382 383
  char*    pAst1;
  char*    pAst2;
S
Shengliang Guan 已提交
384
  SRWLatch lock;
S
Shengliang Guan 已提交
385
} SStbObj;
S
Shengliang Guan 已提交
386

S
Shengliang Guan 已提交
387
typedef struct {
S
Shengliang Guan 已提交
388 389
  char    name[TSDB_FUNC_NAME_LEN];
  int64_t createdTime;
S
Shengliang Guan 已提交
390 391 392 393 394 395
  int8_t  funcType;
  int8_t  scriptType;
  int8_t  align;
  int8_t  outputType;
  int32_t outputLen;
  int32_t bufSize;
S
Shengliang 已提交
396
  int64_t signature;
S
Shengliang Guan 已提交
397 398
  int32_t commentSize;
  int32_t codeSize;
L
Liu Jicong 已提交
399 400
  char*   pComment;
  char*   pCode;
S
Shengliang Guan 已提交
401 402
} SFuncObj;

S
Shengliang Guan 已提交
403
typedef struct {
L
Liu Jicong 已提交
404 405 406 407 408 409 410
  int64_t        id;
  int8_t         type;
  int8_t         replica;
  int16_t        numOfColumns;
  int32_t        numOfRows;
  void*          pIter;
  SMnode*        pMnode;
411
  STableMetaRsp* pMeta;
L
Liu Jicong 已提交
412 413
  bool           sysDbRsp;
  char           db[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
414 415
} SShowObj;

416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432
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 已提交
433 434
typedef struct {
  char    key[TSDB_PARTITION_KEY_LEN];
L
Liu Jicong 已提交
435
  int64_t dbUid;
L
Liu Jicong 已提交
436 437 438
  int64_t offset;
} SMqOffsetObj;

S
Shengliang Guan 已提交
439 440
int32_t tEncodeSMqOffsetObj(void** buf, const SMqOffsetObj* pOffset);
void*   tDecodeSMqOffsetObj(void* buf, SMqOffsetObj* pOffset);
L
Liu Jicong 已提交
441

L
Liu Jicong 已提交
442
typedef struct {
L
Liu Jicong 已提交
443 444 445 446 447
  char           name[TSDB_TOPIC_FNAME_LEN];
  char           db[TSDB_DB_FNAME_LEN];
  int64_t        createTime;
  int64_t        updateTime;
  int64_t        uid;
L
Liu Jicong 已提交
448 449
  int64_t        dbUid;
  int32_t        version;
L
Liu Jicong 已提交
450 451 452 453
  int8_t         subType;  // db or table
  int8_t         withTbName;
  int8_t         withSchema;
  int8_t         withTag;
L
Liu Jicong 已提交
454 455
  SRWLatch       lock;
  int32_t        sqlLen;
L
Liu Jicong 已提交
456
  int32_t        astLen;
L
Liu Jicong 已提交
457
  char*          sql;
L
Liu Jicong 已提交
458
  char*          ast;
L
Liu Jicong 已提交
459 460
  char*          physicalPlan;
  SSchemaWrapper schema;
L
Liu Jicong 已提交
461 462
} SMqTopicObj;

L
Liu Jicong 已提交
463
typedef struct {
S
Shengliang Guan 已提交
464 465 466 467 468 469 470 471
  int64_t  consumerId;
  char     cgroup[TSDB_CGROUP_LEN];
  char     appId[TSDB_CGROUP_LEN];
  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 已提交
472 473 474
  SArray*  currentTopics;     // SArray<char*>
  SArray*  rebNewTopics;      // SArray<char*>
  SArray*  rebRemovedTopics;  // SArray<char*>
L
Liu Jicong 已提交
475

L
Liu Jicong 已提交
476 477 478
  // subscribed by user
  SArray* assignedTopics;  // SArray<char*>

L
Liu Jicong 已提交
479 480 481 482 483 484
  // data for display
  int32_t pid;
  SEpSet  ep;
  int64_t upTime;
  int64_t subscribeTime;
  int64_t rebalanceTime;
L
Liu Jicong 已提交
485 486
} SMqConsumerObj;

487 488 489 490
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 已提交
491

492 493 494
typedef struct {
  int32_t vgId;
  char*   qmsg;
L
Liu Jicong 已提交
495
  SEpSet  epSet;
496 497 498 499 500 501 502 503 504 505
} 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*>
506
} SMqConsumerEp;
507

508 509 510 511
SMqConsumerEp* tCloneSMqConsumerEp(const SMqConsumerEp* pEp);
void           tDeleteSMqConsumerEp(SMqConsumerEp* pEp);
int32_t        tEncodeSMqConsumerEp(void** buf, const SMqConsumerEp* pEp);
void*          tDecodeSMqConsumerEp(const void* buf, SMqConsumerEp* pEp);
512 513 514 515

typedef struct {
  char      key[TSDB_SUBSCRIBE_KEY_LEN];
  SRWLatch  lock;
L
Liu Jicong 已提交
516
  int64_t   dbUid;
517
  int32_t   vgNum;
L
Liu Jicong 已提交
518 519 520 521
  int8_t    subType;
  int8_t    withTbName;
  int8_t    withSchema;
  int8_t    withTag;
522 523
  SHashObj* consumerHash;   // consumerId -> SMqConsumerEp
  SArray*   unassignedVgs;  // SArray<SMqVgEp*>
524 525 526 527 528 529 530 531 532 533
} 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;
534
  SArray* consumers;  // SArray<SMqConsumerEp*>
535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552
} 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 已提交
553 554
  int32_t           oldConsumerNum;
  const SMqRebInfo* pRebInfo;
555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570
} 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 已提交
571

L
Liu Jicong 已提交
572
typedef struct {
S
Shengliang Guan 已提交
573 574 575 576 577 578 579 580 581 582 583 584
  char           name[TSDB_TOPIC_FNAME_LEN];
  char           sourceDb[TSDB_DB_FNAME_LEN];
  char           targetDb[TSDB_DB_FNAME_LEN];
  char           targetSTbName[TSDB_TABLE_FNAME_LEN];
  int64_t        createTime;
  int64_t        updateTime;
  int64_t        uid;
  int64_t        dbUid;
  int32_t        version;
  int32_t        vgNum;
  SRWLatch       lock;
  int8_t         status;
L
Liu Jicong 已提交
585 586 587
  int8_t         createdBy;      // STREAM_CREATED_BY__USER or SMA
  int32_t        fixedSinkVgId;  // 0 for shuffle
  int64_t        smaId;          // 0 for unused
L
fix  
Liu Jicong 已提交
588 589 590
  int8_t         trigger;
  int32_t        triggerParam;
  int64_t        waterMark;
L
Liu Jicong 已提交
591 592 593 594
  char*          sql;
  char*          physicalPlan;
  SArray*        tasks;  // SArray<SArray<SStreamTask>>
  SSchemaWrapper outputSchema;
L
Liu Jicong 已提交
595 596
} SStreamObj;

H
Hongze Cheng 已提交
597 598
int32_t tEncodeSStreamObj(SEncoder* pEncoder, const SStreamObj* pObj);
int32_t tDecodeSStreamObj(SDecoder* pDecoder, SStreamObj* pObj);
L
Liu Jicong 已提交
599

S
Shengliang Guan 已提交
600 601 602
#ifdef __cplusplus
}
#endif
S
Shengliang Guan 已提交
603

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