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

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 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169
  int32_t        id;
  ETrnStage      stage;
  ETrnPolicy     policy;
  ETrnType       type;
  int32_t        code;
  int32_t        failedTimes;
  SRpcHandleInfo rpcInfo;
  void*          rpcRsp;
  int32_t        rpcRspLen;
  SArray*        redoLogs;
  SArray*        undoLogs;
  SArray*        commitLogs;
  SArray*        redoActions;
  SArray*        undoActions;
  int64_t        createdTime;
  int64_t        lastExecTime;
  int64_t        dbUid;
  char           dbname[TSDB_DB_FNAME_LEN];
  char           lastError[TSDB_TRANS_ERROR_LEN];
  int32_t        startFunc;
  int32_t        stopFunc;
  int32_t        paramLen;
  void*          param;
S
Shengliang Guan 已提交
170
} STrans;
S
Shengliang Guan 已提交
171

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

L
Liu Jicong 已提交
463
typedef struct {
S
Shengliang Guan 已提交
464 465
  int64_t  consumerId;
  char     cgroup[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
466
  char     clientId[256];
S
Shengliang Guan 已提交
467 468 469 470 471
  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
  char           name[TSDB_TOPIC_FNAME_LEN];
  char           sourceDb[TSDB_DB_FNAME_LEN];
  char           targetDb[TSDB_DB_FNAME_LEN];
  char           targetSTbName[TSDB_TABLE_FNAME_LEN];
L
Liu Jicong 已提交
577
  int64_t        targetStbUid;
S
Shengliang Guan 已提交
578 579 580 581 582 583 584 585
  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 已提交
586 587 588
  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 已提交
589 590 591
  int8_t         trigger;
  int32_t        triggerParam;
  int64_t        waterMark;
L
Liu Jicong 已提交
592 593 594 595
  char*          sql;
  char*          physicalPlan;
  SArray*        tasks;  // SArray<SArray<SStreamTask>>
  SSchemaWrapper outputSchema;
L
Liu Jicong 已提交
596 597
} SStreamObj;

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

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

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