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

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

D
dapan1121 已提交
83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103
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 已提交
104 105 106 107
typedef struct SCtgDbInfoCtx {
  char dbFName[TSDB_DB_FNAME_LEN];
} SCtgDbInfoCtx;

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

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

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

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

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

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

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

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

  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
} SCtgJob;

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

typedef struct SCtgTask {
  CTG_TASK_TYPE type;
D
dapan1121 已提交
203 204 205 206 207 208
  int32_t       taskId;
  SCtgJob*      pJob;
  void*         taskCtx;
  SCtgMsgCtx    msgCtx;
  int32_t       code;
  void*         res;
D
dapan1121 已提交
209 210 211 212 213 214 215 216 217 218 219 220
} 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 已提交
221 222
typedef struct SCtgApiStat {

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

D
dapan1121 已提交
227 228
} SCtgApiStat;

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

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

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

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

D
dapan1121 已提交
258 259 260 261 262 263 264 265 266 267 268 269
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;

D
dapan1121 已提交
270
typedef struct SCtgDropDBMsg {
D
dapan1121 已提交
271 272 273
  SCatalog* pCtg;
  char  dbFName[TSDB_DB_FNAME_LEN];
  uint64_t dbId;
D
dapan1121 已提交
274
} SCtgDropDBMsg;
D
dapan1121 已提交
275

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


typedef struct SCtgDropStbMetaMsg {
D
dapan1121 已提交
283 284 285 286 287
  SCatalog* pCtg;
  char  dbFName[TSDB_DB_FNAME_LEN];
  char  stbName[TSDB_TABLE_NAME_LEN];
  uint64_t dbId;
  uint64_t suid;
D
dapan1121 已提交
288
} SCtgDropStbMetaMsg;
D
dapan1121 已提交
289

D
dapan1121 已提交
290
typedef struct SCtgDropTblMetaMsg {
D
dapan1121 已提交
291 292 293 294
  SCatalog* pCtg;
  char  dbFName[TSDB_DB_FNAME_LEN];
  char  tbName[TSDB_TABLE_NAME_LEN];
  uint64_t dbId;
D
dapan1121 已提交
295
} SCtgDropTblMetaMsg;
D
dapan1121 已提交
296

D
dapan 已提交
297 298 299
typedef struct SCtgUpdateUserMsg {
  SCatalog* pCtg;
  SGetUserAuthRsp userAuth;
D
dapan 已提交
300
} SCtgUpdateUserMsg;
D
dapan 已提交
301

D
dapan1121 已提交
302 303 304 305 306 307
typedef struct SCtgUpdateEpsetMsg {
  SCatalog* pCtg;
  char  dbFName[TSDB_DB_FNAME_LEN];
  int32_t vgId;
  SEpSet  epSet;
} SCtgUpdateEpsetMsg;
D
dapan1121 已提交
308

D
dapan1121 已提交
309 310
typedef struct SCtgCacheOperation {
  int32_t  opId;
D
dapan1121 已提交
311
  void    *data;
D
dapan1121 已提交
312
  bool     syncOp;
D
dapan1121 已提交
313
  tsem_t   rspSem;  
D
dapan1121 已提交
314
} SCtgCacheOperation;
D
dapan1121 已提交
315 316

typedef struct SCtgQNode {
D
dapan1121 已提交
317
  SCtgCacheOperation    *op;
D
dapan1121 已提交
318 319 320
  struct SCtgQNode      *next;
} SCtgQNode;

D
dapan1121 已提交
321
typedef struct SCtgQueue {
D
dapan1121 已提交
322 323 324
  SRWLatch              qlock;
  SCtgQNode            *head;
  SCtgQNode            *tail;
D
dapan1121 已提交
325
  tsem_t                reqSem;  
D
dapan 已提交
326
  uint64_t              qRemainNum;
D
dapan1121 已提交
327 328 329 330
} SCtgQueue;

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

D
dapan1121 已提交
340
typedef uint32_t (*tableNameHashFp)(const char *, uint32_t);
D
dapan1121 已提交
341
typedef int32_t (*ctgOpFunc)(SCtgCacheOperation *);
D
dapan 已提交
342

D
dapan1121 已提交
343 344
typedef struct SCtgOperation {
  int32_t    opId;
D
dapan 已提交
345
  char       name[32];
D
dapan1121 已提交
346 347
  ctgOpFunc func;
} SCtgOperation;
D
dapan 已提交
348

D
dapan1121 已提交
349 350
#define CTG_QUEUE_INC() atomic_add_fetch_64(&gCtgMgmt.queue.qRemainNum, 1)
#define CTG_QUEUE_DEC() atomic_sub_fetch_64(&gCtgMgmt.queue.qRemainNum, 1)
D
dapan 已提交
351

