mndDef.h 18.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
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
  TRN_STAGE_REDO_ACTION = 1,
  TRN_STAGE_ROLLBACK = 2,
  TRN_STAGE_UNDO_ACTION = 3,
  TRN_STAGE_COMMIT = 4,
  TRN_STAGE_COMMIT_ACTION = 5,
111
  TRN_STAGE_FINISH = 6,
112
  TRN_STAGE_PRE_FINISH = 7
S
Shengliang Guan 已提交
113 114
} ETrnStage;

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

120
typedef enum {
B
Benguang Zhao 已提交
121
  TRN_EXEC_PARALLEL = 0,
122 123
  TRN_EXEC_SERIAL = 1,
} ETrnExec;
124

S
Shengliang Guan 已提交
125 126 127 128 129 130 131 132 133 134 135 136 137 138
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 已提交
139
typedef enum {
wmmhello's avatar
wmmhello 已提交
140 141 142
  CONSUMER_UPDATE_REB_MODIFY_NOTOPIC = 1,   // topic do not need modified after rebalance
  CONSUMER_UPDATE_REB_MODIFY_TOPIC,         // topic need modified after rebalance
  CONSUMER_UPDATE_REB_MODIFY_REMOVE,        // topic need removed after rebalance
wmmhello's avatar
wmmhello 已提交
143
//  CONSUMER_UPDATE_TIMER_LOST,
wmmhello's avatar
wmmhello 已提交
144 145
  CONSUMER_UPDATE_RECOVER,
  CONSUMER_UPDATE_SUB_MODIFY,      // modify after subscribe req
S
Shengliang Guan 已提交
146 147
} ECsmUpdateType;

S
Shengliang Guan 已提交
148
typedef struct {
L
Liu Jicong 已提交
149 150 151 152 153 154 155 156 157 158 159
  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;
160
  SArray*     prepareActions;
L
Liu Jicong 已提交
161 162 163 164 165 166 167 168
  SArray*     redoActions;
  SArray*     undoActions;
  SArray*     commitActions;
  int64_t     createdTime;
  int64_t     lastExecTime;
  int32_t     lastAction;
  int32_t     lastErrorNo;
  SEpSet      lastEpset;
169 170
  tmsg_t      lastMsgType;
  tmsg_t      originRpcType;
171 172
  char        dbname[TSDB_TABLE_FNAME_LEN];
  char        stbname[TSDB_TABLE_FNAME_LEN];
L
Liu Jicong 已提交
173 174 175 176
  int32_t     startFunc;
  int32_t     stopFunc;
  int32_t     paramLen;
  void*       param;
177
  char        opername[TSDB_TRANS_OPER_LEN];
L
Liu Jicong 已提交
178
  SArray*     pRpcArray;
179
  SRWLatch    lockRpcArray;
dengyihao's avatar
dengyihao 已提交
180
  int64_t     mTraceId;
D
dmchen 已提交
181
  TdThreadMutex    mutex;
S
Shengliang Guan 已提交
182
} STrans;
S
Shengliang Guan 已提交
183

S
Shengliang Guan 已提交
184
typedef struct {
185
  int64_t id;
S
Shengliang Guan 已提交
186
  char    name[TSDB_CLUSTER_ID_LEN];
S
Shengliang Guan 已提交
187 188
  int64_t createdTime;
  int64_t updateTime;
189
  int32_t upTime;
S
Shengliang Guan 已提交
190 191
} SClusterObj;

S
Shengliang Guan 已提交
192
typedef struct {
S
Shengliang Guan 已提交
193 194 195 196
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
  int64_t    rebootTime;
197
  int64_t    lastAccessTime;
S
Shengliang Guan 已提交
198
  int32_t    accessTimes;
S
Shengliang Guan 已提交
199
  int32_t    numOfVnodes;
200
  int32_t    numOfOtherNodes;
S
Shengliang Guan 已提交
201
  int32_t    numOfSupportVnodes;
S
Shengliang Guan 已提交
202
  float      numOfCores;
203 204 205
  int64_t    memTotal;
  int64_t    memAvail;
  int64_t    memUsed;
S
Shengliang Guan 已提交
206 207 208 209
  EDndReason offlineReason;
  uint16_t   port;
  char       fqdn[TSDB_FQDN_LEN];
  char       ep[TSDB_EP_LEN];
K
kailixu 已提交
210 211
  char       active[TSDB_ACTIVE_KEY_LEN];
  char       connActive[TSDB_CONN_ACTIVE_KEY_LEN];
S
Shengliang Guan 已提交
212 213
} SDnodeObj;

