mndDef.h 17.3 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
typedef enum {
38 39 40 41 42 43 44 45
  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 已提交
46 47

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

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

S
Shengliang Guan 已提交
67
typedef enum {
S
Shengliang Guan 已提交
68 69 70 71
  TRN_TYPE_BASIC_SCOPE = 1000,
  TRN_TYPE_CREATE_USER = 1001,
  TRN_TYPE_ALTER_USER = 1002,
  TRN_TYPE_DROP_USER = 1003,
S
Shengliang Guan 已提交
72 73
  TRN_TYPE_CREATE_FUNC = 1004,
  TRN_TYPE_DROP_FUNC = 1005,
S
Shengliang Guan 已提交
74 75 76 77 78 79 80 81 82 83 84 85
  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 已提交
86
  TRN_TYPE_COMMIT_OFFSET = 1018,
S
Shengliang Guan 已提交
87 88 89 90 91 92 93 94 95
  TRN_TYPE_BASIC_SCOPE_END,
  TRN_TYPE_GLOBAL_SCOPE = 2000,
  TRN_TYPE_CREATE_DNODE = 2001,
  TRN_TYPE_DROP_DNODE = 2002,
  TRN_TYPE_GLOBAL_SCOPE_END,
  TRN_TYPE_DB_SCOPE = 3000,
  TRN_TYPE_CREATE_DB = 3001,
  TRN_TYPE_ALTER_DB = 3002,
  TRN_TYPE_DROP_DB = 3003,
S
Shengliang Guan 已提交
96 97
  TRN_TYPE_SPLIT_VGROUP = 3004,
  TRN_TYPE_MERGE_VGROUP = 3015,
S
Shengliang Guan 已提交
98
  TRN_TYPE_DB_SCOPE_END,
S
Shengliang Guan 已提交
99 100 101 102 103
  TRN_TYPE_STB_SCOPE = 4000,
  TRN_TYPE_CREATE_STB = 4001,
  TRN_TYPE_ALTER_STB = 4002,
  TRN_TYPE_DROP_STB = 4003,
  TRN_TYPE_STB_SCOPE_END,
S
Shengliang Guan 已提交
104 105
} ETrnType;

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

S
Shengliang Guan 已提交
108 109 110 111 112 113 114 115 116 117 118 119 120 121
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 已提交
122
typedef struct {
S
Shengliang Guan 已提交
123 124 125
  int32_t    id;
  ETrnStage  stage;
  ETrnPolicy policy;
S
Shengliang Guan 已提交
126
  ETrnType   transType;
S
Shengliang Guan 已提交
127 128
  int32_t    code;
  int32_t    failedTimes;
L
Liu Jicong 已提交
129 130
  void*      rpcHandle;
  void*      rpcAHandle;
S
Shengliang Guan 已提交
131 132
  void*      rpcRsp;
  int32_t    rpcRspLen;
L
Liu Jicong 已提交
133 134 135 136 137
  SArray*    redoLogs;
  SArray*    undoLogs;
  SArray*    commitLogs;
  SArray*    redoActions;
  SArray*    undoActions;
S
Shengliang Guan 已提交
138 139
  int64_t    createdTime;
  int64_t    lastExecTime;
L
Liu Jicong 已提交
140
  int64_t    dbUid;
S
Shengliang Guan 已提交
141
  char       dbname[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
142
  char       lastError[TSDB_TRANS_ERROR_LEN];
S
Shengliang Guan 已提交
143
} STrans;
S
Shengliang Guan 已提交
144

S
Shengliang Guan 已提交
145
typedef struct {
146
  int64_t id;
S
Shengliang Guan 已提交
147
  char    name[TSDB_CLUSTER_ID_LEN];
S
Shengliang Guan 已提交
148 149 150 151
  int64_t createdTime;
  int64_t updateTime;
} SClusterObj;

S
Shengliang Guan 已提交
152
typedef struct {
S
Shengliang Guan 已提交
153 154 155 156
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
  int64_t    rebootTime;
157
  int64_t    lastAccessTime;
S
Shengliang Guan 已提交
158
  int32_t    accessTimes;
S
Shengliang Guan 已提交
159
  int32_t    numOfVnodes;
S
Shengliang Guan 已提交
160 161
  int32_t    numOfSupportVnodes;
  int32_t    numOfCores;
S
Shengliang Guan 已提交
162 163 164 165
  EDndReason offlineReason;
  uint16_t   port;
  char       fqdn[TSDB_FQDN_LEN];
  char       ep[TSDB_EP_LEN];
S
Shengliang Guan 已提交
166 167
} SDnodeObj;

S
Shengliang Guan 已提交
168
typedef struct {
S
Shengliang Guan 已提交
169
  int32_t    id;
S
Shengliang Guan 已提交
170 171
  int64_t    createdTime;
  int64_t    updateTime;
S
Shengliang Guan 已提交
172
  ESyncState role;
S
Shengliang Guan 已提交
173 174
  int32_t    roleTerm;
  int64_t    roleTime;
L
Liu Jicong 已提交
175
  SDnodeObj* pDnode;
S
Shengliang Guan 已提交
176 177
} SMnodeObj;

S
Shengliang Guan 已提交
178 179 180 181
typedef struct {
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
L
Liu Jicong 已提交
182
  SDnodeObj* pDnode;
S
Shengliang Guan 已提交
183 184 185 186 187 188
} SQnodeObj;

typedef struct {
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
L
Liu Jicong 已提交
189
  SDnodeObj* pDnode;
S
Shengliang Guan 已提交
190 191 192 193 194 195
} SSnodeObj;

typedef struct {
  int32_t    id;
  int64_t    createdTime;
  int64_t    updateTime;
L
Liu Jicong 已提交
196
  SDnodeObj* pDnode;
S
Shengliang Guan 已提交
197 198
} SBnodeObj;

S
Shengliang Guan 已提交
199 200 201
typedef struct {
  int32_t maxUsers;
  int32_t maxDbs;
S
Shengliang Guan 已提交
202 203
  int32_t maxStbs;
  int32_t maxTbs;
S
Shengliang Guan 已提交
204 205
  int32_t maxTimeSeries;
  int32_t maxStreams;
S
Shengliang Guan 已提交
206 207 208 209
  int32_t maxFuncs;
  int32_t maxConsumers;
  int32_t maxConns;
  int32_t maxTopics;
S
Shengliang Guan 已提交
210 211
  int64_t maxStorage;   // In unit of GB
  int32_t accessState;  // Configured only by command
S
Shengliang Guan 已提交
212 213 214 215 216 217 218
} SAcctCfg;

typedef struct {
  int32_t numOfUsers;
  int32_t numOfDbs;
  int32_t numOfTimeSeries;
  int32_t numOfStreams;
S
Shengliang Guan 已提交
219 220
  int64_t totalStorage;  // Total storage wrtten from this account
  int64_t compStorage;   // Compressed storage on disk
S
Shengliang Guan 已提交
221 222
} SAcctInfo;

S
Shengliang Guan 已提交
223
typedef struct {
S
Shengliang Guan 已提交
224 225 226 227
  char      acct[TSDB_USER_LEN];
  int64_t   createdTime;
  int64_t   updateTime;
  int32_t   acctId;
S
Shengliang Guan 已提交
228
  int32_t   status;
S
Shengliang Guan 已提交
229 230 231 232
  SAcctCfg  cfg;
  SAcctInfo info;
} SAcctObj;

S
Shengliang Guan 已提交
233
typedef struct {
S
Shengliang Guan 已提交
234
  char      user[TSDB_USER_LEN];
235
  char      pass[TSDB_PASSWORD_LEN];
S
Shengliang Guan 已提交
236 237 238
  char      acct[TSDB_USER_LEN];
  int64_t   createdTime;
  int64_t   updateTime;
239
  int8_t    superUser;
S
Shengliang Guan 已提交
240
  int32_t   acctId;
241 242
  SHashObj* readDbs;
  SHashObj* writeDbs;
S
Shengliang Guan 已提交
243 244 245
} SUserObj;

typedef struct {
246
  int32_t numOfVgroups;
S
Shengliang Guan 已提交
247 248 249 250 251 252
  int32_t cacheBlockSize;
  int32_t totalBlocks;
  int32_t daysPerFile;
  int32_t daysToKeep0;
  int32_t daysToKeep1;
  int32_t daysToKeep2;
S
Shengliang Guan 已提交
253 254
  int32_t minRows;
  int32_t maxRows;
S
Shengliang Guan 已提交
255 256
  int32_t commitTime;
  int32_t fsyncPeriod;
S
Shengliang Guan 已提交
257
  int8_t  walLevel;
S
Shengliang Guan 已提交
258 259 260 261 262 263
  int8_t  precision;
  int8_t  compression;
  int8_t  replications;
  int8_t  quorum;
  int8_t  update;
  int8_t  cacheLastRow;
L
Liu Jicong 已提交
264
  int8_t  streamMode;
S
Shengliang Guan 已提交
265 266
} SDbCfg;

S
Shengliang Guan 已提交
267
typedef struct {
L
Liu Jicong 已提交
268 269 270 271 272 273 274 275 276 277
  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;
  int8_t  hashMethod;  // default is 1
  SDbCfg  cfg;
S
Shengliang Guan 已提交
278 279 280 281
} SDbObj;

typedef struct {
  int32_t    dnodeId;
S
Shengliang Guan 已提交
282
  ESyncState role;
S
Shengliang Guan 已提交
283 284
} SVnodeGid;

S
Shengliang Guan 已提交
285
typedef struct {
S
Shengliang Guan 已提交
286
  int32_t   vgId;
S
Shengliang Guan 已提交
287 288
  int64_t   createdTime;
  int64_t   updateTime;
S
Shengliang Guan 已提交
289
  int32_t   version;
S
Shengliang Guan 已提交
290 291
  uint32_t  hashBegin;
  uint32_t  hashEnd;
292
  char      dbName[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
293
  int64_t   dbUid;
S
Shengliang Guan 已提交
294 295
  int64_t   numOfTables;
  int64_t   numOfTimeSeries;
S
Shengliang Guan 已提交
296 297 298
  int64_t   totalStorage;
  int64_t   compStorage;
  int64_t   pointsWritten;
S
Shengliang Guan 已提交
299 300
  int8_t    compact;
  int8_t    replica;
L
Liu Jicong 已提交
301
  int8_t    streamMode;
S
Shengliang Guan 已提交
302
  SVnodeGid vnodeGid[TSDB_MAX_REPLICA];
S
Shengliang Guan 已提交
303 304
} SVgObj;

S
Shengliang Guan 已提交
305
typedef struct {
S
Shengliang Guan 已提交
306
  char     name[TSDB_TABLE_FNAME_LEN];
307
  char     db[TSDB_DB_FNAME_LEN];
308 309
  int64_t  createdTime;
  int64_t  updateTime;
L
Liu Jicong 已提交
310 311
  int64_t  uid;
  int64_t  dbUid;
S
Shengliang Guan 已提交
312
  int32_t  version;
S
Shengliang Guan 已提交
313
  int32_t  nextColId;
S
Shengliang Guan 已提交
314 315
  int32_t  numOfColumns;
  int32_t  numOfTags;
S
Shengliang Guan 已提交
316
  SSchema* pColumns;
S
Shengliang Guan 已提交
317
  SSchema* pTags;
S
Shengliang Guan 已提交
318
  SRWLatch lock;
S
Shengliang Guan 已提交
319
  char     comment[TSDB_STB_COMMENT_LEN];
S
Shengliang Guan 已提交
320
} SStbObj;
S
Shengliang Guan 已提交
321

S
Shengliang Guan 已提交
322
typedef struct {
S
Shengliang Guan 已提交
323 324
  char    name[TSDB_FUNC_NAME_LEN];
  int64_t createdTime;
S
Shengliang Guan 已提交
325 326 327 328 329 330
  int8_t  funcType;
  int8_t  scriptType;
  int8_t  align;
  int8_t  outputType;
  int32_t outputLen;
  int32_t bufSize;
S
Shengliang 已提交
331
  int64_t signature;
S
Shengliang Guan 已提交
332 333
  int32_t commentSize;
  int32_t codeSize;
L
Liu Jicong 已提交
334 335
  char*   pComment;
  char*   pCode;
S
Shengliang Guan 已提交
336
  char    pData[];
S
Shengliang Guan 已提交
337 338
} SFuncObj;

S
Shengliang Guan 已提交
339
typedef struct {
340
  int64_t id;
S
Shengliang Guan 已提交
341 342 343 344 345 346 347
  int8_t  type;
  int8_t  replica;
  int16_t numOfColumns;
  int32_t rowSize;
  int32_t numOfRows;
  int32_t numOfReads;
  int32_t payloadLen;
L
Liu Jicong 已提交
348 349
  void*   pIter;
  SMnode* pMnode;
350
  char    db[TSDB_DB_FNAME_LEN];
S
Shengliang Guan 已提交
351 352 353
  int16_t offset[TSDB_MAX_COLUMNS];
  int32_t bytes[TSDB_MAX_COLUMNS];
  char    payload[];
S
Shengliang Guan 已提交
354 355
} SShowObj;

L
Liu Jicong 已提交
356 357 358 359 360 361 362
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 已提交
363 364
} SMqConsumerEp;

L
Liu Jicong 已提交
365
static FORCE_INLINE int32_t tEncodeSMqConsumerEp(void** buf, const SMqConsumerEp* pConsumerEp) {
L
Liu Jicong 已提交
366 367
  int32_t tlen = 0;
  tlen += taosEncodeFixedI32(buf, pConsumerEp->vgId);
L
Liu Jicong 已提交
368
  tlen += taosEncodeFixedI32(buf, pConsumerEp->status);
369
  tlen += taosEncodeSEpSet(buf, &pConsumerEp->epSet);
L
Liu Jicong 已提交
370
  tlen += taosEncodeFixedI64(buf, pConsumerEp->oldConsumerId);
L
Liu Jicong 已提交
371
  tlen += taosEncodeFixedI64(buf, pConsumerEp->consumerId);
L
Liu Jicong 已提交
372
  tlen += taosEncodeString(buf, pConsumerEp->qmsg);
L
Liu Jicong 已提交
373 374 375 376 377
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqConsumerEp(void** buf, SMqConsumerEp* pConsumerEp) {
  buf = taosDecodeFixedI32(buf, &pConsumerEp->vgId);
L
Liu Jicong 已提交
378
  buf = taosDecodeFixedI32(buf, &pConsumerEp->status);
379
  buf = taosDecodeSEpSet(buf, &pConsumerEp->epSet);
L
Liu Jicong 已提交
380
  buf = taosDecodeFixedI64(buf, &pConsumerEp->oldConsumerId);
L
Liu Jicong 已提交
381
  buf = taosDecodeFixedI64(buf, &pConsumerEp->consumerId);
L
Liu Jicong 已提交
382
  buf = taosDecodeString(buf, &pConsumerEp->qmsg);
L
Liu Jicong 已提交
383 384 385
  return buf;
}

L
Liu Jicong 已提交
386 387 388 389 390 391
static FORCE_INLINE void tDeleteSMqConsumerEp(SMqConsumerEp* pConsumerEp) {
  if (pConsumerEp) {
    tfree(pConsumerEp->qmsg);
  }
}

L
Liu Jicong 已提交
392 393 394 395
typedef struct {
  int64_t consumerId;
  SArray* vgInfo;  // SArray<SMqConsumerEp>
} SMqSubConsumer;
L
Liu Jicong 已提交
396

L
Liu Jicong 已提交
397 398 399 400
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 已提交
401
  tlen += taosEncodeFixedI32(buf, sz);
L
Liu Jicong 已提交
402 403 404
  for (int32_t i = 0; i < sz; i++) {
    SMqConsumerEp* pCEp = taosArrayGet(pConsumer->vgInfo, i);
    tlen += tEncodeSMqConsumerEp(buf, pCEp);
L
Liu Jicong 已提交
405
  }
L
Liu Jicong 已提交
406 407
  return tlen;
}
L
Liu Jicong 已提交
408

L
Liu Jicong 已提交
409 410 411 412 413 414 415 416 417
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 已提交
418
  }
L
Liu Jicong 已提交
419 420 421 422 423 424 425
  return buf;
}

