mndDef.h 18.1 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 29
#include "trpc.h"
#include "ttimer.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
extern int32_t mDebugFlag;

// mnode log function
L
Liu Jicong 已提交
40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75
#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__); \
    }                                                \
  }
S
Shengliang Guan 已提交
76 77

typedef enum {
78 79 80 81 82 83 84 85
  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 已提交
86 87

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

S
Shengliang Guan 已提交
95
typedef enum {
96
  TRN_STAGE_PREPARE = 0,
S
Shengliang Guan 已提交
97 98
  TRN_STAGE_REDO_LOG = 1,
  TRN_STAGE_REDO_ACTION = 2,
99 100
  TRN_STAGE_COMMIT = 3,
  TRN_STAGE_COMMIT_LOG = 4,
101 102
  TRN_STAGE_UNDO_ACTION = 5,
  TRN_STAGE_UNDO_LOG = 6,
S
Shengliang Guan 已提交
103 104
  TRN_STAGE_ROLLBACK = 7,
  TRN_STAGE_FINISHED = 8
S
Shengliang Guan 已提交
105 106
} ETrnStage;

S
Shengliang Guan 已提交
107
typedef enum {
S
Shengliang Guan 已提交
108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135
  TRN_TYPE_START = 0,
  TRN_TYPE_CREATE_MNODE = 1,
  TRN_TYPE_DROP_MNODE = 2,
  TRN_TYPE_CREATE_DNODE = 3,
  TRN_TYPE_DROP_DNODE = 4,
  TRN_TYPE_CREATE_SNODE = 5,
  TRN_TYPE_DROP_SNODE = 6,
  TRN_TYPE_CREATE_QNODE = 7,
  TRN_TYPE_DROP_QNODE = 8,
  TRN_TYPE_CREATE_BNODE = 9,
  TRN_TYPE_DROP_BNODE = 10,
  TRN_TYPE_DB_START = 1000,
  TRN_TYPE_CREATE_DB = 1001,
  TRN_TYPE_ALTER_DB = 1002,
  TRN_TYPE_DROP_DB = 1003,
  TRN_TYPE_CREATE_FUNC = 1004,
  TRN_TYPE_DROP_FUNC = 1005,
  TRN_TYPE_CREATE_STB = 1006,
  TRN_TYPE_ALTER_STB = 1007,
  TRN_TYPE_DROP_STB = 1008,
  TRN_TYPE_CREATE_TOPIC = 1009,
  TRN_TYPE_DROP_TOPIC = 1010,
  TRN_TYPE_CREATE_USER = 1011,
  TRN_TYPE_ALTER_USER = 1012,
  TRN_TYPE_DROP_USER = 1013,
  TRN_TYPE_SUBSCRIBE = 1014,
  TRN_TYPE_REBALANCE = 1015,
  TRN_TYPE_MAX,
S
Shengliang Guan 已提交
136 137
} ETrnType;

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

S
Shengliang Guan 已提交
140 141 142 143 144 145 146 147 148 149 150 151 152 153
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 已提交
154
typedef struct {
S
Shengliang Guan 已提交
155 156 157
  int32_t    id;
  ETrnStage  stage;
  ETrnPolicy policy;
S
Shengliang Guan 已提交
158
  ETrnType   transType;
S
Shengliang Guan 已提交
159 160
  int32_t    code;
  int32_t    failedTimes;
L
Liu Jicong 已提交
161 162
  void*      rpcHandle;
  void*      rpcAHandle;
S
Shengliang Guan 已提交
163 164
  void*      rpcRsp;
  int32_t    rpcRspLen;
L
Liu Jicong 已提交
165 166 167 168 169
  SArray*    redoLogs;
  SArray*    undoLogs;
  SArray*    commitLogs;
  SArray*    redoActions;
  SArray*    undoActions;
S
Shengliang Guan 已提交
170 171 172
  int64_t    createdTime;
  int64_t    lastExecTime;
  uint64_t   dbUid;
S
Shengliang Guan 已提交
173
  char       dbname[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
174
  char       lastError[TSDB_TRANS_ERROR_LEN];
S
Shengliang Guan 已提交
175
} STrans;
S
Shengliang Guan 已提交
176

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

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

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

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

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

typedef struct {
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
L
Liu Jicong 已提交
228
  SDnodeObj* pDnode;
S
Shengliang Guan 已提交
229 230
} SBnodeObj;

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

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

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

S
Shengliang Guan 已提交
265
typedef struct {
S
Shengliang Guan 已提交
266
  char      user[TSDB_USER_LEN];
267
  char      pass[TSDB_PASSWORD_LEN];
S
Shengliang Guan 已提交
268 269 270
  char      acct[TSDB_USER_LEN];
  int64_t   createdTime;
  int64_t   updateTime;
271
  int8_t    superUser;
S
Shengliang Guan 已提交
272
  int32_t   acctId;
273 274
  SHashObj* readDbs;
  SHashObj* writeDbs;
S
Shengliang Guan 已提交
275 276 277
} SUserObj;

typedef struct {
278
  int32_t numOfVgroups;
S
Shengliang Guan 已提交
279 280 281 282 283 284
  int32_t cacheBlockSize;
  int32_t totalBlocks;
  int32_t daysPerFile;
  int32_t daysToKeep0;
  int32_t daysToKeep1;
  int32_t daysToKeep2;
S
Shengliang Guan 已提交
285 286
  int32_t minRows;
  int32_t maxRows;
S
Shengliang Guan 已提交
287 288
  int32_t commitTime;
  int32_t fsyncPeriod;
S
Shengliang Guan 已提交
289
  int8_t  walLevel;
S
Shengliang Guan 已提交
290 291 292 293 294 295 296 297
  int8_t  precision;
  int8_t  compression;
  int8_t  replications;
  int8_t  quorum;
  int8_t  update;
  int8_t  cacheLastRow;
} SDbCfg;

S
Shengliang Guan 已提交
298
typedef struct {
L
Liu Jicong 已提交
299 300
  char     name[TSDB_DB_FNAME_LEN];
  char     acct[TSDB_USER_LEN];
S
Shengliang Guan 已提交
301
  char     createUser[TSDB_USER_LEN];
L
Liu Jicong 已提交
302 303
  int64_t  createdTime;
  int64_t  updateTime;
H
Haojun Liao 已提交
304
  uint64_t uid;
L
Liu Jicong 已提交
305 306 307 308
  int32_t  cfgVersion;
  int32_t  vgVersion;
  int8_t   hashMethod;  // default is 1
  SDbCfg   cfg;
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];
337
  char     db[TSDB_DB_FNAME_LEN];
338 339
  int64_t  createdTime;
  int64_t  updateTime;
S
Shengliang Guan 已提交
340
  uint64_t uid;
S
Shengliang Guan 已提交
341
  uint64_t dbUid;
S
Shengliang Guan 已提交
342
  int32_t  version;
S
Shengliang Guan 已提交
343
  int32_t  nextColId;
S
Shengliang Guan 已提交
344 345
  int32_t  numOfColumns;
  int32_t  numOfTags;
S
Shengliang Guan 已提交
346
  SSchema* pColumns;
S
Shengliang Guan 已提交
347
  SSchema* pTags;
S
Shengliang Guan 已提交
348
  SRWLatch lock;
S
Shengliang Guan 已提交
349
  char     comment[TSDB_STB_COMMENT_LEN];
S
Shengliang Guan 已提交
350
} SStbObj;
S
Shengliang Guan 已提交
351

S
Shengliang Guan 已提交
352
typedef struct {
S
Shengliang Guan 已提交
353 354
  char    name[TSDB_FUNC_NAME_LEN];
  int64_t createdTime;
S
Shengliang Guan 已提交
355 356 357 358 359 360
  int8_t  funcType;
  int8_t  scriptType;
  int8_t  align;
  int8_t  outputType;
  int32_t outputLen;
  int32_t bufSize;
S
Shengliang 已提交
361
  int64_t signature;
S
Shengliang Guan 已提交
362 363
  int32_t commentSize;
  int32_t codeSize;
L
Liu Jicong 已提交
364 365
  char*   pComment;
  char*   pCode;
S
Shengliang Guan 已提交
366
  char    pData[];
S
Shengliang Guan 已提交
367 368
} SFuncObj;

S
Shengliang Guan 已提交
369
typedef struct {
370
  int64_t id;
S
Shengliang Guan 已提交
371 372 373 374 375 376 377
  int8_t  type;
  int8_t  replica;
  int16_t numOfColumns;
  int32_t rowSize;
  int32_t numOfRows;
  int32_t numOfReads;
  int32_t payloadLen;
L
Liu Jicong 已提交
378 379
  void*   pIter;
  SMnode* pMnode;
380
  char    db[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
381 382 383
  int16_t offset[TSDB_MAX_COLUMNS];
  int32_t bytes[TSDB_MAX_COLUMNS];
  char    payload[];
S
Shengliang Guan 已提交
384 385
} SShowObj;

L
Liu Jicong 已提交
386 387 388 389 390 391 392
typedef struct {
  int32_t vgId;  // -1 for unassigned
  int32_t status;
  SEpSet  epSet;
  int64_t oldConsumerId;
  int64_t consumerId;  // -1 for unassigned
  char*   qmsg;
L
Liu Jicong 已提交
393 394
} SMqConsumerEp;

L
Liu Jicong 已提交
395
static FORCE_INLINE int32_t tEncodeSMqConsumerEp(void** buf, const SMqConsumerEp* pConsumerEp) {
L
Liu Jicong 已提交
396 397
  int32_t tlen = 0;
  tlen += taosEncodeFixedI32(buf, pConsumerEp->vgId);
L
Liu Jicong 已提交
398
  tlen += taosEncodeFixedI32(buf, pConsumerEp->status);
399
  tlen += taosEncodeSEpSet(buf, &pConsumerEp->epSet);
L
Liu Jicong 已提交
400
  tlen += taosEncodeFixedI64(buf, pConsumerEp->oldConsumerId);
L
Liu Jicong 已提交
401
  tlen += taosEncodeFixedI64(buf, pConsumerEp->consumerId);
L
Liu Jicong 已提交
402
  tlen += taosEncodeString(buf, pConsumerEp->qmsg);
L
Liu Jicong 已提交
403 404 405 406 407
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqConsumerEp(void** buf, SMqConsumerEp* pConsumerEp) {
  buf = taosDecodeFixedI32(buf, &pConsumerEp->vgId);
L
Liu Jicong 已提交
408
  buf = taosDecodeFixedI32(buf, &pConsumerEp->status);
409
  buf = taosDecodeSEpSet(buf, &pConsumerEp->epSet);
L
Liu Jicong 已提交
410
  buf = taosDecodeFixedI64(buf, &pConsumerEp->oldConsumerId);
L
Liu Jicong 已提交
411
  buf = taosDecodeFixedI64(buf, &pConsumerEp->consumerId);
L
Liu Jicong 已提交
412
  buf = taosDecodeString(buf, &pConsumerEp->qmsg);
L
Liu Jicong 已提交
413 414 415
  return buf;
}

L
Liu Jicong 已提交
416 417 418 419 420 421
static FORCE_INLINE void tDeleteSMqConsumerEp(SMqConsumerEp* pConsumerEp) {
  if (pConsumerEp) {
    tfree(pConsumerEp->qmsg);
  }
}

L
Liu Jicong 已提交
422 423 424 425
typedef struct {
  int64_t consumerId;
  SArray* vgInfo;  // SArray<SMqConsumerEp>
} SMqSubConsumer;
L
Liu Jicong 已提交
426

L
Liu Jicong 已提交
427 428 429 430
static FORCE_INLINE int32_t tEncodeSMqSubConsumer(void** buf, const SMqSubConsumer* pConsumer) {
  int32_t tlen = 0;
  tlen += taosEncodeFixedI64(buf, pConsumer->consumerId);
  int32_t sz = taosArrayGetSize(pConsumer->vgInfo);
L
Liu Jicong 已提交
431
  tlen += taosEncodeFixedI32(buf, sz);
L
Liu Jicong 已提交
432 433 434
  for (int32_t i = 0; i < sz; i++) {
    SMqConsumerEp* pCEp = taosArrayGet(pConsumer->vgInfo, i);
    tlen += tEncodeSMqConsumerEp(buf, pCEp);
L
Liu Jicong 已提交
435
  }
L
Liu Jicong 已提交
436 437
  return tlen;
}
L
Liu Jicong 已提交
438

L
Liu Jicong 已提交
439 440 441 442 443 444 445 446 447
static FORCE_INLINE void* tDecodeSMqSubConsumer(void** buf, SMqSubConsumer* pConsumer) {
  int32_t sz;
  buf = taosDecodeFixedI64(buf, &pConsumer->consumerId);
  buf = taosDecodeFixedI32(buf, &sz);
  pConsumer->vgInfo = taosArrayInit(sz, sizeof(SMqConsumerEp));
  for (int32_t i = 0; i < sz; i++) {
    SMqConsumerEp consumerEp;
    buf = tDecodeSMqConsumerEp(buf, &consumerEp);
    taosArrayPush(pConsumer->vgInfo, &consumerEp);
L
Liu Jicong 已提交
448
  }
L
Liu Jicong 已提交
449 450 451 452 453 454 455
  return buf;
}

static FORCE_INLINE void tDeleteSMqSubConsumer(SMqSubConsumer* pSubConsumer) {
  if (pSubConsumer->vgInfo) {
    taosArrayDestroyEx(pSubConsumer->vgInfo, (void (*)(void*))tDeleteSMqConsumerEp);
    pSubConsumer->vgInfo = NULL;
L
Liu Jicong 已提交
456
  }
L
Liu Jicong 已提交
457 458 459 460 461 462 463
}

typedef struct {
  char    key[TSDB_SUBSCRIBE_KEY_LEN];
  int32_t status;
  int32_t vgNum;
  SArray* consumers;     // SArray<SMqSubConsumer>
L
Liu Jicong 已提交
464
  SArray* lostConsumers; // SArray<SMqSubConsumer>
L
Liu Jicong 已提交
465 466 467 468 469 470
  SArray* unassignedVg;  // SArray<SMqConsumerEp>
} SMqSubscribeObj;

static FORCE_INLINE SMqSubscribeObj* tNewSubscribeObj() {
  SMqSubscribeObj* pSub = calloc(1, sizeof(SMqSubscribeObj));
  if (pSub == NULL) {
L
Liu Jicong 已提交
471 472
    return NULL;
  }
L
Liu Jicong 已提交
473

L
Liu Jicong 已提交
474 475 476
  pSub->consumers = taosArrayInit(0, sizeof(SMqSubConsumer));
  if (pSub->consumers == NULL) {
    goto _err;
L
Liu Jicong 已提交
477
  }
L
Liu Jicong 已提交
478 479 480 481 482 483

  pSub->lostConsumers = taosArrayInit(0, sizeof(SMqSubConsumer));
  if (pSub->lostConsumers == NULL) {
    goto _err;
  }

L
Liu Jicong 已提交
484
  pSub->unassignedVg = taosArrayInit(0, sizeof(SMqConsumerEp));
L
Liu Jicong 已提交
485
  if (pSub->unassignedVg == NULL) {
L
Liu Jicong 已提交
486
    goto _err;
L
Liu Jicong 已提交
487
  }
L
Liu Jicong 已提交
488 489 490 491 492

  pSub->key[0] = 0;
  pSub->vgNum = 0;
  pSub->status = 0;

L
Liu Jicong 已提交
493
  return pSub;
L
Liu Jicong 已提交
494 495 496

_err:
  tfree(pSub->consumers);
L
Liu Jicong 已提交
497 498
  tfree(pSub->lostConsumers);
  tfree(pSub->unassignedVg);
L
Liu Jicong 已提交
499 500
  tfree(pSub);
  return NULL;
L
Liu Jicong 已提交
501 502 503 504 505
}

static FORCE_INLINE int32_t tEncodeSubscribeObj(void** buf, const SMqSubscribeObj* pSub) {
  int32_t tlen = 0;
  tlen += taosEncodeString(buf, pSub->key);
L
Liu Jicong 已提交
506 507
  tlen += taosEncodeFixedI32(buf, pSub->vgNum);
  tlen += taosEncodeFixedI32(buf, pSub->status);
L
Liu Jicong 已提交
508 509
  int32_t sz;

L
Liu Jicong 已提交
510
  sz = taosArrayGetSize(pSub->consumers);
L
Liu Jicong 已提交
511 512
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
L
Liu Jicong 已提交
513 514
    SMqSubConsumer* pSubConsumer = taosArrayGet(pSub->consumers, i);
    tlen += tEncodeSMqSubConsumer(buf, pSubConsumer);
L
Liu Jicong 已提交
515 516
  }

L
Liu Jicong 已提交
517 518 519 520 521 522 523
  sz = taosArrayGetSize(pSub->lostConsumers);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    SMqSubConsumer* pSubConsumer = taosArrayGet(pSub->lostConsumers, i);
    tlen += tEncodeSMqSubConsumer(buf, pSubConsumer);
  }

L
Liu Jicong 已提交
524 525 526 527 528 529 530 531 532 533 534 535
  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);
L
Liu Jicong 已提交
536 537
  buf = taosDecodeFixedI32(buf, &pSub->vgNum);
  buf = taosDecodeFixedI32(buf, &pSub->status);
L
Liu Jicong 已提交
538 539 540 541

  int32_t sz;

  buf = taosDecodeFixedI32(buf, &sz);
L
Liu Jicong 已提交
542 543
  pSub->consumers = taosArrayInit(sz, sizeof(SMqSubConsumer));
  if (pSub->consumers == NULL) {
L
Liu Jicong 已提交
544 545 546
    return NULL;
  }
  for (int32_t i = 0; i < sz; i++) {
L
Liu Jicong 已提交
547 548 549
    SMqSubConsumer subConsumer = {0};
    buf = tDecodeSMqSubConsumer(buf, &subConsumer);
    taosArrayPush(pSub->consumers, &subConsumer);
L
Liu Jicong 已提交
550 551
  }

L
Liu Jicong 已提交
552 553 554 555 556 557 558 559 560 561 562
  buf = taosDecodeFixedI32(buf, &sz);
  pSub->lostConsumers = taosArrayInit(sz, sizeof(SMqSubConsumer));
  if (pSub->lostConsumers == NULL) {
    return NULL;
  }
  for (int32_t i = 0; i < sz; i++) {
    SMqSubConsumer subConsumer = {0};
    buf = tDecodeSMqSubConsumer(buf, &subConsumer);
    taosArrayPush(pSub->lostConsumers, &subConsumer);
  }

L
Liu Jicong 已提交
563 564 565 566 567 568
  buf = taosDecodeFixedI32(buf, &sz);
  pSub->unassignedVg = taosArrayInit(sz, sizeof(SMqConsumerEp));
  if (pSub->unassignedVg == NULL) {
    return NULL;
  }
  for (int32_t i = 0; i < sz; i++) {
L
Liu Jicong 已提交
569 570 571
    SMqConsumerEp consumerEp = {0};
    buf = tDecodeSMqConsumerEp(buf, &consumerEp);
    taosArrayPush(pSub->unassignedVg, &consumerEp);
L
Liu Jicong 已提交
572 573 574
  }
  return buf;
}
L
Liu Jicong 已提交
575