S
Shengliang Guan 已提交
214
typedef struct {
S
Shengliang Guan 已提交
215
  int32_t    id;
S
Shengliang Guan 已提交
216 217
  int64_t    createdTime;
  int64_t    updateTime;
218
  ESyncState syncState;
S
Shungang Li 已提交
219
  SyncTerm   syncTerm;
220
  bool       syncRestore;
S
Shungang Li 已提交
221
  int64_t    roleTimeMs;
L
Liu Jicong 已提交
222
  SDnodeObj* pDnode;
C
cadem 已提交
223 224
  int32_t    role;
  SyncIndex  lastIndex;
S
Shengliang Guan 已提交
225 226
} SMnodeObj;

S
Shengliang Guan 已提交
227 228 229 230
typedef struct {
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
L
Liu Jicong 已提交
231
  SDnodeObj* pDnode;
L
Liu Jicong 已提交
232
  SQnodeLoad load;
S
Shengliang Guan 已提交
233 234 235 236 237 238
} SQnodeObj;

typedef struct {
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
L
Liu Jicong 已提交
239
  SDnodeObj* pDnode;
S
Shengliang Guan 已提交
240 241
} SSnodeObj;

S
Shengliang Guan 已提交
242 243 244
typedef struct {
  int32_t maxUsers;
  int32_t maxDbs;
S
Shengliang Guan 已提交
245 246
  int32_t maxStbs;
  int32_t maxTbs;
S
Shengliang Guan 已提交
247 248
  int32_t maxTimeSeries;
  int32_t maxStreams;
S
Shengliang Guan 已提交
249 250 251 252
  int32_t maxFuncs;
  int32_t maxConsumers;
  int32_t maxConns;
  int32_t maxTopics;
253
  int64_t maxStorage;
S
Shengliang Guan 已提交
254
  int32_t accessState;  // Configured only by command
S
Shengliang Guan 已提交
255 256 257 258 259 260 261
} SAcctCfg;

typedef struct {
  int32_t numOfUsers;
  int32_t numOfDbs;
  int32_t numOfTimeSeries;
  int32_t numOfStreams;
S
Shengliang Guan 已提交
262 263
  int64_t totalStorage;  // Total storage wrtten from this account
  int64_t compStorage;   // Compressed storage on disk
S
Shengliang Guan 已提交
264 265
} SAcctInfo;

S
Shengliang Guan 已提交
266
typedef struct {
S
Shengliang Guan 已提交
267 268 269 270
  char      acct[TSDB_USER_LEN];
  int64_t   createdTime;
  int64_t   updateTime;
  int32_t   acctId;
S
Shengliang Guan 已提交
271
  int32_t   status;
S
Shengliang Guan 已提交
272 273 274 275
  SAcctCfg  cfg;
  SAcctInfo info;
} SAcctObj;

S
Shengliang Guan 已提交
276
typedef struct {
S
Shengliang Guan 已提交
277
  char      user[TSDB_USER_LEN];
278
  char      pass[TSDB_PASSWORD_LEN];
S
Shengliang Guan 已提交
279 280 281
  char      acct[TSDB_USER_LEN];
  int64_t   createdTime;
  int64_t   updateTime;
282
  int8_t    superUser;
283 284 285
  int8_t    sysInfo;
  int8_t    enable;
  int8_t    reserve;
S
Shengliang Guan 已提交
286
  int32_t   acctId;
D
dapan 已提交
287
  int32_t   authVersion;
K
kailixu 已提交
288
  int32_t   passVersion;
289 290
  SHashObj* readDbs;
  SHashObj* writeDbs;
291
  SHashObj* topics;
C
cademfly 已提交
292 293
  SHashObj* readTbs;
  SHashObj* writeTbs;
X
Xiaoyu Wang 已提交
294
  SHashObj* useDbs;
S
Shengliang Guan 已提交
295
  SRWLatch  lock;
S
Shengliang Guan 已提交
296 297 298
} SUserObj;