static FORCE_INLINE void tDeleteSMqSubConsumer(SMqSubConsumer* pSubConsumer) {
  if (pSubConsumer->vgInfo) {
    taosArrayDestroyEx(pSubConsumer->vgInfo, (void (*)(void*))tDeleteSMqConsumerEp);
    pSubConsumer->vgInfo = NULL;
L
Liu Jicong 已提交
426
  }
L
Liu Jicong 已提交
427 428
}

L
Liu Jicong 已提交
429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446
typedef struct {
  char    key[TSDB_PARTITION_KEY_LEN];
  int64_t offset;
} SMqOffsetObj;

static FORCE_INLINE int32_t tEncodeSMqOffsetObj(void** buf, const SMqOffsetObj* pOffset) {
  int32_t tlen = 0;
  tlen += taosEncodeString(buf, pOffset->key);
  tlen += taosEncodeFixedI64(buf, pOffset->offset);
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqOffsetObj(void* buf, SMqOffsetObj* pOffset) {
  buf = taosDecodeStringTo(buf, pOffset->key);
  buf = taosDecodeFixedI64(buf, &pOffset->offset);
  return buf;
}

L
Liu Jicong 已提交
447 448 449 450
typedef struct {
  char    key[TSDB_SUBSCRIBE_KEY_LEN];
  int32_t status;
  int32_t vgNum;
L
Liu Jicong 已提交
451 452 453
  SArray* consumers;      // SArray<SMqSubConsumer>
  SArray* lostConsumers;  // SArray<SMqSubConsumer>
  SArray* unassignedVg;   // SArray<SMqConsumerEp>
L
Liu Jicong 已提交
454 455 456 457 458
} SMqSubscribeObj;

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

L
Liu Jicong 已提交
462 463 464
  pSub->consumers = taosArrayInit(0, sizeof(SMqSubConsumer));
  if (pSub->consumers == NULL) {
    goto _err;
L
Liu Jicong 已提交
465
  }
L
Liu Jicong 已提交
466 467 468 469 470 471

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

L
Liu Jicong 已提交
472
  pSub->unassignedVg = taosArrayInit(0, sizeof(SMqConsumerEp));
L
Liu Jicong 已提交
473
  if (pSub->unassignedVg == NULL) {
L
Liu Jicong 已提交
474
    goto _err;
L
Liu Jicong 已提交
475
  }
L
Liu Jicong 已提交
476 477 478 479 480

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

L
Liu Jicong 已提交
481
  return pSub;
L
Liu Jicong 已提交
482 483 484

_err:
  tfree(pSub->consumers);
L
Liu Jicong 已提交
485 486
  tfree(pSub->lostConsumers);
  tfree(pSub->unassignedVg);
L
Liu Jicong 已提交
487 488
  tfree(pSub);
  return NULL;
L
Liu Jicong 已提交
489 490 491 492 493
}

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