L
Liu Jicong 已提交
576
static FORCE_INLINE void tDeleteSMqSubscribeObj(SMqSubscribeObj* pSub) {
L
Liu Jicong 已提交
577 578 579 580
  if (pSub->consumers) {
    taosArrayDestroyEx(pSub->consumers, (void (*)(void*))tDeleteSMqSubConsumer);
    //taosArrayDestroy(pSub->consumers);
    pSub->consumers = NULL;
L
Liu Jicong 已提交
581
  }
L
Liu Jicong 已提交
582

L
Liu Jicong 已提交
583
  if (pSub->unassignedVg) {
L
Liu Jicong 已提交
584 585
    taosArrayDestroyEx(pSub->unassignedVg, (void (*)(void*))tDeleteSMqConsumerEp);
    //taosArrayDestroy(pSub->unassignedVg);
L
Liu Jicong 已提交
586 587
    pSub->unassignedVg = NULL;
  }
L
Liu Jicong 已提交
588 589
}

L
Liu Jicong 已提交
590
typedef struct {
L
Liu Jicong 已提交
591 592 593 594
  char     name[TSDB_TOPIC_FNAME_LEN];
  char     db[TSDB_DB_FNAME_LEN];
  int64_t  createTime;
  int64_t  updateTime;
L
Liu Jicong 已提交
595
  int64_t  uid;
L
Liu Jicong 已提交
596
  int64_t  dbUid;
L
Liu Jicong 已提交
597 598 599 600 601 602
  int32_t  version;
  SRWLatch lock;
  int32_t  sqlLen;
  char*    sql;
  char*    logicalPlan;
  char*    physicalPlan;
L
Liu Jicong 已提交
603 604
} SMqTopicObj;

