mndDef.h 17.5 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 22

#include "cJSON.h"
#include "sync.h"
H
Hongze Cheng 已提交
23
#include "tmsg.h"
S
Shengliang Guan 已提交
24
#include "thash.h"
L
Liu Jicong 已提交
25
#include "tlist.h"
S
Shengliang Guan 已提交
26 27 28
#include "tlog.h"
#include "trpc.h"
#include "ttimer.h"
L
Liu Jicong 已提交
29
#include "scheduler.h"
S
Shengliang Guan 已提交
30

S
Shengliang Guan 已提交
31
#include "mnode.h"
S
Shengliang Guan 已提交
32

S
Shengliang Guan 已提交
33 34 35
#ifdef __cplusplus
extern "C" {
#endif
S
Shengliang Guan 已提交
36

S
Shengliang Guan 已提交
37 38 39 40 41 42 43 44 45 46 47
extern int32_t mDebugFlag;

// mnode log function
#define mFatal(...) { if (mDebugFlag & DEBUG_FATAL) { taosPrintLog("MND FATAL ", 255, __VA_ARGS__); }}
#define mError(...) { if (mDebugFlag & DEBUG_ERROR) { taosPrintLog("MND ERROR ", 255, __VA_ARGS__); }}
#define mWarn(...)  { if (mDebugFlag & DEBUG_WARN)  { taosPrintLog("MND WARN ", 255, __VA_ARGS__); }}
#define mInfo(...)  { if (mDebugFlag & DEBUG_INFO)  { taosPrintLog("MND ", 255, __VA_ARGS__); }}
#define mDebug(...) { if (mDebugFlag & DEBUG_DEBUG) { taosPrintLog("MND ", mDebugFlag, __VA_ARGS__); }}
#define mTrace(...) { if (mDebugFlag & DEBUG_TRACE) { taosPrintLog("MND ", mDebugFlag, __VA_ARGS__); }}

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

typedef enum {
58 59 60 61 62 63
  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 已提交
64

S
Shengliang Guan 已提交
65
typedef enum {
66
  TRN_STAGE_PREPARE = 0,
S
Shengliang Guan 已提交
67 68
  TRN_STAGE_REDO_LOG = 1,
  TRN_STAGE_REDO_ACTION = 2,
69 70
  TRN_STAGE_COMMIT = 3,
  TRN_STAGE_COMMIT_LOG = 4,
71 72
  TRN_STAGE_UNDO_ACTION = 5,
  TRN_STAGE_UNDO_LOG = 6,
S
Shengliang Guan 已提交
73 74
  TRN_STAGE_ROLLBACK = 7,
  TRN_STAGE_FINISHED = 8
S
Shengliang Guan 已提交
75 76
} ETrnStage;

77
typedef enum { TRN_POLICY_ROLLBACK = 0, TRN_POLICY_RETRY = 1 } ETrnPolicy;
S
Shengliang Guan 已提交
78

S
Shengliang Guan 已提交
79 80 81 82 83 84 85 86 87 88 89 90 91 92
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 已提交
93
typedef struct {
S
Shengliang Guan 已提交
94 95 96
  int32_t    id;
  ETrnStage  stage;
  ETrnPolicy policy;
S
Shengliang Guan 已提交
97 98
  int32_t    code;
  int32_t    failedTimes;
S
Shengliang Guan 已提交
99
  void      *rpcHandle;
S
Shengliang Guan 已提交
100
  void      *rpcAHandle;
S
Shengliang Guan 已提交
101 102 103 104 105 106
  SArray    *redoLogs;
  SArray    *undoLogs;
  SArray    *commitLogs;
  SArray    *redoActions;
  SArray    *undoActions;
} STrans;
S
Shengliang Guan 已提交
107

S
Shengliang Guan 已提交
108
typedef struct {
109
  int64_t id;
S
Shengliang Guan 已提交
110
  char    name[TSDB_CLUSTER_ID_LEN];
S
Shengliang Guan 已提交
111 112 113 114
  int64_t createdTime;
  int64_t updateTime;
} SClusterObj;

S
Shengliang Guan 已提交
115
typedef struct {
S
Shengliang Guan 已提交
116 117 118 119
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
  int64_t    rebootTime;
120
  int64_t    lastAccessTime;
S
Shengliang Guan 已提交
121
  int32_t    accessTimes;
S
Shengliang Guan 已提交
122
  int32_t    numOfVnodes;
S
Shengliang Guan 已提交
123 124
  int32_t    numOfSupportVnodes;
  int32_t    numOfCores;
S
Shengliang Guan 已提交
125 126 127 128
  EDndReason offlineReason;
  uint16_t   port;
  char       fqdn[TSDB_FQDN_LEN];
  char       ep[TSDB_EP_LEN];
S
Shengliang Guan 已提交
129 130
} SDnodeObj;

S
Shengliang Guan 已提交
131
typedef struct {
S
Shengliang Guan 已提交
132
  int32_t    id;
S
Shengliang Guan 已提交
133 134
  int64_t    createdTime;
  int64_t    updateTime;
S
Shengliang Guan 已提交
135
  ESyncState role;
S
Shengliang Guan 已提交
136 137
  int32_t    roleTerm;
  int64_t    roleTime;
138
  SDnodeObj *pDnode;
S
Shengliang Guan 已提交
139 140
} SMnodeObj;

S
Shengliang Guan 已提交
141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161
typedef struct {
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
  SDnodeObj *pDnode;
} SQnodeObj;

typedef struct {
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
  SDnodeObj *pDnode;
} SSnodeObj;

typedef struct {
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
  SDnodeObj *pDnode;
} SBnodeObj;

S
Shengliang Guan 已提交
162 163 164
typedef struct {
  int32_t maxUsers;
  int32_t maxDbs;
S
Shengliang Guan 已提交
165 166
  int32_t maxStbs;
  int32_t maxTbs;
S
Shengliang Guan 已提交
167 168
  int32_t maxTimeSeries;
  int32_t maxStreams;
S
Shengliang Guan 已提交
169 170 171 172
  int32_t maxFuncs;
  int32_t maxConsumers;
  int32_t maxConns;
  int32_t maxTopics;
S
Shengliang Guan 已提交
173 174
  int64_t maxStorage;   // In unit of GB
  int32_t accessState;  // Configured only by command
S
Shengliang Guan 已提交
175 176 177 178 179 180 181
} SAcctCfg;

typedef struct {
  int32_t numOfUsers;
  int32_t numOfDbs;
  int32_t numOfTimeSeries;
  int32_t numOfStreams;
S
Shengliang Guan 已提交
182 183
  int64_t totalStorage;  // Total storage wrtten from this account
  int64_t compStorage;   // Compressed storage on disk
S
Shengliang Guan 已提交
184 185
} SAcctInfo;

S
Shengliang Guan 已提交
186
typedef struct {
S
Shengliang Guan 已提交
187 188 189 190
  char      acct[TSDB_USER_LEN];
  int64_t   createdTime;
  int64_t   updateTime;
  int32_t   acctId;
S
Shengliang Guan 已提交
191
  int32_t   status;
S
Shengliang Guan 已提交
192 193 194 195
  SAcctCfg  cfg;
  SAcctInfo info;
} SAcctObj;

S
Shengliang Guan 已提交
196
typedef struct {
S
Shengliang Guan 已提交
197
  char      user[TSDB_USER_LEN];
198
  char      pass[TSDB_PASSWORD_LEN];
S
Shengliang Guan 已提交
199 200 201
  char      acct[TSDB_USER_LEN];
  int64_t   createdTime;
  int64_t   updateTime;
202
  int8_t    superUser;
S
Shengliang Guan 已提交
203
  int32_t   acctId;
S
Shengliang Guan 已提交
204
  SHashObj *prohibitDbHash;
S
Shengliang Guan 已提交
205 206 207
} SUserObj;

typedef struct {
208
  int32_t numOfVgroups;
S
Shengliang Guan 已提交
209 210 211 212 213 214
  int32_t cacheBlockSize;
  int32_t totalBlocks;
  int32_t daysPerFile;
  int32_t daysToKeep0;
  int32_t daysToKeep1;
  int32_t daysToKeep2;
S
Shengliang Guan 已提交
215 216
  int32_t minRows;
  int32_t maxRows;
S
Shengliang Guan 已提交
217 218
  int32_t commitTime;
  int32_t fsyncPeriod;
S
Shengliang Guan 已提交
219
  int8_t  walLevel;
S
Shengliang Guan 已提交
220 221 222 223 224 225 226 227
  int8_t  precision;
  int8_t  compression;
  int8_t  replications;
  int8_t  quorum;
  int8_t  update;
  int8_t  cacheLastRow;
} SDbCfg;

S
Shengliang Guan 已提交
228
typedef struct {
229
  char    name[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
230 231 232
  char    acct[TSDB_USER_LEN];
  int64_t createdTime;
  int64_t updateTime;
H
Haojun Liao 已提交
233
  uint64_t uid;
S
Shengliang Guan 已提交
234 235
  int32_t cfgVersion;
  int32_t vgVersion;
S
Shengliang Guan 已提交
236
  int8_t  hashMethod;  // default is 1
S
Shengliang Guan 已提交
237
  SDbCfg  cfg;
S
Shengliang Guan 已提交
238 239 240 241
} SDbObj;

typedef struct {
  int32_t    dnodeId;
S
Shengliang Guan 已提交
242
  ESyncState role;
S
Shengliang Guan 已提交
243 244
} SVnodeGid;

S
Shengliang Guan 已提交
245
typedef struct {
S
Shengliang Guan 已提交
246
  int32_t   vgId;
S
Shengliang Guan 已提交
247 248
  int64_t   createdTime;
  int64_t   updateTime;
S
Shengliang Guan 已提交
249
  int32_t   version;
S
Shengliang Guan 已提交
250 251
  uint32_t  hashBegin;
  uint32_t  hashEnd;
252
  char      dbName[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
253
  int64_t   dbUid;
S
Shengliang Guan 已提交
254
  int32_t   numOfTables;
S
Shengliang Guan 已提交
255
  int32_t   numOfTimeSeries;
S
Shengliang Guan 已提交
256 257 258
  int64_t   totalStorage;
  int64_t   compStorage;
  int64_t   pointsWritten;
S
Shengliang Guan 已提交
259 260 261
  int8_t    compact;
  int8_t    replica;
  SVnodeGid vnodeGid[TSDB_MAX_REPLICA];
S
Shengliang Guan 已提交
262 263
} SVgObj;

S
Shengliang Guan 已提交
264
typedef struct {
S
Shengliang Guan 已提交
265
  char     name[TSDB_TABLE_FNAME_LEN];
266
  char     db[TSDB_DB_FNAME_LEN];
267 268
  int64_t  createdTime;
  int64_t  updateTime;
S
Shengliang Guan 已提交
269
  uint64_t uid;
S
Shengliang Guan 已提交
270
  uint64_t dbUid;
S
Shengliang Guan 已提交
271
  int32_t  version;
S
Shengliang Guan 已提交
272 273
  int32_t  numOfColumns;
  int32_t  numOfTags;
S
Shengliang Guan 已提交
274
  SRWLatch lock;
S
Shengliang Guan 已提交
275 276
  SSchema *pSchema;
} SStbObj;
S
Shengliang Guan 已提交
277

S
Shengliang Guan 已提交
278
typedef struct {
S
Shengliang Guan 已提交
279 280
  char    name[TSDB_FUNC_NAME_LEN];
  int64_t createdTime;
S
Shengliang Guan 已提交
281 282 283 284 285 286
  int8_t  funcType;
  int8_t  scriptType;
  int8_t  align;
  int8_t  outputType;
  int32_t outputLen;
  int32_t bufSize;
S
Shengliang 已提交
287
  int64_t signature;
S
Shengliang Guan 已提交
288 289 290 291
  int32_t commentSize;
  int32_t codeSize;
  char   *pComment;
  char   *pCode;
S
Shengliang Guan 已提交
292 293
} SFuncObj;

S
Shengliang Guan 已提交
294
typedef struct {
295
  int64_t id;
S
Shengliang Guan 已提交
296 297 298 299 300 301 302 303 304
  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;
305
  char    db[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
306 307 308
  int16_t offset[TSDB_MAX_COLUMNS];
  int32_t bytes[TSDB_MAX_COLUMNS];
  char    payload[];
S
Shengliang Guan 已提交
309 310
} SShowObj;

L
Liu Jicong 已提交
311
#if 0
L
Liu Jicong 已提交
312
typedef struct SConsumerObj {
L
Liu Jicong 已提交
313
  uint64_t uid;
L
Liu Jicong 已提交
314 315
  int64_t  createTime;
  int64_t  updateTime;
L
Liu Jicong 已提交
316
  //uint64_t dbUid;
L
Liu Jicong 已提交
317
  int32_t  version;
L
Liu Jicong 已提交
318
  SRWLatch lock;
L
Liu Jicong 已提交
319
  SArray*   topics;
L
Liu Jicong 已提交
320 321
} SConsumerObj;

L
Liu Jicong 已提交
322 323 324 325 326 327
typedef struct SMqTopicConsumer {
  int64_t consumerId;
  SList*  topicList;
} SMqTopicConsumer;
#endif

L
Liu Jicong 已提交
328
typedef struct SMqConsumerEp {
L
Liu Jicong 已提交
329
  int32_t      vgId;      // -1 for unassigned
330
  SEpSet       epSet;
L
Liu Jicong 已提交
331 332 333 334 335
  int64_t      consumerId; // -1 for unassigned
  int64_t      lastConsumerHbTs;
  int64_t      lastVgHbTs;
  int32_t      execLen;
  SSubQueryMsg qExec;
L
Liu Jicong 已提交
336 337
} SMqConsumerEp;

L
Liu Jicong 已提交
338 339 340
static FORCE_INLINE int32_t tEncodeSMqConsumerEp(void** buf, SMqConsumerEp* pConsumerEp) {
  int32_t tlen = 0;
  tlen += taosEncodeFixedI32(buf, pConsumerEp->vgId);
341
  tlen += taosEncodeSEpSet(buf, &pConsumerEp->epSet);
L
Liu Jicong 已提交
342
  tlen += taosEncodeFixedI64(buf, pConsumerEp->consumerId);
L
Liu Jicong 已提交
343
  tlen += tEncodeSSubQueryMsg(buf, &pConsumerEp->qExec);
L
Liu Jicong 已提交
344 345 346 347 348
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqConsumerEp(void** buf, SMqConsumerEp* pConsumerEp) {
  buf = taosDecodeFixedI32(buf, &pConsumerEp->vgId);
349
  buf = taosDecodeSEpSet(buf, &pConsumerEp->epSet);
L
Liu Jicong 已提交
350
  buf = taosDecodeFixedI64(buf, &pConsumerEp->consumerId);
L
Liu Jicong 已提交
351 352
  buf = tDecodeSSubQueryMsg(buf, &pConsumerEp->qExec);
  pConsumerEp->execLen = sizeof(SSubQueryMsg) + pConsumerEp->qExec.contentLen;
L
Liu Jicong 已提交
353 354 355 356 357 358 359 360
  return buf;
}

//unit for rebalance
typedef struct SMqSubscribeObj {
  char    key[TSDB_SUBSCRIBE_KEY_LEN];
  int32_t epoch;
  //TODO: replace with priority queue
L
Liu Jicong 已提交
361
  int32_t nextConsumerIdx;
L
Liu Jicong 已提交
362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484
  SArray* availConsumer;        // SArray<int64_t> (consumerId)
  SArray* assigned;             // SArray<SMqConsumerEp>
  SArray* unassignedConsumer;   // SArray<SMqConsumerEp>
  SArray* unassignedVg;         // SArray<SMqConsumerEp>
} SMqSubscribeObj;

static FORCE_INLINE SMqSubscribeObj* tNewSubscribeObj() {
  SMqSubscribeObj* pSub = malloc(sizeof(SMqSubscribeObj));
  pSub->key[0] = 0;
  pSub->epoch = 0;
  if (pSub == NULL) {
    return NULL;
  }
  pSub->availConsumer = taosArrayInit(0, sizeof(int64_t));
  if (pSub->availConsumer == NULL) {
    free(pSub);
    return NULL;
  }
  pSub->assigned = taosArrayInit(0, sizeof(SMqConsumerEp));
  if (pSub->assigned == NULL) {
    taosArrayDestroy(pSub->availConsumer);
    free(pSub);
    return NULL;
  }
  pSub->unassignedConsumer = taosArrayInit(0, sizeof(SMqConsumerEp));
  if (pSub->assigned == NULL) {
    taosArrayDestroy(pSub->availConsumer);
    taosArrayDestroy(pSub->unassignedConsumer);
    free(pSub);
    return NULL;
  }
  pSub->unassignedVg = taosArrayInit(0, sizeof(SMqConsumerEp));
  if (pSub->assigned == NULL) {
    taosArrayDestroy(pSub->availConsumer);
    taosArrayDestroy(pSub->unassignedConsumer);
    taosArrayDestroy(pSub->unassignedVg);
    free(pSub);
    return NULL;
  }
  return NULL;
}

static FORCE_INLINE int32_t tEncodeSubscribeObj(void** buf, const SMqSubscribeObj* pSub) {
  int32_t tlen = 0;
  tlen += taosEncodeString(buf, pSub->key);
  tlen += taosEncodeFixedI32(buf, pSub->epoch);
  int32_t sz;

  sz = taosArrayGetSize(pSub->availConsumer);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    int64_t* pConsumerId = taosArrayGet(pSub->availConsumer, i);
    tlen += taosEncodeFixedI64(buf, *pConsumerId);
  }

  sz = taosArrayGetSize(pSub->assigned);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    SMqConsumerEp* pCEp = taosArrayGet(pSub->assigned, i);
    tlen += tEncodeSMqConsumerEp(buf, pCEp);
  }

  sz = taosArrayGetSize(pSub->unassignedConsumer);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    SMqConsumerEp* pCEp = taosArrayGet(pSub->unassignedConsumer, i);
    tlen += tEncodeSMqConsumerEp(buf, pCEp);
  }

  sz = taosArrayGetSize(pSub->unassignedVg);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    SMqConsumerEp* pCEp = taosArrayGet(pSub->unassignedVg, i);
    tlen += tEncodeSMqConsumerEp(buf, pCEp);
  }

  return tlen;
}

static FORCE_INLINE void* tDecodeSubscribeObj(void* buf, SMqSubscribeObj* pSub) {
  buf = taosDecodeStringTo(buf, pSub->key);
  buf = taosDecodeFixedI32(buf, &pSub->epoch);

  int32_t sz;

  buf = taosDecodeFixedI32(buf, &sz);
  pSub->assigned = taosArrayInit(sz, sizeof(int64_t));
  if (pSub->assigned == NULL) {
    return NULL;
  }
  for (int32_t i = 0; i < sz; i++) {
    int64_t consumerId;
    buf = taosDecodeFixedI64(buf, &consumerId);
    taosArrayPush(pSub->assigned, &consumerId);
  }

  buf = taosDecodeFixedI32(buf, &sz);
  pSub->unassignedConsumer = taosArrayInit(sz, sizeof(SMqConsumerEp));
  if (pSub->unassignedConsumer == NULL) {
    taosArrayDestroy(pSub->assigned);
    return NULL;
  }
  for (int32_t i = 0; i < sz; i++) {
    SMqConsumerEp cEp;
    buf = tDecodeSMqConsumerEp(buf, &cEp);
    taosArrayPush(pSub->unassignedConsumer, &cEp);
  }

  buf = taosDecodeFixedI32(buf, &sz);
  pSub->unassignedVg = taosArrayInit(sz, sizeof(SMqConsumerEp));
  if (pSub->unassignedVg == NULL) {
    taosArrayDestroy(pSub->assigned);
    taosArrayDestroy(pSub->unassignedConsumer);
    return NULL;
  }
  for (int32_t i = 0; i < sz; i++) {
    SMqConsumerEp cEp;
    buf = tDecodeSMqConsumerEp(buf, &cEp);
    taosArrayPush(pSub->unassignedVg, &cEp);
  }

  return buf;
}
L
Liu Jicong 已提交
485

L
Liu Jicong 已提交
486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505
typedef struct SMqCGroup {
  char    name[TSDB_CONSUMER_GROUP_LEN];
  int32_t status;       // 0 - uninitialized, 1 - wait rebalance, 2- normal
  SList  *consumerIds;  // SList<int64_t>
  SList  *idleVGroups;  // SList<int32_t>
} SMqCGroup;

typedef struct SMqTopicObj {
  char      name[TSDB_TOPIC_FNAME_LEN];
  char      db[TSDB_DB_FNAME_LEN];
  int64_t   createTime;
  int64_t   updateTime;
  uint64_t  uid;
  uint64_t  dbUid;
  int32_t   version;
  SRWLatch  lock;
  int32_t   sqlLen;
  char     *sql;
  char     *logicalPlan;
  char     *physicalPlan;
L
Liu Jicong 已提交
506 507
  //SHashObj *cgroups;  // SHashObj<SMqCGroup>
  //SHashObj *consumers;  // SHashObj<SMqConsumerObj>
L
Liu Jicong 已提交
508 509 510 511
} SMqTopicObj;

// TODO: add cache and change name to id
typedef struct SMqConsumerTopic {
L
Liu Jicong 已提交
512
  char    name[TSDB_TOPIC_FNAME_LEN];
L
Liu Jicong 已提交
513 514
  int32_t epoch;
  //TODO: replace with something with ep
L
Liu Jicong 已提交
515 516
  //SList   *vgroups;  // SList<int32_t>
  //vg assigned to the consumer on the topic
L
Liu Jicong 已提交
517
  SArray  *pVgInfo;  // SArray<int32_t>
L
Liu Jicong 已提交
518 519
} SMqConsumerTopic;

L
Liu Jicong 已提交
520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566
static FORCE_INLINE SMqConsumerTopic* tNewConsumerTopic(int64_t consumerId, SMqTopicObj* pTopic, SMqSubscribeObj* pSub) {
  SMqConsumerTopic* pCTopic = malloc(sizeof(SMqConsumerTopic));
  if (pCTopic == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }
  strcpy(pCTopic->name, pTopic->name);
  pCTopic->epoch = 0;
  pCTopic->pVgInfo = taosArrayInit(0, sizeof(int32_t));

  int32_t unassignedVgSz = taosArrayGetSize(pSub->unassignedVg);
  if (unassignedVgSz > 0) {
    SMqConsumerEp* pCEp = taosArrayPop(pSub->unassignedVg);
    pCEp->consumerId = consumerId;
    taosArrayPush(pCTopic->pVgInfo, &pCEp->vgId);
    taosArrayPush(pSub->assigned, pCEp);
  }
  return pCTopic;
}

static FORCE_INLINE int32_t tEncodeSMqConsumerTopic(void** buf, SMqConsumerTopic* pConsumerTopic) {
  int32_t tlen = 0;
  tlen += taosEncodeString(buf, pConsumerTopic->name);
  tlen += taosEncodeFixedI32(buf, pConsumerTopic->epoch);
  int32_t sz = taosArrayGetSize(pConsumerTopic->pVgInfo);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    int32_t* pVgInfo = taosArrayGet(pConsumerTopic->pVgInfo, i);
    tlen += taosEncodeFixedI32(buf, *pVgInfo);
  }
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqConsumerTopic(void* buf, SMqConsumerTopic* pConsumerTopic) {
  buf = taosDecodeStringTo(buf, pConsumerTopic->name);
  buf = taosDecodeFixedI32(buf, &pConsumerTopic->epoch);
  int32_t sz;
  buf = taosDecodeFixedI32(buf, &sz);
  pConsumerTopic->pVgInfo = taosArrayInit(sz, sizeof(SMqConsumerTopic));
  for (int32_t i = 0; i < sz; i++) {
    int32_t vgInfo;
    buf = taosDecodeFixedI32(buf, &vgInfo);
    taosArrayPush(pConsumerTopic->pVgInfo, &vgInfo);
  }
  return buf;
}

L
Liu Jicong 已提交
567 568
typedef struct SMqConsumerObj {
  int64_t  consumerId;
L
Liu Jicong 已提交
569
  SRWLatch lock;
L
Liu Jicong 已提交
570 571
  char     cgroup[TSDB_CONSUMER_GROUP_LEN];
  SArray  *topics;  // SArray<SMqConsumerTopic>
L
Liu Jicong 已提交
572
  //SHashObj *topicHash; //SHashObj<SMqTopicObj>
L
Liu Jicong 已提交
573 574
} SMqConsumerObj;

L
Liu Jicong 已提交
575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601
static FORCE_INLINE int32_t tEncodeSMqConsumerObj(void** buf, const SMqConsumerObj* pConsumer) {
  int32_t tlen = 0;
  tlen += taosEncodeFixedI64(buf, pConsumer->consumerId);
  tlen += taosEncodeString(buf, pConsumer->cgroup);
  int32_t sz = taosArrayGetSize(pConsumer->topics);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    SMqConsumerTopic* pConsumerTopic = taosArrayGet(pConsumer->topics, i);
    tlen += tEncodeSMqConsumerTopic(buf, pConsumerTopic);
  }
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqConsumerObj(void* buf, SMqConsumerObj* pConsumer) {
  buf = taosDecodeFixedI64(buf, &pConsumer->consumerId);
  buf = taosDecodeStringTo(buf, pConsumer->cgroup);
  int32_t sz;
  buf = taosDecodeFixedI32(buf, &sz);
  pConsumer->topics = taosArrayInit(sz, sizeof(SMqConsumerObj));
  for (int32_t i = 0; i < sz; i++) {
    SMqConsumerTopic cTopic;
    buf = tDecodeSMqConsumerTopic(buf, &cTopic);
    taosArrayPush(pConsumer->topics, &cTopic);
  }
  return buf;
}

L
Liu Jicong 已提交
602 603
typedef struct SMqSubConsumerObj {
  int64_t consumerUid;  // if -1, unassigned
L
Liu Jicong 已提交
604
  SList  *vgId;         // SList<int32_t>
L
Liu Jicong 已提交
605 606 607 608
} SMqSubConsumerObj;

typedef struct SMqSubCGroupObj {
  char   name[TSDB_CONSUMER_GROUP_LEN];
L
Liu Jicong 已提交
609
  SList *consumers;  // SList<SMqConsumerObj>
L
Liu Jicong 已提交
610 611 612 613 614 615 616 617 618 619
} SMqSubCGroupObj;

typedef struct SMqSubTopicObj {
  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 已提交
620 621
  SRWLatch lock;
  int32_t  sqlLen;
L
Liu Jicong 已提交
622 623 624 625
  char    *sql;
  char    *logicalPlan;
  char    *physicalPlan;
  SList   *cgroups;  // SList<SMqSubCGroupObj>
L
Liu Jicong 已提交
626 627 628 629
} SMqSubTopicObj;

typedef struct SMqConsumerSubObj {
  int64_t topicUid;
L
Liu Jicong 已提交
630
  SList  *vgIds;  // SList<int64_t>
L
Liu Jicong 已提交
631 632 633 634
} SMqConsumerSubObj;

typedef struct SMqConsumerHbObj {
  int64_t consumerId;
L
Liu Jicong 已提交
635
  SList  *consumerSubs;  // SList<SMqConsumerSubObj>
L
Liu Jicong 已提交
636 637 638 639
} SMqConsumerHbObj;

typedef struct SMqVGroupSubObj {
  int64_t topicUid;
L
Liu Jicong 已提交
640
  SList  *consumerIds;  // SList<int64_t>
L
Liu Jicong 已提交
641 642 643 644
} SMqVGroupSubObj;

typedef struct SMqVGroupHbObj {
  int64_t vgId;
L
Liu Jicong 已提交
645
  SList  *vgSubs;  // SList<SMqVGroupSubObj>
L
Liu Jicong 已提交
646 647
} SMqVGroupHbObj;

L
Liu Jicong 已提交
648
#if 0
L
Liu Jicong 已提交
649
typedef struct SCGroupObj {
650
  char     name[TSDB_TOPIC_NAME_LEN];
L
Liu Jicong 已提交
651 652
  int64_t  createTime;
  int64_t  updateTime;
L
Liu Jicong 已提交
653 654
  uint64_t uid;
  //uint64_t dbUid;
L
Liu Jicong 已提交
655
  int32_t  version;
L
Liu Jicong 已提交
656
  SRWLatch lock;
L
Liu Jicong 已提交
657
  SList*   consumerIds;
L
Liu Jicong 已提交
658
} SCGroupObj;
L
Liu Jicong 已提交
659
#endif
L
Liu Jicong 已提交
660

S
Shengliang Guan 已提交
661
typedef struct SMnodeMsg {
S
Shengliang Guan 已提交
662
  char    user[TSDB_USER_LEN];
663
  char    db[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
664
  int32_t acctId;
665
  SMnode *pMnode;
S
Shengliang Guan 已提交
666 667 668 669
  int64_t createdTime;
  SRpcMsg rpcMsg;
  int32_t contLen;
  void   *pCont;
S
Shengliang Guan 已提交
670
} SMnodeMsg;
S
Shengliang Guan 已提交
671

S
Shengliang Guan 已提交
672 673 674
#ifdef __cplusplus
}
#endif
S
Shengliang Guan 已提交
675

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