L
Liu Jicong 已提交
498
  sz = taosArrayGetSize(pSub->consumers);
L
Liu Jicong 已提交
499 500
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
L
Liu Jicong 已提交
501 502
    SMqSubConsumer* pSubConsumer = taosArrayGet(pSub->consumers, i);
    tlen += tEncodeSMqSubConsumer(buf, pSubConsumer);
L
Liu Jicong 已提交
503 504
  }

L
Liu Jicong 已提交
505 506 507 508 509 510 511
  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 已提交
512 513 514 515 516 517 518 519 520 521 522 523
  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 已提交
524 525
  buf = taosDecodeFixedI32(buf, &pSub->vgNum);
  buf = taosDecodeFixedI32(buf, &pSub->status);
L
Liu Jicong 已提交
526 527 528 529

  int32_t sz;

  buf = taosDecodeFixedI32(buf, &sz);
L
Liu Jicong 已提交
530 531
  pSub->consumers = taosArrayInit(sz, sizeof(SMqSubConsumer));
  if (pSub->consumers == NULL) {
L
Liu Jicong 已提交
532 533 534
    return NULL;
  }
  for (int32_t i = 0; i < sz; i++) {
L
Liu Jicong 已提交
535 536 537
    SMqSubConsumer subConsumer = {0};
    buf = tDecodeSMqSubConsumer(buf, &subConsumer);
    taosArrayPush(pSub->consumers, &subConsumer);
L
Liu Jicong 已提交
538 539
  }