L
Liu Jicong 已提交
605
typedef struct {
L
Liu Jicong 已提交
606
  int64_t  consumerId;
L
Liu Jicong 已提交
607
  int64_t  connId;
L
Liu Jicong 已提交
608
  SRWLatch lock;
L
Liu Jicong 已提交
609
  char     cgroup[TSDB_CONSUMER_GROUP_LEN];
L
Liu Jicong 已提交
610 611
  SArray*  currentTopics;  // SArray<char*>
  SArray*  recentRemovedTopics;   // SArray<char*>
L
Liu Jicong 已提交
612
  int32_t  epoch;
L
Liu Jicong 已提交
613
  // stat
L
Liu Jicong 已提交
614 615 616 617 618 619 620
  int64_t pollCnt;
  // status
  int32_t status;
  // heartbeat from the consumer reset hbStatus to 0
  // each checkConsumerAlive msg add hbStatus by 1
  // if checkConsumerAlive > CONSUMER_REBALANCE_CNT, mask to lost
  int32_t hbStatus;
L
Liu Jicong 已提交
621 622
} SMqConsumerObj;

L
Liu Jicong 已提交
623
static FORCE_INLINE int32_t tEncodeSMqConsumerObj(void** buf, const SMqConsumerObj* pConsumer) {
L
Liu Jicong 已提交
624
  int32_t sz;
L
Liu Jicong 已提交
625 626
  int32_t tlen = 0;
  tlen += taosEncodeFixedI64(buf, pConsumer->consumerId);
L
Liu Jicong 已提交
627
  tlen += taosEncodeFixedI64(buf, pConsumer->connId);
L
Liu Jicong 已提交
628
  tlen += taosEncodeFixedI32(buf, pConsumer->epoch);
L
Liu Jicong 已提交
629
  tlen += taosEncodeFixedI64(buf, pConsumer->pollCnt);
L
Liu Jicong 已提交
630
  tlen += taosEncodeFixedI32(buf, pConsumer->status);
L
Liu Jicong 已提交
631
  tlen += taosEncodeString(buf, pConsumer->cgroup);
L
Liu Jicong 已提交
632 633 634 635 636 637 638 639 640

  sz = taosArrayGetSize(pConsumer->currentTopics);
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
    char* topic = taosArrayGetP(pConsumer->currentTopics, i);
    tlen += taosEncodeString(buf, topic);
  }

  sz = taosArrayGetSize(pConsumer->recentRemovedTopics);