typedef struct {
299
  int32_t numOfVgroups;
S
Shengliang Guan 已提交
300 301 302 303
  int32_t numOfStables;
  int32_t buffer;
  int32_t pageSize;
  int32_t pages;
304
  int32_t cacheLastSize;
S
Shengliang Guan 已提交
305 306 307 308
  int32_t daysPerFile;
  int32_t daysToKeep0;
  int32_t daysToKeep1;
  int32_t daysToKeep2;
S
Shengliang Guan 已提交
309 310
  int32_t minRows;
  int32_t maxRows;
311
  int32_t walFsyncPeriod;
S
Shengliang Guan 已提交
312
  int8_t  walLevel;
S
Shengliang Guan 已提交
313 314 315
  int8_t  precision;
  int8_t  compression;
  int8_t  replications;
S
Shengliang Guan 已提交
316 317
  int8_t  strict;
  int8_t  hashMethod;  // default is 1
318
  int8_t  cacheLast;
S
Shengliang Guan 已提交
319
  int8_t  schemaless;
320 321 322
  int16_t hashPrefix;
  int16_t hashSuffix;
  int16_t sstTrigger;
323
  int32_t tsdbPageSize;
S
sma  
Shengliang Guan 已提交
324 325
  int32_t numOfRetensions;
  SArray* pRetensions;
S
Shengliang Guan 已提交
326 327
  int32_t walRetentionPeriod;
  int32_t walRollPeriod;
328
  int64_t walRetentionSize;
L
Liu Jicong 已提交
329
  int64_t walSegmentSize;
S
Shengliang Guan 已提交
330 331
} SDbCfg;

S
Shengliang Guan 已提交
332
typedef struct {
333 334 335 336 337 338 339 340 341 342
  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;
343
  int64_t  stateTs;
344
  int64_t  compactStartTime;
S
Shengliang Guan 已提交
345 346 347 348
} SDbObj;

typedef struct {
  int32_t    dnodeId;
349 350
  ESyncState syncState;
  bool       syncRestore;
351
  bool       syncCanRead;
C
cadem 已提交
352
  ESyncRole  nodeRole;
S
Shengliang Guan 已提交
353 354
} SVnodeGid;