L
Liu Jicong 已提交
540 541 542 543 544 545 546 547 548 549 550
  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 已提交
551 552 553 554 555 556
  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 已提交
557 558 559
    SMqConsumerEp consumerEp = {0};
    buf = tDecodeSMqConsumerEp(buf, &consumerEp);
    taosArrayPush(pSub->unassignedVg, &consumerEp);
L
Liu Jicong 已提交
560 561 562
  }
  return buf;
}
L
Liu Jicong 已提交
563

L
Liu Jicong 已提交
564
static FORCE_INLINE void tDeleteSMqSubscribeObj(SMqSubscribeObj* pSub) {
L
Liu Jicong 已提交
565 566
  if (pSub->consumers) {
    taosArrayDestroyEx(pSub->consumers, (void (*)(void*))tDeleteSMqSubConsumer);
L
Liu Jicong 已提交
567
    // taosArrayDestroy(pSub->consumers);
L
Liu Jicong 已提交
568
    pSub->consumers = NULL;
L
Liu Jicong 已提交
569
  }
L
Liu Jicong 已提交
570

L
Liu Jicong 已提交
571
  if (pSub->unassignedVg) {
L
Liu Jicong 已提交
572
    taosArrayDestroyEx(pSub->unassignedVg, (void (*)(void*))tDeleteSMqConsumerEp);
L
Liu Jicong 已提交
573
    // taosArrayDestroy(pSub->unassignedVg);
L
Liu Jicong 已提交
574 575
    pSub->unassignedVg = NULL;
  }
L
Liu Jicong 已提交
576 577
}

