catalogInt.h 18.4 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
enum {
D
dapan1121 已提交
52 53 54 55 56 57 58 59
  CTG_OP_UPDATE_VGROUP = 0,
  CTG_OP_UPDATE_TB_META,
  CTG_OP_DROP_DB_CACHE,
  CTG_OP_DROP_STB_META,
  CTG_OP_DROP_TB_META,
  CTG_OP_UPDATE_USER,
  CTG_OP_UPDATE_VG_EPSET,
  CTG_OP_MAX
D
dapan1121 已提交
60 61
};

D
dapan1121 已提交
62 63 64 65
typedef enum {
  CTG_TASK_GET_QNODE = 0,
  CTG_TASK_GET_DB_VGROUP,
  CTG_TASK_GET_DB_CFG,
D
dapan1121 已提交
66
  CTG_TASK_GET_DB_INFO,
D
dapan1121 已提交
67 68 69 70 71 72 73
  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 已提交
74
typedef struct SCtgDebug {
D
dapan1121 已提交
75 76 77 78
  bool     lockEnable;
  bool     cacheEnable;
  bool     apiEnable;
  bool     metaEnable;
D
dapan1121 已提交
79
  uint32_t showCachePeriodSec;
D
dapan1121 已提交
80
} SCtgDebug;
81

D
dapan1121 已提交
82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102
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;

D
dapan1121 已提交
103 104 105 106
typedef struct SCtgDbInfoCtx {
  char dbFName[TSDB_DB_FNAME_LEN];
} SCtgDbInfoCtx;

D
dapan1121 已提交
107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122
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;
123

D
dapan1121 已提交
124 125
typedef struct SCtgTbMetaCache {
  SRWLatch  stbLock;
D
dapan1121 已提交
126 127
  SRWLatch  metaLock;        // RC between cache destroy and all other operations
  SHashObj *metaCache;       //key:tbname, value:STableMeta
D
dapan1121 已提交
128 129
  SHashObj *stbCache;        //key:suid, value:STableMeta*
} SCtgTbMetaCache;
D
dapan1121 已提交
130

D
dapan1121 已提交
131 132
typedef struct SCtgDBCache {
  SRWLatch         vgLock;
D
dapan1121 已提交
133
  uint64_t         dbId;
D
dapan1121 已提交
134
  int8_t           deleted;
D
dapan1121 已提交
135
  SDBVgInfo       *vgInfo;  
D
dapan1121 已提交
136 137
  SCtgTbMetaCache  tbCache;
} SCtgDBCache;
D
dapan1121 已提交
138

D
dapan1121 已提交
139
typedef struct SCtgRentSlot {
D
dapan1121 已提交
140 141
  SRWLatch lock;
  bool     needSort;
D
dapan1121 已提交
142
  SArray  *meta;  // element is SDbVgVersion or SSTableMetaVersion
D
dapan1121 已提交
143
} SCtgRentSlot;
D
dapan1121 已提交
144

D
dapan1121 已提交
145
typedef struct SCtgRentMgmt {
D
dapan1121 已提交
146 147 148 149
  int8_t         type;
  uint16_t       slotNum;
  uint16_t       slotRIdx;
  int64_t        lastReadMsec;
D
dapan1121 已提交
150 151
  SCtgRentSlot  *slots;
} SCtgRentMgmt;
D
dapan1121 已提交
152

D
dapan 已提交
153 154 155 156 157 158 159 160 161
typedef struct SCtgUserAuth {
  int32_t   version;
  SRWLatch  lock;
  bool      superUser;
  SHashObj *createdDbs;
  SHashObj *readDbs;
  SHashObj *writeDbs;
} SCtgUserAuth;

D
dapan1121 已提交
162
typedef struct SCatalog {
D
dapan1121 已提交
163
  uint64_t         clusterId;  
D
dapan 已提交
164
  SHashObj        *userCache;    //key:user, value:SCtgUserAuth
D
dapan1121 已提交
165 166 167
  SHashObj        *dbCache;      //key:dbname, value:SCtgDBCache
  SCtgRentMgmt     dbRent;
  SCtgRentMgmt     stbRent;
H
Haojun Liao 已提交
168 169
} SCatalog;

D
dapan1121 已提交
170 171 172 173 174 175 176 177 178 179
typedef struct SCtgJob {
  int64_t          refId;
  SArray*          pTasks;
  int32_t          taskDone;
  SMetaData        jobRes;
  int32_t          rspCode;

  uint64_t         queryId;
  SCatalog*        pCtg; 
  void*            pTrans; 
D
dapan1121 已提交
180
  SEpSet           pMgmtEps;
D
dapan1121 已提交
181 182 183 184 185 186 187 188 189 190
  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;
D
dapan1121 已提交
191
  int32_t          dbInfoNum;
D
dapan1121 已提交
192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219
} 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 已提交
220 221
typedef struct SCtgApiStat {

wafwerar's avatar
wafwerar 已提交
222 223 224 225
#ifdef WINDOWS
  size_t avoidCompilationErrors;
#endif

D
dapan1121 已提交
226 227
} SCtgApiStat;

D
dapan1121 已提交
228
typedef struct SCtgRuntimeStat {
D
dapan 已提交
229 230
  uint64_t qNum;
  uint64_t qDoneNum;
D
dapan1121 已提交
231
} SCtgRuntimeStat;
D
dapan1121 已提交
232 233

typedef struct SCtgCacheStat {
D
dapan1121 已提交
234 235 236
  uint64_t clusterNum;
  uint64_t dbNum;
  uint64_t tblNum;
D
dapan1121 已提交
237
  uint64_t stblNum;
D
dapan1121 已提交
238
  uint64_t userNum;
D
dapan1121 已提交
239 240 241 242
  uint64_t vgHitNum;
  uint64_t vgMissNum;
  uint64_t tblHitNum;
  uint64_t tblMissNum;
D
dapan 已提交
243 244
  uint64_t userHitNum;
  uint64_t userMissNum;
D
dapan1121 已提交
245 246 247 248
} SCtgCacheStat;

typedef struct SCatalogStat {
  SCtgApiStat      api;
D
dapan1121 已提交
249
  SCtgRuntimeStat  runtime;
D
dapan1121 已提交
250 251 252
  SCtgCacheStat    cache;
} SCatalogStat;

D
dapan1121 已提交
253 254 255 256
typedef struct SCtgUpdateMsgHeader {
  SCatalog* pCtg;
} SCtgUpdateMsgHeader;

D
dapan1121 已提交
257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282
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 已提交
283 284 285 286 287 288 289
typedef struct SCtgRemoveTblMsg {
  SCatalog* pCtg;
  char  dbFName[TSDB_DB_FNAME_LEN];
  char  tbName[TSDB_TABLE_NAME_LEN];
  uint64_t dbId;
} SCtgRemoveTblMsg;

D
dapan 已提交
290 291 292
typedef struct SCtgUpdateUserMsg {
  SCatalog* pCtg;
  SGetUserAuthRsp userAuth;
D
dapan 已提交
293
} SCtgUpdateUserMsg;
D
dapan 已提交
294

D
dapan1121 已提交
295 296 297 298 299 300
typedef struct SCtgUpdateEpsetMsg {
  SCatalog* pCtg;
  char  dbFName[TSDB_DB_FNAME_LEN];
  int32_t vgId;
  SEpSet  epSet;
} SCtgUpdateEpsetMsg;
D
dapan1121 已提交
301

D
dapan1121 已提交
302 303
typedef struct SCtgCacheOperation {
  int32_t  opId;
D
dapan1121 已提交
304 305 306
  void    *data;
  bool     syncReq;
  uint64_t seqId;
D
dapan1121 已提交
307
} SCtgCacheOperation;
D
dapan1121 已提交
308 309

typedef struct SCtgQNode {
D
dapan1121 已提交
310
  SCtgCacheOperation     op;
D
dapan1121 已提交
311 312 313
  struct SCtgQNode      *next;
} SCtgQNode;

D
dapan1121 已提交
314
typedef struct SCtgQueue {
D
dapan1121 已提交
315
  SRWLatch              qlock;
D
dapan1121 已提交
316 317
  uint64_t              seqId;
  uint64_t              seqDone;
D
dapan1121 已提交
318 319
  SCtgQNode            *head;
  SCtgQNode            *tail;
D
dapan1121 已提交
320 321
  tsem_t                reqSem;  
  tsem_t                rspSem;  
D
dapan 已提交
322
  uint64_t              qRemainNum;
D
dapan1121 已提交
323 324 325 326
} SCtgQueue;

typedef struct SCatalogMgmt {
  bool                  exit;
D
dapan1121 已提交
327
  int32_t               jobPool;
D
dapan1121 已提交
328 329
  SRWLatch              lock;
  SCtgQueue             queue;
D
dapan1121 已提交
330
  TdThread              updateThread;  
D
dapan1121 已提交
331 332 333
  SHashObj             *pCluster;     //key: clusterId, value: SCatalog*
  SCatalogStat          stat;
  SCatalogCfg           cfg;
D
dapan1121 已提交
334 335
} SCatalogMgmt;

D
dapan1121 已提交
336
typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
D
dapan1121 已提交
337
typedef int32_t (*ctgOpFunc)(SCtgCacheOperation *);
D
dapan 已提交
338

D
dapan1121 已提交
339 340
typedef struct SCtgOperation {
  int32_t    opId;
D
dapan 已提交
341
  char       name[32];
D
dapan1121 已提交
342 343
  ctgOpFunc func;
} SCtgOperation;
D
dapan 已提交
344

D
dapan1121 已提交
345 346
#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 已提交
347

D
dapan1121 已提交
348 349 350
#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 已提交
351

D
dapan1121 已提交
352 353 354
#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 已提交
355

D
dapan1121 已提交
356
#define CTG_IS_META_NULL(type) ((type) == META_TYPE_NULL_TABLE)
D
dapan1121 已提交
357 358 359 360
#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 已提交
361 362 363
#define CTG_FLAG_STB          0x1
#define CTG_FLAG_NOT_STB      0x2
#define CTG_FLAG_UNKNOWN_STB  0x4
D
dapan1121 已提交
364
#define CTG_FLAG_SYS_DB       0x8
D
dapan1121 已提交
365 366
#define CTG_FLAG_FORCE_UPDATE 0x10

D
dapan 已提交
367 368
#define CTG_FLAG_SET(_flag, _v) ((_flag) |= (_v))

D
dapan1121 已提交
369 370 371
#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 已提交
372
#define CTG_FLAG_IS_SYS_DB(_flag) ((_flag) & CTG_FLAG_SYS_DB)
D
dapan1121 已提交
373
#define CTG_FLAG_IS_FORCE_UPDATE(_flag) ((_flag) & CTG_FLAG_FORCE_UPDATE)
D
dapan1121 已提交
374
#define CTG_FLAG_SET_SYS_DB(_flag) ((_flag) |= CTG_FLAG_SYS_DB)
D
dapan1121 已提交
375 376 377
#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 已提交
378

D
dapan1121 已提交
379
#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 已提交
380

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

D
dapan1121 已提交
383
#define CTG_TABLE_NOT_EXIST(code) (code == CTG_ERR_CODE_TABLE_NOT_EXIST) 
D
dapan1121 已提交
384
#define CTG_DB_NOT_EXIST(code) (code == TSDB_CODE_MND_DB_NOT_EXIST) 
D
dapan1121 已提交
385

D
dapan1121 已提交
386 387 388 389 390 391
#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 已提交
392

D
dapan1121 已提交
393 394 395
#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)
396

D
dapan1121 已提交
397 398
#define TD_RWLATCH_WRITE_FLAG_COPY 0x40000000

D
dapan1121 已提交
399 400
#define CTG_IS_LOCKED(_lock) atomic_load_32((_lock))

D
dapan1121 已提交
401 402
#define CTG_LOCK(type, _lock) do {   \
  if (CTG_READ == (type)) {          \
D
dapan1121 已提交
403
    assert(atomic_load_32((_lock)) >= 0);  \
404
    CTG_LOCK_DEBUG("CTG RLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
D
dapan1121 已提交
405
    taosRLockLatch(_lock);           \
406
    CTG_LOCK_DEBUG("CTG RLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
D
dapan1121 已提交
407
    assert(atomic_load_32((_lock)) > 0);  \
D
dapan1121 已提交
408
  } else {                                                \
D
dapan1121 已提交
409
    assert(atomic_load_32((_lock)) >= 0);  \
410
    CTG_LOCK_DEBUG("CTG WLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__);  \
D
dapan1121 已提交
411
    taosWLockLatch(_lock);                                \
412
    CTG_LOCK_DEBUG("CTG WLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__);  \
D
dapan1121 已提交
413
    assert(atomic_load_32((_lock)) == TD_RWLATCH_WRITE_FLAG_COPY);  \
D
dapan1121 已提交
414 415 416 417 418
  }                                                       \
} while (0)

#define CTG_UNLOCK(type, _lock) do {                       \
  if (CTG_READ == (type)) {                                \
D
dapan1121 已提交
419
    assert(atomic_load_32((_lock)) > 0);  \
420
    CTG_LOCK_DEBUG("CTG RULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
D
dapan1121 已提交
421
    taosRUnLockLatch(_lock);                              \
422
    CTG_LOCK_DEBUG("CTG RULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
D
dapan1121 已提交
423
    assert(atomic_load_32((_lock)) >= 0);  \
D
dapan1121 已提交
424
  } else {                                                \
D
dapan1121 已提交
425
    assert(atomic_load_32((_lock)) == TD_RWLATCH_WRITE_FLAG_COPY);  \
426
    CTG_LOCK_DEBUG("CTG WULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
D
dapan1121 已提交
427
    taosWUnLockLatch(_lock);                              \
428
    CTG_LOCK_DEBUG("CTG WULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \
D
dapan1121 已提交
429
    assert(atomic_load_32((_lock)) >= 0);  \
D
dapan1121 已提交
430 431 432
  }                                                       \
} while (0)

D
dapan1121 已提交
433 434 435 436 437
  
#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 已提交
438
#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 已提交
439
#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 已提交
440

D
dapan1121 已提交
441 442 443 444 445 446 447 448 449 450
#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);

D
dapan1121 已提交
451 452 453 454 455 456 457
int32_t ctgOpUpdateVgroup(SCtgCacheOperation *action);
int32_t ctgOpUpdateTbMeta(SCtgCacheOperation *action);
int32_t ctgOpDropDbCache(SCtgCacheOperation *action);
int32_t ctgOpDropStbMeta(SCtgCacheOperation *action);
int32_t ctgOpDropTbMeta(SCtgCacheOperation *action);
int32_t ctgOpUpdateUser(SCtgCacheOperation *action);
int32_t ctgOpUpdateEpset(SCtgCacheOperation *operation);
D
dapan1121 已提交
458 459 460 461 462 463
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);
D
dapan1121 已提交
464
int32_t ctgReadTbVerFromCache(SCatalog *pCtg, const SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType, uint64_t *suid, char *stbName);
D
dapan1121 已提交
465
int32_t ctgChkAuthFromCache(SCatalog* pCtg, const char* user, const char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass);
D
dapan1121 已提交
466 467 468 469 470 471 472
int32_t ctgDropDbCacheEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId);
int32_t ctgDropStbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid, bool syncReq);
int32_t ctgDropTbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *tbName, bool syncReq);
int32_t ctgUpdateVgroupEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, SDBVgInfo* dbInfo, bool syncReq);
int32_t ctgUpdateTbMetaEnqueue(SCatalog* pCtg, STableMetaOutput *output, bool syncReq);
int32_t ctgUpdateUserEnqueue(SCatalog* pCtg, SGetUserAuthRsp *pAuth, bool syncReq);
int32_t ctgUpdateVgEpsetEnqueue(SCatalog* pCtg, char *dbFName, int32_t vgId, SEpSet* pEpSet);
D
dapan1121 已提交
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 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516
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 已提交
517

H
Hongze Cheng 已提交
518 519 520 521
#ifdef __cplusplus
}
#endif

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