S
Shengliang Guan 已提交
355
typedef struct {
S
Shengliang Guan 已提交
356
  int32_t   vgId;
S
Shengliang Guan 已提交
357 358
  int64_t   createdTime;
  int64_t   updateTime;
S
Shengliang Guan 已提交
359
  int32_t   version;
S
Shengliang Guan 已提交
360 361
  uint32_t  hashBegin;
  uint32_t  hashEnd;
362
  char      dbName[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
363
  int64_t   dbUid;
364
  int64_t   cacheUsage;
S
Shengliang Guan 已提交
365 366
  int64_t   numOfTables;
  int64_t   numOfTimeSeries;
S
Shengliang Guan 已提交
367 368 369
  int64_t   totalStorage;
  int64_t   compStorage;
  int64_t   pointsWritten;
S
Shengliang Guan 已提交
370
  int8_t    compact;
S
Shengliang Guan 已提交
371
  int8_t    isTsma;
S
Shengliang Guan 已提交
372
  int8_t    replica;
C
cadem 已提交
373
  SVnodeGid vnodeGid[TSDB_MAX_REPLICA + TSDB_MAX_LEARNER_REPLICA];
C
Cary Xu 已提交
374
  void*     pTsma;
D
dapan1121 已提交
375
  int32_t   numOfCachedTables;
S
Shengliang Guan 已提交
376 377
} SVgObj;

S
Shengliang Guan 已提交
378
typedef struct {
C
Cary Xu 已提交
379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404
  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 已提交
405 406
} SSmaObj;

dengyihao's avatar
dengyihao 已提交
407
typedef struct {
dengyihao's avatar
dengyihao 已提交
408
  char    name[TSDB_INDEX_FNAME_LEN];
dengyihao's avatar
dengyihao 已提交
409 410 411 412 413 414 415 416 417 418
  char    stb[TSDB_TABLE_FNAME_LEN];
  char    db[TSDB_DB_FNAME_LEN];
  char    dstTbName[TSDB_TABLE_FNAME_LEN];
  char    colName[TSDB_COL_NAME_LEN];
  int64_t createdTime;
  int64_t uid;
  int64_t stbUid;
  int64_t dbUid;
} SIdxObj;

S
Shengliang Guan 已提交
419
typedef struct {
S
Shengliang Guan 已提交
420
  char     name[TSDB_TABLE_FNAME_LEN];
421
  char     db[TSDB_DB_FNAME_LEN];
422 423
  int64_t  createdTime;
  int64_t  updateTime;
L
Liu Jicong 已提交
424 425
  int64_t  uid;
  int64_t  dbUid;
426 427
  int32_t  tagVer;
  int32_t  colVer;
D
dapan1121 已提交
428
  int32_t  smaVer;
S
Shengliang Guan 已提交
429
  int32_t  nextColId;
430
  int64_t  maxdelay[2];
S
Shengliang Guan 已提交
431
  int64_t  watermark[2];
S
sma  
Shengliang Guan 已提交
432
  int32_t  ttl;
S
Shengliang Guan 已提交
433 434
  int32_t  numOfColumns;
  int32_t  numOfTags;
S
Shengliang Guan 已提交
435
  int32_t  numOfFuncs;
S
sma  
Shengliang Guan 已提交
436
  int32_t  commentLen;
437 438
  int32_t  ast1Len;
  int32_t  ast2Len;
D
dapan1121 已提交
439
  SArray*  pFuncs;
S
Shengliang Guan 已提交
440
  SSchema* pColumns;
S
Shengliang Guan 已提交
441
  SSchema* pTags;
S
sma  
Shengliang Guan 已提交
442
  char*    comment;
443 444
  char*    pAst1;
  char*    pAst2;
S
Shengliang Guan 已提交
445
  SRWLatch lock;
S
Shengliang Guan 已提交
446
} SStbObj;
S
Shengliang Guan 已提交
447

S
Shengliang Guan 已提交
448
typedef struct {
S
Shengliang Guan 已提交
449 450
  char    name[TSDB_FUNC_NAME_LEN];
  int64_t createdTime;
S
Shengliang Guan 已提交
451 452 453 454 455 456
  int8_t  funcType;
  int8_t  scriptType;
  int8_t  align;
  int8_t  outputType;
  int32_t outputLen;
  int32_t bufSize;
S
Shengliang 已提交
457
  int64_t signature;
S
Shengliang Guan 已提交
458 459
  int32_t commentSize;
  int32_t codeSize;
L
Liu Jicong 已提交
460 461
  char*   pComment;
  char*   pCode;
462
  int32_t funcVersion;
“happyguoxy” 已提交
463
  SRWLatch lock;
S
Shengliang Guan 已提交
464 465
} SFuncObj;

S
Shengliang Guan 已提交
466
typedef struct {
L
Liu Jicong 已提交
467 468 469 470 471
  int64_t        id;
  int8_t         type;
  int8_t         replica;
  int16_t        numOfColumns;
  int32_t        numOfRows;
D
dapan1121 已提交
472
  int32_t        curIterPackedRows;
L
Liu Jicong 已提交
473 474
  void*          pIter;
  SMnode*        pMnode;
475
  STableMetaRsp* pMeta;
D
dapan1121 已提交
476
  bool           restore;
L
Liu Jicong 已提交
477 478
  bool           sysDbRsp;
  char           db[TSDB_DB_FNAME_LEN];
wmmhello's avatar
wmmhello 已提交
479
  char           filterTb[TSDB_TABLE_NAME_LEN];
S
Shengliang Guan 已提交
480 481
} SShowObj;

482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498
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 已提交
499 500
typedef struct {
  char    key[TSDB_PARTITION_KEY_LEN];
L
Liu Jicong 已提交
501
  int64_t dbUid;
L
Liu Jicong 已提交
502 503 504
  int64_t offset;
} SMqOffsetObj;

S
Shengliang Guan 已提交
505 506
int32_t tEncodeSMqOffsetObj(void** buf, const SMqOffsetObj* pOffset);
void*   tDecodeSMqOffsetObj(void* buf, SMqOffsetObj* pOffset);
L
Liu Jicong 已提交
507

L
Liu Jicong 已提交
508
typedef struct {
L
Liu Jicong 已提交
509 510
  char           name[TSDB_TOPIC_FNAME_LEN];
  char           db[TSDB_DB_FNAME_LEN];
511
  char           createUser[TSDB_USER_LEN];
L
Liu Jicong 已提交
512 513 514 515 516
  int64_t        createTime;
  int64_t        updateTime;
  int64_t        uid;
  int64_t        dbUid;
  int32_t        version;
L
Liu Jicong 已提交
517 518
  int8_t         subType;   // column, db or stable
  int8_t         withMeta;  // TODO
L
Liu Jicong 已提交
519 520
  SRWLatch       lock;
  int32_t        sqlLen;
L
Liu Jicong 已提交
521
  int32_t        astLen;
L
Liu Jicong 已提交
522
  char*          sql;
L
Liu Jicong 已提交
523
  char*          ast;
L
Liu Jicong 已提交
524 525
  char*          physicalPlan;
  SSchemaWrapper schema;
L
Liu Jicong 已提交
526
  int64_t        stbUid;
527
  char           stbName[TSDB_TABLE_FNAME_LEN];
L
Liu Jicong 已提交
528 529 530
  // forbid condition
  int64_t ntbUid;
  SArray* ntbColIds;
L
Liu Jicong 已提交
531
  int64_t ctbStbUid;
L
Liu Jicong 已提交
532 533
} SMqTopicObj;

L
Liu Jicong 已提交
534
typedef struct {
S
Shengliang Guan 已提交
535 536
  int64_t  consumerId;
  char     cgroup[TSDB_CGROUP_LEN];
L
Liu Jicong 已提交
537
  char     clientId[256];
S
Shengliang Guan 已提交
538 539 540 541 542
  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 已提交
543 544 545
  SArray*  currentTopics;     // SArray<char*>
  SArray*  rebNewTopics;      // SArray<char*>
  SArray*  rebRemovedTopics;  // SArray<char*>
L
Liu Jicong 已提交
546

L
Liu Jicong 已提交
547 548 549
  // subscribed by user
  SArray* assignedTopics;  // SArray<char*>

L
Liu Jicong 已提交
550 551 552
  // data for display
  int32_t pid;
  SEpSet  ep;
wmmhello's avatar
wmmhello 已提交
553
  int64_t createTime;
L
Liu Jicong 已提交
554 555
  int64_t subscribeTime;
  int64_t rebalanceTime;
556 557 558 559 560

  int8_t         withTbName;
  int8_t         autoCommit;
  int32_t        autoCommitInterval;
  int32_t        resetOffsetCfg;
L
Liu Jicong 已提交
561 562
} SMqConsumerObj;

563
SMqConsumerObj* tNewSMqConsumerObj(int64_t consumerId, char cgroup[TSDB_CGROUP_LEN]);
wmmhello's avatar
wmmhello 已提交
564
void            tDeleteSMqConsumerObj(SMqConsumerObj* pConsumer, bool delete);
565
int32_t         tEncodeSMqConsumerObj(void** buf, const SMqConsumerObj* pConsumer);
566
void*           tDecodeSMqConsumerObj(const void* buf, SMqConsumerObj* pConsumer, int8_t sver);
L
Liu Jicong 已提交
567

568 569
typedef struct {
  int32_t vgId;
570
//  char*   qmsg;  // SubPlanToString
L
Liu Jicong 已提交
571
  SEpSet  epSet;
572 573 574 575 576
} SMqVgEp;

SMqVgEp* tCloneSMqVgEp(const SMqVgEp* pVgEp);
void     tDeleteSMqVgEp(SMqVgEp* pVgEp);
int32_t  tEncodeSMqVgEp(void** buf, const SMqVgEp* pVgEp);
577
void*    tDecodeSMqVgEp(const void* buf, SMqVgEp* pVgEp, int8_t sver);
578 579 580 581

typedef struct {
  int64_t consumerId;  // -1 for unassigned
  SArray* vgs;         // SArray<SMqVgEp*>
582
  SArray* offsetRows;  // SArray<OffsetRows*>
583
} SMqConsumerEp;
584

585 586
//SMqConsumerEp* tCloneSMqConsumerEp(const SMqConsumerEp* pEp);
//void           tDeleteSMqConsumerEp(void* pEp);
587
int32_t        tEncodeSMqConsumerEp(void** buf, const SMqConsumerEp* pEp);
588
void*          tDecodeSMqConsumerEp(const void* buf, SMqConsumerEp* pEp, int8_t sver);
589 590

typedef struct {
L
Liu Jicong 已提交
591 592 593 594 595
  char      key[TSDB_SUBSCRIBE_KEY_LEN];
  SRWLatch  lock;
  int64_t   dbUid;
  int32_t   vgNum;
  int8_t    subType;
L
Liu Jicong 已提交
596
  int8_t    withMeta;
L
Liu Jicong 已提交
597
  int64_t   stbUid;
598 599
  SHashObj* consumerHash;   // consumerId -> SMqConsumerEp
  SArray*   unassignedVgs;  // SArray<SMqVgEp*>
600
  SArray*   offsetRows;
L
Liu Jicong 已提交
601
  char      dbName[TSDB_DB_FNAME_LEN];
602
  char*     qmsg;           // SubPlanToString
603 604 605 606 607 608
} 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);
609
void*            tDecodeSubscribeObj(const void* buf, SMqSubscribeObj* pSub, int8_t sver);
610