L
Liu Jicong 已提交
578
typedef struct {
L
Liu Jicong 已提交
579 580 581 582
  char     name[TSDB_TOPIC_FNAME_LEN];
  char     db[TSDB_DB_FNAME_LEN];
  int64_t  createTime;
  int64_t  updateTime;
L
Liu Jicong 已提交
583
  int64_t  uid;
L
Liu Jicong 已提交
584
  int64_t  dbUid;
L
Liu Jicong 已提交
585 586 587 588 589 590
  int32_t  version;
  SRWLatch lock;
  int32_t  sqlLen;
  char*    sql;
  char*    logicalPlan;
  char*    physicalPlan;
L
Liu Jicong 已提交
591 592
} SMqTopicObj;

L
Liu Jicong 已提交
593
typedef struct {
L
Liu Jicong 已提交
594
  int64_t  consumerId;
L
Liu Jicong 已提交
595
  int64_t  connId;
L
Liu Jicong 已提交
596
  SRWLatch lock;
L
Liu Jicong 已提交
597
  char     cgroup[TSDB_CONSUMER_GROUP_LEN];
L
Liu Jicong 已提交
598 599
  SArray*  currentTopics;        // SArray<char*>
  SArray*  recentRemovedTopics;  // SArray<char*>
L
Liu Jicong 已提交
600
  int32_t  epoch;
L
Liu Jicong 已提交
601
  // stat
L
Liu Jicong 已提交
602 603 604 605 606 607 608
  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 已提交
609 610
} SMqConsumerObj;

