catalogInt.h 17.8 KB
Newer Older
H
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
/*
 * 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/>.
 */

#ifndef _TD_CATALOG_INT_H_
#define _TD_CATALOG_INT_H_

#ifdef __cplusplus
extern "C" {
#endif

H
Haojun Liao 已提交
23
#include "catalog.h"
S
common  
Shengliang Guan 已提交
24
#include "tcommon.h"
D
dapan1121 已提交
25
#include "query.h"
H
Haojun Liao 已提交
26

D
dapan1121 已提交
27 28 29
#define CTG_DEFAULT_CACHE_CLUSTER_NUMBER 6
#define CTG_DEFAULT_CACHE_VGROUP_NUMBER 100
#define CTG_DEFAULT_CACHE_DB_NUMBER 20
D
dapan1121 已提交
30
#define CTG_DEFAULT_CACHE_TBLMETA_NUMBER 1000
D
dapan1121 已提交
31 32
#define CTG_DEFAULT_RENT_SECOND 10
#define CTG_DEFAULT_RENT_SLOT_SIZE 10
D
dapan1121 已提交
33
#define CTG_DEFAULT_MAX_RETRY_TIMES 3
D
dapan1121 已提交
34

D
dapan1121 已提交
35
#define CTG_RENT_SLOT_SECOND 1.5
D
dapan1121 已提交
36 37

#define CTG_DEFAULT_INVALID_VERSION (-1)
D
dapan 已提交
38

D
dapan1121 已提交
39
#define CTG_ERR_CODE_TABLE_NOT_EXIST TSDB_CODE_PAR_TABLE_NOT_EXIST
D
dapan1121 已提交
40

D
dapan1121 已提交
41 42 43 44 45
enum {
  CTG_READ = 1,
  CTG_WRITE,
};

D
dapan1121 已提交
46 47 48 49 50
enum {
  CTG_RENT_DB = 1,
  CTG_RENT_STABLE,
};

D
dapan1121 已提交
51 52 53 54 55 56
enum {
  CTG_ACT_UPDATE_VG = 0,
  CTG_ACT_UPDATE_TBL,
  CTG_ACT_REMOVE_DB,
  CTG_ACT_REMOVE_STB,
  CTG_ACT_REMOVE_TBL,
D
dapan 已提交
57
  CTG_ACT_UPDATE_USER,
D
dapan1121 已提交
58 59 60
  CTG_ACT_MAX
};

D
dapan1121 已提交
61 62 63 64 65 66 67 68 69 70 71
typedef enum {
  CTG_TASK_GET_QNODE = 0,
  CTG_TASK_GET_DB_VGROUP,
  CTG_TASK_GET_DB_CFG,
  CTG_TASK_GET_TB_META,
  CTG_TASK_GET_TB_HASH,
  CTG_TASK_GET_INDEX,
  CTG_TASK_GET_UDF,
  CTG_TASK_GET_USER,
} CTG_TASK_TYPE;

D
dapan1121 已提交
72
typedef struct SCtgDebug {
D
dapan1121 已提交
73 74 75 76
  bool     lockEnable;
  bool     cacheEnable;
  bool     apiEnable;
  bool     metaEnable;
D
dapan1121 已提交
77
  uint32_t showCachePeriodSec;
D
dapan1121 已提交
78
} SCtgDebug;
79

D
dapan1121 已提交
80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116
typedef struct SCtgTbCacheInfo {
  bool     inCache;
  uint64_t dbId;
  uint64_t suid;
  int32_t  tbType;
} SCtgTbCacheInfo;

typedef struct SCtgTbMetaCtx {
  SCtgTbCacheInfo tbInfo;
  SName* pName;
  int32_t flag;
} SCtgTbMetaCtx;

typedef struct SCtgDbVgCtx {
  char dbFName[TSDB_DB_FNAME_LEN];
} SCtgDbVgCtx;

typedef struct SCtgDbCfgCtx {
  char dbFName[TSDB_DB_FNAME_LEN];
} SCtgDbCfgCtx;

typedef struct SCtgTbHashCtx {
  char dbFName[TSDB_DB_FNAME_LEN];
  SName* pName;
} SCtgTbHashCtx;

typedef struct SCtgIndexCtx {
  char indexFName[TSDB_INDEX_FNAME_LEN];
} SCtgIndexCtx;

typedef struct SCtgUdfCtx {
  char udfName[TSDB_FUNC_NAME_LEN];
} SCtgUdfCtx;

typedef struct SCtgUserCtx {
  SUserAuthInfo user;
} SCtgUserCtx;
117

D
dapan1121 已提交
118 119
typedef struct SCtgTbMetaCache {
  SRWLatch  stbLock;
D
dapan1121 已提交
120 121
  SRWLatch  metaLock;        // RC between cache destroy and all other operations
  SHashObj *metaCache;       //key:tbname, value:STableMeta
D
dapan1121 已提交
122 123
  SHashObj *stbCache;        //key:suid, value:STableMeta*
} SCtgTbMetaCache;
D
dapan1121 已提交
124

D
dapan1121 已提交
125 126
typedef struct SCtgDBCache {
  SRWLatch         vgLock;
D
dapan1121 已提交
127
  uint64_t         dbId;
D
dapan1121 已提交
128
  int8_t           deleted;
D
dapan1121 已提交
129
  SDBVgInfo       *vgInfo;  
D
dapan1121 已提交
130 131
  SCtgTbMetaCache  tbCache;
} SCtgDBCache;
D
dapan1121 已提交
132

D
dapan1121 已提交
133
typedef struct SCtgRentSlot {
D
dapan1121 已提交
134 135
  SRWLatch lock;
  bool     needSort;
D
dapan1121 已提交
136
  SArray  *meta;  // element is SDbVgVersion or SSTableMetaVersion
D
dapan1121 已提交
137
} SCtgRentSlot;
D
dapan1121 已提交
138

D
dapan1121 已提交
139
typedef struct SCtgRentMgmt {
D
dapan1121 已提交
140 141 142 143
  int8_t         type;
  uint16_t       slotNum;
  uint16_t       slotRIdx;
  int64_t        lastReadMsec;
D
dapan1121 已提交
144 145
  SCtgRentSlot  *slots;
} SCtgRentMgmt;
D
dapan1121 已提交
146

D
dapan 已提交
147 148 149 150 151 152 153 154 155
typedef struct SCtgUserAuth {
  int32_t   version;
  SRWLatch  lock;
  bool      superUser;
  SHashObj *createdDbs;
  SHashObj *readDbs;
  SHashObj *writeDbs;
} SCtgUserAuth;

D
dapan1121 已提交
156
typedef struct SCatalog {
D
dapan1121 已提交
157
  uint64_t         clusterId;  
D
dapan 已提交
158
  SHashObj        *userCache;    //key:user, value:SCtgUserAuth
D
dapan1121 已提交
159 160 161
  SHashObj        *dbCache;      //key:dbname, value:SCtgDBCache
  SCtgRentMgmt     dbRent;
  SCtgRentMgmt     stbRent;
H
Haojun Liao 已提交
162 163
} SCatalog;

D
dapan1121 已提交
164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212
typedef struct SCtgJob {
  int64_t          refId;
  SArray*          pTasks;
  int32_t          taskDone;
  SMetaData        jobRes;
  int32_t          rspCode;

  uint64_t         queryId;
  SCatalog*        pCtg; 
  void*            pTrans; 
  const SEpSet*    pMgmtEps;
  void*            userParam;
  catalogCallback  userFp;
  int32_t          tbMetaNum;
  int32_t          tbHashNum;
  int32_t          dbVgNum;
  int32_t          udfNum;
  int32_t          qnodeNum;
  int32_t          dbCfgNum;
  int32_t          indexNum;
  int32_t          userNum;
} SCtgJob;

typedef struct SCtgMsgCtx {
  int32_t reqType;
  void* lastOut;
  void* out;
  char* target;
} SCtgMsgCtx;

typedef struct SCtgTask {
  CTG_TASK_TYPE type;
  int32_t  taskId;
  SCtgJob *pJob;
  void* taskCtx;
  SCtgMsgCtx msgCtx;
  void* res;
} SCtgTask;

typedef int32_t (*ctgLanchTaskFp)(SCtgTask*);
typedef int32_t (*ctgHandleTaskMsgRspFp)(SCtgTask*, int32_t, const SDataBuf *, int32_t);
typedef int32_t (*ctgDumpTaskResFp)(SCtgTask*);

typedef struct SCtgAsyncFps {
  ctgLanchTaskFp launchFp;
  ctgHandleTaskMsgRspFp handleRspFp;
  ctgDumpTaskResFp dumpResFp;
} SCtgAsyncFps;

D
dapan1121 已提交
213 214
typedef struct SCtgApiStat {

wafwerar's avatar
wafwerar 已提交
215 216 217 218
#ifdef WINDOWS
  size_t avoidCompilationErrors;
#endif

D
dapan1121 已提交
219 220
} SCtgApiStat;

D
dapan1121 已提交
221
typedef struct SCtgRuntimeStat {
D
dapan 已提交
222 223
  uint64_t qNum;
  uint64_t qDoneNum;
D
dapan1121 已提交
224
} SCtgRuntimeStat;
D
dapan1121 已提交
225 226

typedef struct SCtgCacheStat {
D
dapan1121 已提交
227 228 229
  uint64_t clusterNum;
  uint64_t dbNum;
  uint64_t tblNum;
D
dapan1121 已提交
230
  uint64_t stblNum;
D
dapan1121 已提交
231
  uint64_t userNum;
D
dapan1121 已提交
232 233 234 235
  uint64_t vgHitNum;
  uint64_t vgMissNum;
  uint64_t tblHitNum;
  uint64_t tblMissNum;
D
dapan 已提交
236 237
  uint64_t userHitNum;
  uint64_t userMissNum;
D
dapan1121 已提交
238 239 240 241
} SCtgCacheStat;

typedef struct SCatalogStat {
  SCtgApiStat      api;
D
dapan1121 已提交
242
  SCtgRuntimeStat  runtime;
D
dapan1121 已提交
243 244 245
  SCtgCacheStat    cache;
} SCatalogStat;

D
dapan1121 已提交
246 247 248 249
typedef struct SCtgUpdateMsgHeader {
  SCatalog* pCtg;
} SCtgUpdateMsgHeader;

D
dapan1121 已提交
250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275
typedef struct SCtgUpdateVgMsg {
  SCatalog* pCtg;
  char  dbFName[TSDB_DB_FNAME_LEN];
  uint64_t dbId;
  SDBVgInfo* dbInfo;
} SCtgUpdateVgMsg;

typedef struct SCtgUpdateTblMsg {
  SCatalog* pCtg;
  STableMetaOutput* output;
} SCtgUpdateTblMsg;

typedef struct SCtgRemoveDBMsg {
  SCatalog* pCtg;
  char  dbFName[TSDB_DB_FNAME_LEN];
  uint64_t dbId;
} SCtgRemoveDBMsg;

typedef struct SCtgRemoveStbMsg {
  SCatalog* pCtg;
  char  dbFName[TSDB_DB_FNAME_LEN];
  char  stbName[TSDB_TABLE_NAME_LEN];
  uint64_t dbId;
  uint64_t suid;
} SCtgRemoveStbMsg;

D
dapan1121 已提交
276 277 278 279 280 281 282
typedef struct SCtgRemoveTblMsg {
  SCatalog* pCtg;
  char  dbFName[TSDB_DB_FNAME_LEN];
  char  tbName[TSDB_TABLE_NAME_LEN];
  uint64_t dbId;
} SCtgRemoveTblMsg;

D
dapan 已提交
283 284 285
typedef struct SCtgUpdateUserMsg {
  SCatalog* pCtg;
  SGetUserAuthRsp userAuth;
D
dapan 已提交
286
} SCtgUpdateUserMsg;
D
dapan 已提交
287

D
dapan1121 已提交
288

D
dapan1121 已提交
289
typedef struct SCtgMetaAction {
D
dapan1121 已提交
290 291 292 293
  int32_t  act;
  void    *data;
  bool     syncReq;
  uint64_t seqId;
D
dapan1121 已提交
294 295 296 297 298 299 300
} SCtgMetaAction;

typedef struct SCtgQNode {
  SCtgMetaAction         action;
  struct SCtgQNode      *next;
} SCtgQNode;

D
dapan1121 已提交
301
typedef struct SCtgQueue {
D
dapan1121 已提交
302
  SRWLatch              qlock;
D
dapan1121 已提交
303 304
  uint64_t              seqId;
  uint64_t              seqDone;
D
dapan1121 已提交
305 306
  SCtgQNode            *head;
  SCtgQNode            *tail;
D
dapan1121 已提交
307 308
  tsem_t                reqSem;  
  tsem_t                rspSem;  
D
dapan 已提交
309
  uint64_t              qRemainNum;
D
dapan1121 已提交
310 311 312 313
} SCtgQueue;

typedef struct SCatalogMgmt {
  bool                  exit;
D
dapan1121 已提交
314
  int32_t               jobPool;
D
dapan1121 已提交
315 316
  SRWLatch              lock;
  SCtgQueue             queue;
D
dapan1121 已提交
317
  TdThread              updateThread;  
D
dapan1121 已提交
318 319 320
  SHashObj             *pCluster;     //key: clusterId, value: SCatalog*
  SCatalogStat          stat;
  SCatalogCfg           cfg;
D
dapan1121 已提交
321 322
} SCatalogMgmt;

D
dapan1121 已提交
323
typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
D
dapan1121 已提交
324
typedef int32_t (*ctgActFunc)(SCtgMetaAction *);
D
dapan 已提交
325

D
dapan 已提交
326 327 328 329 330 331
typedef struct SCtgAction {
  int32_t    actId;
  char       name[32];
  ctgActFunc func;
} SCtgAction;

D
dapan1121 已提交
332 333
#define CTG_QUEUE_ADD() atomic_add_fetch_64(&gCtgMgmt.queue.qRemainNum, 1)
#define CTG_QUEUE_SUB() atomic_sub_fetch_64(&gCtgMgmt.queue.qRemainNum, 1)
D
dapan 已提交
334

D
dapan1121 已提交
335 336 337
#define CTG_STAT_ADD(_item, _n) atomic_add_fetch_64(&(_item), _n)
#define CTG_STAT_SUB(_item, _n) atomic_sub_fetch_64(&(_item), _n)
#define CTG_STAT_GET(_item) atomic_load_64(&(_item))
D
dapan1121 已提交
338

D
dapan1121 已提交
339 340 341
#define CTG_RUNTIME_STAT_ADD(item, n) (CTG_STAT_ADD(gCtgMgmt.stat.runtime.item, n))
#define CTG_CACHE_STAT_ADD(item, n) (CTG_STAT_ADD(gCtgMgmt.stat.cache.item, n))
#define CTG_CACHE_STAT_SUB(item, n) (CTG_STAT_SUB(gCtgMgmt.stat.cache.item, n))
D
dapan 已提交
342

D
dapan1121 已提交
343
#define CTG_IS_META_NULL(type) ((type) == META_TYPE_NULL_TABLE)
D
dapan1121 已提交
344 345 346 347
#define CTG_IS_META_CTABLE(type) ((type) == META_TYPE_CTABLE)
#define CTG_IS_META_TABLE(type) ((type) == META_TYPE_TABLE)
#define CTG_IS_META_BOTH(type) ((type) == META_TYPE_BOTH_TABLE)

D
dapan1121 已提交
348 349 350
#define CTG_FLAG_STB          0x1
#define CTG_FLAG_NOT_STB      0x2
#define CTG_FLAG_UNKNOWN_STB  0x4
D
dapan1121 已提交
351
#define CTG_FLAG_SYS_DB       0x8
D
dapan1121 已提交
352 353
#define CTG_FLAG_FORCE_UPDATE 0x10

D
dapan 已提交
354 355
#define CTG_FLAG_SET(_flag, _v) ((_flag) |= (_v))

D
dapan1121 已提交
356 357 358
#define CTG_FLAG_IS_STB(_flag) ((_flag) & CTG_FLAG_STB)
#define CTG_FLAG_IS_NOT_STB(_flag) ((_flag) & CTG_FLAG_NOT_STB)
#define CTG_FLAG_IS_UNKNOWN_STB(_flag) ((_flag) & CTG_FLAG_UNKNOWN_STB)
D
dapan1121 已提交
359
#define CTG_FLAG_IS_SYS_DB(_flag) ((_flag) & CTG_FLAG_SYS_DB)
D
dapan1121 已提交
360
#define CTG_FLAG_IS_FORCE_UPDATE(_flag) ((_flag) & CTG_FLAG_FORCE_UPDATE)
D
dapan1121 已提交
361
#define CTG_FLAG_SET_SYS_DB(_flag) ((_flag) |= CTG_FLAG_SYS_DB)
D
dapan1121 已提交
362 363 364
#define CTG_FLAG_SET_STB(_flag, tbType) do { (_flag) |= ((tbType) == TSDB_SUPER_TABLE) ? CTG_FLAG_STB : ((tbType) > TSDB_SUPER_TABLE ? CTG_FLAG_NOT_STB : CTG_FLAG_UNKNOWN_STB); } while (0)
#define CTG_FLAG_MAKE_STB(_isStb) (((_isStb) == 1) ? CTG_FLAG_STB : ((_isStb) == 0 ? CTG_FLAG_NOT_STB : CTG_FLAG_UNKNOWN_STB))
#define CTG_FLAG_MATCH_STB(_flag, tbType) (CTG_FLAG_IS_UNKNOWN_STB(_flag) || (CTG_FLAG_IS_STB(_flag) && (tbType) == TSDB_SUPER_TABLE) || (CTG_FLAG_IS_NOT_STB(_flag) && (tbType) != TSDB_SUPER_TABLE))
D
dapan1121 已提交
365

D
dapan1121 已提交
366
#define CTG_IS_SYS_DBNAME(_dbname) (((*(_dbname) == 'i') && (0 == strcmp(_dbname, TSDB_INFORMATION_SCHEMA_DB))) || ((*(_dbname) == 'p') && (0 == strcmp(_dbname, TSDB_PERFORMANCE_SCHEMA_DB))))
D
dapan1121 已提交
367

D
dapan1121 已提交
368 369
#define CTG_META_SIZE(pMeta) (sizeof(STableMeta) + ((pMeta)->tableInfo.numOfTags + (pMeta)->tableInfo.numOfColumns) * sizeof(SSchema))

D
dapan1121 已提交
370
#define CTG_TABLE_NOT_EXIST(code) (code == CTG_ERR_CODE_TABLE_NOT_EXIST) 
D
dapan1121 已提交
371
#define CTG_DB_NOT_EXIST(code) (code == TSDB_CODE_MND_DB_NOT_EXIST) 
D
dapan1121 已提交
372

D
dapan1121 已提交
373 374 375 376 377 378
#define ctgFatal(param, ...)  qFatal("CTG:%p " param, pCtg, __VA_ARGS__)
#define ctgError(param, ...)  qError("CTG:%p " param, pCtg, __VA_ARGS__)
#define ctgWarn(param, ...)   qWarn("CTG:%p " param, pCtg, __VA_ARGS__)
#define ctgInfo(param, ...)   qInfo("CTG:%p " param, pCtg, __VA_ARGS__)
#define ctgDebug(param, ...)  qDebug("CTG:%p " param, pCtg, __VA_ARGS__)
#define ctgTrace(param, ...)  qTrace("CTG:%p " param, pCtg, __VA_ARGS__)
D
dapan 已提交
379

D
dapan1121 已提交
380 381 382
#define CTG_LOCK_DEBUG(...) do { if (gCTGDebug.lockEnable) { qDebug(__VA_ARGS__); } } while (0)
#define CTG_CACHE_DEBUG(...) do { if (gCTGDebug.cacheEnable) { qDebug(__VA_ARGS__); } } while (0)
#define CTG_API_DEBUG(...) do { if (gCTGDebug.apiEnable) { qDebug(__VA_ARGS__); } } while (0)
383

D
dapan1121 已提交
384 385
#define TD_RWLATCH_WRITE_FLAG_COPY 0x40000000

D
dapan1121 已提交
386 387
#define CTG_IS_LOCKED(_lock) atomic_load_32((_lock))

D
dapan1121 已提交
388 389
#define CTG_LOCK(type, _lock) do {   \
  if (CTG_READ == (type)) {          \
D
dapan1121 已提交
390
    assert(atomic_load_32((_lock)) >= 0);  \
391
    CTG_LOCK_DEBUG("CTG RLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
D
dapan1121 已提交
392
    taosRLockLatch(_lock);           \
393
    CTG_LOCK_DEBUG("CTG RLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
D
dapan1121 已提交
394
    assert(atomic_load_32((_lock)) > 0);  \
D
dapan1121 已提交
395
  } else {                                                \
D
dapan1121 已提交
396
    assert(atomic_load_32((_lock)) >= 0);  \
397
    CTG_LOCK_DEBUG("CTG WLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__);  \
D
dapan1121 已提交
398
    taosWLockLatch(_lock);                                \
399
    CTG_LOCK_DEBUG("CTG WLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__);  \
D
dapan1121 已提交
400
    assert(atomic_load_32((_lock)) == TD_RWLATCH_WRITE_FLAG_COPY);  \
D
dapan1121 已提交
401 402 403 404 405
  }                                                       \
} while (0)

#define CTG_UNLOCK(type, _lock) do {                       \
  if (CTG_READ == (type)) {                                \
D
dapan1121 已提交
406
    assert(atomic_load_32((_lock)) > 0);  \
407
    CTG_LOCK_DEBUG("CTG RULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
D
dapan1121 已提交
408
    taosRUnLockLatch(_lock);                              \
409
    CTG_LOCK_DEBUG("CTG RULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
D
dapan1121 已提交
410
    assert(atomic_load_32((_lock)) >= 0);  \
D
dapan1121 已提交
411
  } else {                                                \
D
dapan1121 已提交
412
    assert(atomic_load_32((_lock)) == TD_RWLATCH_WRITE_FLAG_COPY);  \
413
    CTG_LOCK_DEBUG("CTG WULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
D
dapan1121 已提交
414
    taosWUnLockLatch(_lock);                              \
415
    CTG_LOCK_DEBUG("CTG WULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
D
dapan1121 已提交
416
    assert(atomic_load_32((_lock)) >= 0);  \
D
dapan1121 已提交
417 418 419
  }                                                       \
} while (0)

D
dapan1121 已提交
420 421 422 423 424
  
#define CTG_ERR_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; return _code; } } while (0)
#define CTG_RET(c) do { int32_t _code = c; if (_code != TSDB_CODE_SUCCESS) { terrno = _code; } return _code; } while (0)
#define CTG_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)

D
dapan 已提交
425
#define CTG_API_LEAVE(c) do { int32_t __code = c; CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock); CTG_API_DEBUG("CTG API leave %s", __FUNCTION__); CTG_RET(__code); } while (0)
wafwerar's avatar
wafwerar 已提交
426
#define CTG_API_ENTER() do { CTG_API_DEBUG("CTG API enter %s", __FUNCTION__); CTG_LOCK(CTG_READ, &gCtgMgmt.lock); if (atomic_load_8((int8_t*)&gCtgMgmt.exit)) { CTG_API_LEAVE(TSDB_CODE_CTG_OUT_OF_SERVICE); }  } while (0)
D
dapan1121 已提交
427

D
dapan1121 已提交
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 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501
#define CTG_PARAMS SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps
#define CTG_PARAMS_LIST() pCtg, pTrans, pMgmtEps

void ctgdShowTableMeta(SCatalog* pCtg, const char *tbName, STableMeta* p);
void ctgdShowClusterCache(SCatalog* pCtg);
int32_t ctgdShowCacheInfo(void);

int32_t ctgRemoveTbMetaFromCache(SCatalog* pCtg, SName* pTableName, bool syncReq);
int32_t ctgGetTbMetaFromCache(CTG_PARAMS, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta);

int32_t ctgActUpdateVg(SCtgMetaAction *action);
int32_t ctgActUpdateTb(SCtgMetaAction *action);
int32_t ctgActRemoveDB(SCtgMetaAction *action);
int32_t ctgActRemoveStb(SCtgMetaAction *action);
int32_t ctgActRemoveTb(SCtgMetaAction *action);
int32_t ctgActUpdateUser(SCtgMetaAction *action);
int32_t ctgAcquireVgInfoFromCache(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache);
void ctgReleaseDBCache(SCatalog *pCtg, SCtgDBCache *dbCache);
void ctgReleaseVgInfo(SCtgDBCache *dbCache);
int32_t ctgAcquireVgInfoFromCache(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache);
int32_t ctgTbMetaExistInCache(SCatalog* pCtg, char *dbFName, char* tbName, int32_t *exist);
int32_t ctgReadTbMetaFromCache(SCatalog* pCtg, SCtgTbMetaCtx* ctx, STableMeta** pTableMeta);
int32_t ctgReadTbSverFromCache(SCatalog* pCtg, const SName* pTableName, int32_t* sver);
int32_t ctgChkAuthFromCache(SCatalog* pCtg, const char* user, const char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass);
int32_t ctgPutRmDBToQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId);
int32_t ctgPutRmStbToQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid, bool syncReq);
int32_t ctgPutRmTbToQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *tbName, bool syncReq);
int32_t ctgPutUpdateVgToQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, SDBVgInfo* dbInfo, bool syncReq);
int32_t ctgPutUpdateTbToQueue(SCatalog* pCtg, STableMetaOutput *output, bool syncReq);
int32_t ctgPutUpdateUserToQueue(SCatalog* pCtg, SGetUserAuthRsp *pAuth, bool syncReq);
int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type);
int32_t ctgMetaRentAdd(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size);
int32_t ctgMetaRentGet(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size);
int32_t ctgUpdateTbMetaToCache(SCatalog* pCtg, STableMetaOutput* pOut, bool syncReq);
int32_t ctgStartUpdateThread();
int32_t ctgRelaunchGetTbMetaTask(SCtgTask *pTask);



int32_t ctgProcessRspMsg(void* out, int32_t reqType, char* msg, int32_t msgSize, int32_t rspCode, char* target);
int32_t ctgGetDBVgInfoFromMnode(CTG_PARAMS, SBuildUseDBInput *input, SUseDbOutput *out, SCtgTask* pTask);
int32_t ctgGetQnodeListFromMnode(CTG_PARAMS, SArray *out, SCtgTask* pTask);
int32_t ctgGetDBCfgFromMnode(CTG_PARAMS, const char *dbFName, SDbCfgInfo *out, SCtgTask* pTask);
int32_t ctgGetIndexInfoFromMnode(CTG_PARAMS, const char *indexName, SIndexInfo *out, SCtgTask* pTask);
int32_t ctgGetUdfInfoFromMnode(CTG_PARAMS, const char *funcName, SFuncInfo *out, SCtgTask* pTask);
int32_t ctgGetUserDbAuthFromMnode(CTG_PARAMS, const char *user, SGetUserAuthRsp *out, SCtgTask* pTask);
int32_t ctgGetTbMetaFromMnodeImpl(CTG_PARAMS, char *dbFName, char* tbName, STableMetaOutput* out, SCtgTask* pTask);
int32_t ctgGetTbMetaFromMnode(CTG_PARAMS, const SName* pTableName, STableMetaOutput* out, SCtgTask* pTask);
int32_t ctgGetTbMetaFromVnode(CTG_PARAMS, const SName* pTableName, SVgroupInfo *vgroupInfo, STableMetaOutput* out, SCtgTask* pTask);

int32_t ctgInitJob(CTG_PARAMS, SCtgJob** job, uint64_t reqId, const SCatalogReq* pReq, catalogCallback fp, void* param);
int32_t ctgLaunchJob(SCtgJob *pJob);
int32_t ctgMakeAsyncRes(SCtgJob *pJob);

int32_t ctgCloneVgInfo(SDBVgInfo *src, SDBVgInfo **dst);
int32_t ctgCloneMetaOutput(STableMetaOutput *output, STableMetaOutput **pOutput);
int32_t ctgGenerateVgList(SCatalog *pCtg, SHashObj *vgHash, SArray** pList);
void ctgFreeJob(void* job);
void ctgFreeHandle(SCatalog* pCtg);
void ctgFreeVgInfo(SDBVgInfo *vgInfo);
int32_t ctgGetVgInfoFromHashValue(SCatalog *pCtg, SDBVgInfo *dbInfo, const SName *pTableName, SVgroupInfo *pVgroup);
void ctgResetTbMetaTask(SCtgTask* pTask);
void ctgFreeDbCache(SCtgDBCache *dbCache);
int32_t ctgStbVersionSortCompare(const void* key1, const void* key2);
int32_t ctgDbVgVersionSortCompare(const void* key1, const void* key2);
int32_t ctgStbVersionSearchCompare(const void* key1, const void* key2);
int32_t ctgDbVgVersionSearchCompare(const void* key1, const void* key2);
void ctgFreeSTableMetaOutput(STableMetaOutput* pOutput);
int32_t ctgUpdateMsgCtx(SCtgMsgCtx* pCtx, int32_t reqType, void* out, char* target);


extern SCatalogMgmt gCtgMgmt;
extern SCtgDebug gCTGDebug;
extern SCtgAsyncFps gCtgAsyncFps[];
D
dapan1121 已提交
502

H
Hongze Cheng 已提交
503 504 505 506
#ifdef __cplusplus
}
#endif

D
dapan 已提交
507
#endif /*_TD_CATALOG_INT_H_*/