D
dapan1121 已提交
611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629
//typedef struct {
//  int32_t epoch;
//  SArray* consumers;  // SArray<SMqConsumerEp*>
//} 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);
630 631

typedef struct {
L
Liu Jicong 已提交
632 633
  int32_t           oldConsumerNum;
  const SMqRebInfo* pRebInfo;
634 635 636 637 638 639 640 641 642 643 644 645
} 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>
646
  SArray*               modifyConsumers;   // SArray<int64_t>
647
  SMqSubscribeObj*      pSub;
D
dapan1121 已提交
648
//  SMqSubActionLogEntry* pLogEntry;
649
} SMqRebOutputObj;
L
Liu Jicong 已提交
650

651 652 653 654 655 656 657 658
typedef struct SStreamConf {
  int8_t  igExpired;
  int8_t  trigger;
  int8_t  fillHistory;
  int64_t triggerParam;
  int64_t watermark;
} SStreamConf;

L
Liu Jicong 已提交
659
typedef struct {
L
Liu Jicong 已提交
660 661 662 663 664 665 666
  char name[TSDB_STREAM_FNAME_LEN];
  // ctl
  SRWLatch lock;
  // create info
  int64_t createTime;
  int64_t updateTime;
  int32_t version;
L
Liu Jicong 已提交
667
  int32_t totalLevel;
L
Liu Jicong 已提交
668 669 670 671
  int64_t smaId;  // 0 for unused
  // info
  int64_t uid;
  int8_t  status;
672
  SStreamConf conf;
L
Liu Jicong 已提交
673 674 675 676 677 678 679
  // 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;
680

L
Liu Jicong 已提交
681
  // fixedSinkVg is not applicable for encode and decode
682
  SVgObj  fixedSinkVg;
683
  int32_t fixedSinkVgId;  // 0 for shuffle
L
Liu Jicong 已提交
684

L
Liu Jicong 已提交
685
  // transformation
L
Liu Jicong 已提交
686
  char*          sql;
L
Liu Jicong 已提交
687
  char*          ast;
L
Liu Jicong 已提交
688
  char*          physicalPlan;
689 690
  SArray*        tasks;        // SArray<SArray<SStreamTask>>

691 692
  SArray*        pHTasksList;   // generate the results for already stored ts data
  int64_t        hTaskUid; // stream task for history ts data
693

L
Liu Jicong 已提交
694
  SSchemaWrapper outputSchema;
695
  SSchemaWrapper tagSchema;
696 697 698 699

  // 3.0.20
  int64_t checkpointFreq;  // ms
  int64_t currentTick;     // do not serialize
700
  int64_t deleteMark;
701
  int8_t  igCheckUpdate;
L
Liu Jicong 已提交
702 703
} SStreamObj;

H
Hongze Cheng 已提交
704
int32_t tEncodeSStreamObj(SEncoder* pEncoder, const SStreamObj* pObj);
705
int32_t tDecodeSStreamObj(SDecoder* pDecoder, SStreamObj* pObj, int32_t sver);
L
Liu Jicong 已提交
706
void    tFreeStreamObj(SStreamObj* pObj);
L
Liu Jicong 已提交
707

708 709 710 711 712 713
//typedef struct {
//  char    streamName[TSDB_STREAM_FNAME_LEN];
//  int64_t uid;
//  int64_t streamUid;
//  SArray* childInfo;  // SArray<SStreamChildEpInfo>
//} SStreamCheckpointObj;
L
Liu Jicong 已提交
714

S
Shengliang Guan 已提交
715 716 717
#ifdef __cplusplus
}
#endif
S
Shengliang Guan 已提交
718

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