L
Liu Jicong 已提交
611
static FORCE_INLINE int32_t tEncodeSMqConsumerObj(void** buf, const SMqConsumerObj* pConsumer) {
L
Liu Jicong 已提交
612
  int32_t sz;
L
Liu Jicong 已提交
613 614
  int32_t tlen = 0;
  tlen += taosEncodeFixedI64(buf, pConsumer->consumerId);
L
Liu Jicong 已提交
615
  tlen += taosEncodeFixedI64(buf, pConsumer->connId);
L
Liu Jicong 已提交
616
  tlen += taosEncodeFixedI32(buf, pConsumer->epoch);
L
Liu Jicong 已提交
617
  tlen += taosEncodeFixedI64(buf, pConsumer->pollCnt);
L
Liu Jicong 已提交
618
  tlen += taosEncodeFixedI32(buf, pConsumer->status);
L
Liu Jicong 已提交
619
  tlen += taosEncodeString(buf, pConsumer->cgroup);
L
Liu Jicong 已提交
620 621 622 623 624 625 626 627 628

  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 已提交
629 630
  tlen += taosEncodeFixedI32(buf, sz);
  for (int32_t i = 0; i < sz; i++) {
L
Liu Jicong 已提交
631
    char* topic = taosArrayGetP(pConsumer->recentRemovedTopics, i);
L
Liu Jicong 已提交
632
    tlen += taosEncodeString(buf, topic);
L
Liu Jicong 已提交
633 634 635 636 637
  }
  return tlen;
}