L
Liu Jicong 已提交
641 642
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
L
Liu Jicong 已提交
643
    char* topic = taosArrayGetP(pConsumer->recentRemovedTopics, i);
L
Liu Jicong 已提交
644
    tlen += taosEncodeString(buf, topic);
L
Liu Jicong 已提交
645 646 647 648 649
  }
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqConsumerObj(void* buf, SMqConsumerObj* pConsumer) {
L
Liu Jicong 已提交
650
  int32_t sz;
L
Liu Jicong 已提交
651
  buf = taosDecodeFixedI64(buf, &pConsumer->consumerId);
L
Liu Jicong 已提交
652
  buf = taosDecodeFixedI64(buf, &pConsumer->connId);
L
Liu Jicong 已提交
653
  buf = taosDecodeFixedI32(buf, &pConsumer->epoch);
L
Liu Jicong 已提交
654
  buf = taosDecodeFixedI64(buf, &pConsumer->pollCnt);
L
Liu Jicong 已提交
655
  buf = taosDecodeFixedI32(buf, &pConsumer->status);
L
Liu Jicong 已提交
656
  buf = taosDecodeStringTo(buf, pConsumer->cgroup);
L
Liu Jicong 已提交
657 658 659 660 661 662 663 664 665

  buf = taosDecodeFixedI32(buf, &sz);
  pConsumer->currentTopics = taosArrayInit(sz, sizeof(SMqConsumerObj));
  for (int32_t i = 0; i < sz; i++) {
    char* topic;
    buf = taosDecodeString(buf, &topic);
    taosArrayPush(pConsumer->currentTopics, &topic);
  }

L
Liu Jicong 已提交
666
  buf = taosDecodeFixedI32(buf, &sz);
L
Liu Jicong 已提交
667
  pConsumer->recentRemovedTopics = taosArrayInit(sz, sizeof(SMqConsumerObj));
L
Liu Jicong 已提交
668
  for (int32_t i = 0; i < sz; i++) {
L
Liu Jicong 已提交
669 670
    char* topic;
    buf = taosDecodeString(buf, &topic);
L
Liu Jicong 已提交
671
    taosArrayPush(pConsumer->recentRemovedTopics, &topic);
L
Liu Jicong 已提交
672 673 674 675
  }
  return buf;
}

S
Shengliang Guan 已提交
676
typedef struct SMnodeMsg {
S
Shengliang Guan 已提交
677
  char    user[TSDB_USER_LEN];
678
  char    db[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
679
  int32_t acctId;
L
Liu Jicong 已提交
680
  SMnode* pMnode;
S
Shengliang Guan 已提交
681 682 683
  int64_t createdTime;
  SRpcMsg rpcMsg;
  int32_t contLen;
L
Liu Jicong 已提交
684
  void*   pCont;
S
Shengliang Guan 已提交
685
} SMnodeMsg;
S
Shengliang Guan 已提交
686

S
Shengliang Guan 已提交
687 688 689
#ifdef __cplusplus
}
#endif
S
Shengliang Guan 已提交
690

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