D
dapan1121 已提交
352 353
#define CTG_STAT_INC(_item, _n) atomic_add_fetch_64(&(_item), _n)
#define CTG_STAT_DEC(_item, _n) atomic_sub_fetch_64(&(_item), _n)
D
dapan1121 已提交
354
#define CTG_STAT_GET(_item) atomic_load_64(&(_item))
D
dapan1121 已提交
355

D
dapan1121 已提交
356 357 358
#define CTG_RT_STAT_INC(item, n) (CTG_STAT_INC(gCtgMgmt.stat.runtime.item, n))
#define CTG_CACHE_STAT_INC(item, n) (CTG_STAT_INC(gCtgMgmt.stat.cache.item, n))
#define CTG_CACHE_STAT_DEC(item, n) (CTG_STAT_DEC(gCtgMgmt.stat.cache.item, n))
D
dapan 已提交
359

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

D
dapan 已提交
371 372
#define CTG_FLAG_SET(_flag, _v) ((_flag) |= (_v))

D
dapan1121 已提交
373 374 375
#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 已提交
376
#define CTG_FLAG_IS_SYS_DB(_flag) ((_flag) & CTG_FLAG_SYS_DB)
D
dapan1121 已提交
377
#define CTG_FLAG_IS_FORCE_UPDATE(_flag) ((_flag) & CTG_FLAG_FORCE_UPDATE)
D
dapan1121 已提交
378
#define CTG_FLAG_SET_SYS_DB(_flag) ((_flag) |= CTG_FLAG_SYS_DB)
D
dapan1121 已提交
379 380 381
#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 已提交
382

D
dapan1121 已提交
383
#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 已提交
384

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

D
dapan1121 已提交
387
#define CTG_TABLE_NOT_EXIST(code) (code == CTG_ERR_CODE_TABLE_NOT_EXIST) 
D
dapan1121 已提交
388
#define CTG_DB_NOT_EXIST(code) (code == TSDB_CODE_MND_DB_NOT_EXIST) 
D
dapan1121 已提交
389

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

D
dapan1121 已提交
397 398 399
#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)
400

D
dapan1121 已提交
401 402
#define TD_RWLATCH_WRITE_FLAG_COPY 0x40000000

D
dapan1121 已提交
403 404
#define CTG_IS_LOCKED(_lock) atomic_load_32((_lock))

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

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

D
dapan1121 已提交
437 438 439 440 441
  
#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 已提交
442
#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 已提交
443
#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 已提交
444

D
dapan1121 已提交
445 446 447 448 449 450 451 452 453 454
#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 已提交
455 456 457
int32_t ctgOpUpdateVgroup(SCtgCacheOperation *action);
int32_t ctgOpUpdateTbMeta(SCtgCacheOperation *action);
int32_t ctgOpDropDbCache(SCtgCacheOperation *action);
D
dapan1121 已提交
458
int32_t ctgOpDropDbVgroup(SCtgCacheOperation *action);
D
dapan1121 已提交
459 460 461 462
int32_t ctgOpDropStbMeta(SCtgCacheOperation *action);
int32_t ctgOpDropTbMeta(SCtgCacheOperation *action);
int32_t ctgOpUpdateUser(SCtgCacheOperation *action);
int32_t ctgOpUpdateEpset(SCtgCacheOperation *operation);
D
dapan1121 已提交
463 464 465 466 467 468
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 已提交
469
int32_t ctgReadTbVerFromCache(SCatalog *pCtg, const SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType, uint64_t *suid, char *stbName);
D
dapan1121 已提交
470
int32_t ctgChkAuthFromCache(SCatalog* pCtg, const char* user, const char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass);
D
dapan1121 已提交
471
int32_t ctgDropDbCacheEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId);
D
dapan1121 已提交
472
int32_t ctgDropDbVgroupEnqueue(SCatalog* pCtg, const char *dbFName, bool syncReq);
D
dapan1121 已提交
473 474 475 476 477 478
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 已提交
479 480 481 482 483 484 485 486 487 488 489 490 491 492
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);
D
dapan1121 已提交
493
int32_t ctgGetTbIndexFromMnode(CTG_PARAMS, const char *tbFName, SArray** out, SCtgTask* pTask);
D
dapan1121 已提交
494 495 496 497 498 499
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);

500
int32_t ctgInitJob(CTG_PARAMS, SCtgJob** job, uint64_t reqId, const SCatalogReq* pReq, catalogCallback fp, void* param, int32_t* taskNum);
D
dapan1121 已提交
501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518
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);
D
dapan1121 已提交
519
char *ctgTaskTypeStr(CTG_TASK_TYPE type);
D
dapan1121 已提交
520 521 522 523 524


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

H
Hongze Cheng 已提交
526 527 528 529
#ifdef __cplusplus
}
#endif

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