static FORCE_INLINE void* tDecodeSMqConsumerObj(void* buf, SMqConsumerObj* pConsumer) {
L
Liu Jicong 已提交
638
  int32_t sz;
L
Liu Jicong 已提交
639
  buf = taosDecodeFixedI64(buf, &pConsumer->consumerId);
L
Liu Jicong 已提交
640
  buf = taosDecodeFixedI64(buf, &pConsumer->connId);
L
Liu Jicong 已提交
641
  buf = taosDecodeFixedI32(buf, &pConsumer->epoch);
L
Liu Jicong 已提交
642
  buf = taosDecodeFixedI64(buf, &pConsumer->pollCnt);
L
Liu Jicong 已提交
643
  buf = taosDecodeFixedI32(buf, &pConsumer->status);
L
Liu Jicong 已提交
644
  buf = taosDecodeStringTo(buf, pConsumer->cgroup);
L
Liu Jicong 已提交
645 646 647 648 649 650 651 652 653

  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 已提交
654
  buf = taosDecodeFixedI32(buf, &sz);
L
Liu Jicong 已提交
655
  pConsumer->recentRemovedTopics = taosArrayInit(sz, sizeof(SMqConsumerObj));
L
Liu Jicong 已提交
656
  for (int32_t i = 0; i < sz; i++) {
L
Liu Jicong 已提交
657 658
    char* topic;
    buf = taosDecodeString(buf, &topic);
L
Liu Jicong 已提交
659
    taosArrayPush(pConsumer->recentRemovedTopics, &topic);
L
Liu Jicong 已提交
660 661 662 663
  }
  return buf;
}

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

S
Shengliang Guan 已提交
675 676 677
#ifdef __cplusplus
}
#endif
S
Shengliang Guan 已提交
678

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