catalog.c 42.9 KB
Newer Older
H
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13
/*
 * 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/>.
H
Haojun Liao 已提交
14 15
 */

D
dapan1121 已提交
16
#include "trpc.h"
D
dapan1121 已提交
17
#include "query.h"
D
dapan1121 已提交
18
#include "tname.h"
H
Haojun Liao 已提交
19
#include "catalogInt.h"
20

D
dapan1121 已提交
21 22
SCatalogMgmt ctgMgmt = {0};

23 24 25
SCTGDebug gCTGDebug = {0};


D
dapan1121 已提交
26
int32_t ctgGetDBVgroupFromCache(struct SCatalog* pCatalog, const char *dbName, SDBVgroupInfo **dbInfo, bool *inCache) {
D
dapan1121 已提交
27
  if (NULL == pCatalog->dbCache.cache) {
D
dapan1121 已提交
28
    *inCache = false;
D
dapan1121 已提交
29
    ctgWarn("empty db cache, dbName:%s", dbName);
D
dapan1121 已提交
30 31 32
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
33
  SDBVgroupInfo *info = NULL;
D
dapan1121 已提交
34

D
dapan1121 已提交
35 36
  while (true) {
    info = taosHashAcquire(pCatalog->dbCache.cache, dbName, strlen(dbName));
D
dapan1121 已提交
37

D
dapan1121 已提交
38 39
    if (NULL == info) {
      *inCache = false;
D
dapan1121 已提交
40
      ctgWarn("not in db vgroup cache, dbName:%s", dbName);
D
dapan1121 已提交
41 42 43 44 45 46 47 48 49 50 51 52 53
      return TSDB_CODE_SUCCESS;
    }

    CTG_LOCK(CTG_READ, &info->lock);
    if (NULL == info->vgInfo) {
      CTG_UNLOCK(CTG_READ, &info->lock);
      taosHashRelease(pCatalog->dbCache.cache, info);
      ctgWarn("db cache vgInfo is NULL, dbName:%s", dbName);
      
      continue;
    }

    break;
D
dapan1121 已提交
54
  }
D
dapan1121 已提交
55

D
dapan1121 已提交
56 57
  *dbInfo = info;
  *inCache = true;
D
dapan1121 已提交
58 59

  ctgDebug("Got db vgroup from cache, dbName:%s", dbName);
D
dapan1121 已提交
60 61 62 63 64 65 66 67 68 69 70
  
  return TSDB_CODE_SUCCESS;
}



int32_t ctgGetDBVgroupFromMnode(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, SBuildUseDBInput *input, SUseDbOutput *out) {
  char *msg = NULL;
  SEpSet *pVnodeEpSet = NULL;
  int32_t msgLen = 0;

D
dapan1121 已提交
71 72 73 74 75 76 77
  ctgDebug("try to get db vgroup from mnode, db:%s", input->db);

  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_MND_USE_DB)](input, &msg, 0, &msgLen);
  if (code) {
    ctgError("Build use db msg failed, code:%x, db:%s", code, input->db);
    CTG_ERR_RET(code);
  }
D
ut test  
dapan1121 已提交
78
  
D
dapan1121 已提交
79
  SRpcMsg rpcMsg = {
H
Hongze Cheng 已提交
80
      .msgType = TDMT_MND_USE_DB,
D
catalog  
dapan1121 已提交
81
      .pCont   = msg,
D
dapan1121 已提交
82 83 84 85 86 87
      .contLen = msgLen,
  };

  SRpcMsg rpcRsp = {0};

  rpcSendRecv(pRpc, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
D
dapan1121 已提交
88
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
D
dapan1121 已提交
89
    ctgError("error rsp for use db, code:%x, db:%s", rpcRsp.code, input->db);
D
dapan1121 已提交
90
    CTG_ERR_RET(rpcRsp.code);
D
dapan1121 已提交
91
  }
D
dapan1121 已提交
92

D
dapan1121 已提交
93 94 95 96 97
  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_USE_DB)](out, rpcRsp.pCont, rpcRsp.contLen);
  if (code) {
    ctgError("Process use db rsp failed, code:%x, db:%s", code, input->db);
    CTG_ERR_RET(code);
  }
D
dapan1121 已提交
98

D
dapan1121 已提交
99 100
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
101

D
dapan1121 已提交
102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124
int32_t ctgIsTableMetaExistInCache(struct SCatalog* pCatalog, const char* tbFullName, int32_t *exist) {
  if (NULL == pCatalog->tableCache.cache) {
    *exist = 0;
    ctgWarn("empty tablemeta cache, tbName:%s", tbFullName);
    return TSDB_CODE_SUCCESS;
  }

  size_t sz = 0;
  STableMeta *tbMeta = taosHashGet(pCatalog->tableCache.cache, tbFullName, strlen(tbFullName));

  if (NULL == tbMeta) {
    *exist = 0;
    ctgDebug("tablemeta not in cache, tbName:%s", tbFullName);
    return TSDB_CODE_SUCCESS;
  }

  *exist = 1;
  
  ctgDebug("tablemeta is in cache, tbName:%s", tbFullName);
  
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
125

H
Haojun Liao 已提交
126
int32_t ctgGetTableMetaFromCache(struct SCatalog* pCatalog, const SName* pTableName, STableMeta** pTableMeta, int32_t *exist) {
D
dapan1121 已提交
127 128
  if (NULL == pCatalog->tableCache.cache) {
    *exist = 0;
D
dapan1121 已提交
129
    ctgWarn("empty tablemeta cache, tbName:%s", pTableName->tname);
D
dapan1121 已提交
130 131 132
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
133
  char tbFullName[TSDB_TABLE_FNAME_LEN];
H
Haojun Liao 已提交
134
  tNameExtractFullName(pTableName, tbFullName);
D
dapan1121 已提交
135

D
dapan1121 已提交
136 137 138 139
  *pTableMeta = NULL;

  size_t sz = 0;
  STableMeta *tbMeta = taosHashGetCloneExt(pCatalog->tableCache.cache, tbFullName, strlen(tbFullName), NULL, (void **)pTableMeta, &sz);
D
dapan1121 已提交
140

D
dapan1121 已提交
141
  if (NULL == *pTableMeta) {
D
dapan1121 已提交
142
    *exist = 0;
D
dapan1121 已提交
143
    ctgDebug("tablemeta not in cache, tbName:%s", tbFullName);
D
dapan1121 已提交
144 145 146
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
147
  *exist = 1;
D
dapan1121 已提交
148 149
  
  tbMeta = *pTableMeta;
D
dapan1121 已提交
150

D
dapan1121 已提交
151
  if (tbMeta->tableType != TSDB_CHILD_TABLE) {
D
dapan1121 已提交
152 153
    ctgDebug("Got tablemeta from cache, tbName:%s", tbFullName);

D
dapan1121 已提交
154 155 156 157 158 159 160 161
    return TSDB_CODE_SUCCESS;
  }
  
  CTG_LOCK(CTG_READ, &pCatalog->tableCache.stableLock);
  
  STableMeta **stbMeta = taosHashGet(pCatalog->tableCache.stableCache, &tbMeta->suid, sizeof(tbMeta->suid));
  if (NULL == stbMeta || NULL == *stbMeta) {
    CTG_UNLOCK(CTG_READ, &pCatalog->tableCache.stableLock);
D
dapan1121 已提交
162
    ctgError("stable not in stableCache, suid:%"PRIx64, tbMeta->suid);
D
dapan1121 已提交
163 164 165 166
    tfree(*pTableMeta);
    *exist = 0;
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
167

D
dapan1121 已提交
168 169 170
  if ((*stbMeta)->suid != tbMeta->suid) {    
    CTG_UNLOCK(CTG_READ, &pCatalog->tableCache.stableLock);
    tfree(*pTableMeta);
D
dapan1121 已提交
171
    ctgError("stable suid in stableCache mis-match, expected suid:%"PRIx64 ",actual suid:%"PRIx64, tbMeta->suid, (*stbMeta)->suid);
D
dapan1121 已提交
172 173
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }
D
dapan1121 已提交
174

D
dapan1121 已提交
175 176 177 178
  int32_t metaSize = sizeof(STableMeta) + ((*stbMeta)->tableInfo.numOfTags + (*stbMeta)->tableInfo.numOfColumns) * sizeof(SSchema);
  *pTableMeta = realloc(*pTableMeta, metaSize);
  if (NULL == *pTableMeta) {    
    CTG_UNLOCK(CTG_READ, &pCatalog->tableCache.stableLock);
D
dapan1121 已提交
179
    ctgError("realloc size[%d] failed", metaSize);
D
dapan1121 已提交
180
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
181 182
  }

D
dapan1121 已提交
183 184 185
  memcpy(&(*pTableMeta)->sversion, &(*stbMeta)->sversion, metaSize - sizeof(SCTableMeta));

  CTG_UNLOCK(CTG_READ, &pCatalog->tableCache.stableLock);
D
dapan1121 已提交
186 187

  ctgDebug("Got tablemeta from cache, tbName:%s", tbFullName);
D
dapan1121 已提交
188 189 190 191
  
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
192 193
int32_t ctgGetTableTypeFromCache(struct SCatalog* pCatalog, const SName* pTableName, int32_t *tbType) {
  if (NULL == pCatalog->tableCache.cache) {
D
dapan1121 已提交
194
    ctgWarn("empty tablemeta cache, tbName:%s", pTableName->tname);  
D
dapan1121 已提交
195 196 197 198 199 200 201 202 203 204 205 206
    return TSDB_CODE_SUCCESS;
  }

  char tbFullName[TSDB_TABLE_FNAME_LEN];
  tNameExtractFullName(pTableName, tbFullName);

  size_t sz = 0;
  STableMeta *pTableMeta = NULL;
  
  taosHashGetCloneExt(pCatalog->tableCache.cache, tbFullName, strlen(tbFullName), NULL, (void **)&pTableMeta, &sz);

  if (NULL == pTableMeta) {
D
dapan1121 已提交
207 208
    ctgWarn("tablemeta not in cache, tbName:%s", tbFullName);  
  
D
dapan1121 已提交
209 210 211 212
    return TSDB_CODE_SUCCESS;
  }

  *tbType = pTableMeta->tableType;
D
dapan1121 已提交
213 214

  ctgDebug("Got tabletype from cache, tbName:%s, type:%d", tbFullName, *tbType);  
D
dapan1121 已提交
215 216 217 218 219
  
  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
220 221 222 223 224 225 226 227 228 229
void ctgGenEpSet(SEpSet *epSet, SVgroupInfo *vgroupInfo) {
  epSet->inUse = 0;
  epSet->numOfEps = vgroupInfo->numOfEps;

  for (int32_t i = 0; i < vgroupInfo->numOfEps; ++i) {
    memcpy(&epSet->port[i], &vgroupInfo->epAddr[i].port, sizeof(epSet->port[i]));
    memcpy(&epSet->fqdn[i], &vgroupInfo->epAddr[i].fqdn, sizeof(epSet->fqdn[i]));
  }
}

D
dapan1121 已提交
230
int32_t ctgGetTableMetaFromMnodeImpl(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, char* tbFullName, STableMetaOutput* output) {
D
dapan1121 已提交
231
  SBuildTableMetaInput bInput = {.vgId = 0, .dbName = NULL, .tableFullName = tbFullName};
D
dapan1121 已提交
232 233 234 235
  char *msg = NULL;
  SEpSet *pVnodeEpSet = NULL;
  int32_t msgLen = 0;

D
dapan1121 已提交
236 237 238 239 240 241 242
  ctgDebug("try to get table meta from mnode, tbName:%s", tbFullName);

  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_MND_STB_META)](&bInput, &msg, 0, &msgLen);
  if (code) {
    ctgError("Build mnode stablemeta msg failed, code:%x", code);
    CTG_ERR_RET(code);
  }
D
dapan1121 已提交
243 244 245 246 247 248

  SRpcMsg rpcMsg = {
      .msgType = TDMT_MND_STB_META,
      .pCont   = msg,
      .contLen = msgLen,
  };
D
dapan1121 已提交
249

D
dapan1121 已提交
250 251
  SRpcMsg rpcRsp = {0};

D
dapan1121 已提交
252
  rpcSendRecv(pTransporter, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
D
dapan1121 已提交
253 254
  
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
D
dapan1121 已提交
255
    if (CTG_TABLE_NOT_EXIST(rpcRsp.code)) {
D
dapan1121 已提交
256
      SET_META_TYPE_NONE(output->metaType);
D
dapan1121 已提交
257
      ctgDebug("stablemeta not exist in mnode, tbName:%s", tbFullName);
D
dapan1121 已提交
258 259 260
      return TSDB_CODE_SUCCESS;
    }
    
D
dapan1121 已提交
261
    ctgError("error rsp for stablemeta from mnode, code:%x, tbName:%s", rpcRsp.code, tbFullName);
D
dapan1121 已提交
262 263 264
    CTG_ERR_RET(rpcRsp.code);
  }

D
dapan1121 已提交
265 266 267 268 269 270 271
  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_STB_META)](output, rpcRsp.pCont, rpcRsp.contLen);
  if (code) {
    ctgError("Process mnode stablemeta rsp failed, code:%x, tbName:%s", code, tbFullName);
    CTG_ERR_RET(code);
  }

  ctgDebug("Got table meta from mnode, tbName:%s", tbFullName);
D
dapan1121 已提交
272 273 274 275

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
276
int32_t ctgGetTableMetaFromMnode(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMetaOutput* output) {
D
dapan1121 已提交
277 278 279
  char tbFullName[TSDB_TABLE_FNAME_LEN];
  tNameExtractFullName(pTableName, tbFullName);

D
dapan1121 已提交
280
  return ctgGetTableMetaFromMnodeImpl(pCatalog, pTransporter, pMgmtEps, tbFullName, output);
D
dapan1121 已提交
281
}
D
dapan1121 已提交
282

D
dapan1121 已提交
283 284
int32_t ctgGetTableMetaFromVnode(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, SVgroupInfo *vgroupInfo, STableMetaOutput* output) {
  if (NULL == pCatalog || NULL == pTransporter || NULL == pMgmtEps || NULL == pTableName || NULL == vgroupInfo || NULL == output) {
D
dapan1121 已提交
285
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
286 287
  }

D
dapan1121 已提交
288 289
  char dbFullName[TSDB_DB_FNAME_LEN];
  tNameGetFullDbName(pTableName, dbFullName);
D
dapan1121 已提交
290

291
  ctgDebug("try to get table meta from vnode, db:%s, tbName:%s", dbFullName, tNameGetTableName(pTableName));
D
dapan1121 已提交
292

293
  SBuildTableMetaInput bInput = {.vgId = vgroupInfo->vgId, .dbName = dbFullName, .tableFullName = (char *)tNameGetTableName(pTableName)};
D
dapan1121 已提交
294 295 296 297
  char *msg = NULL;
  SEpSet *pVnodeEpSet = NULL;
  int32_t msgLen = 0;

D
dapan1121 已提交
298 299
  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_VND_TABLE_META)](&bInput, &msg, 0, &msgLen);
  if (code) {
300
    ctgError("Build vnode tablemeta msg failed, code:%x, tbName:%s", code, tNameGetTableName(pTableName));
D
dapan1121 已提交
301 302
    CTG_ERR_RET(code);
  }
D
dapan1121 已提交
303 304

  SRpcMsg rpcMsg = {
H
Hongze Cheng 已提交
305
      .msgType = TDMT_VND_TABLE_META,
D
dapan1121 已提交
306 307 308 309 310 311 312 313
      .pCont   = msg,
      .contLen = msgLen,
  };

  SRpcMsg rpcRsp = {0};
  SEpSet  epSet;
  
  ctgGenEpSet(&epSet, vgroupInfo);
D
dapan1121 已提交
314
  rpcSendRecv(pTransporter, &epSet, &rpcMsg, &rpcRsp);
D
dapan1121 已提交
315
  
D
dapan1121 已提交
316
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
D
dapan1121 已提交
317
    if (CTG_TABLE_NOT_EXIST(rpcRsp.code)) {
D
dapan1121 已提交
318
      SET_META_TYPE_NONE(output->metaType);
319
      ctgDebug("tablemeta not exist in vnode, tbName:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
320 321 322
      return TSDB_CODE_SUCCESS;
    }
  
323
    ctgError("error rsp for table meta from vnode, code:%x, tbName:%s", rpcRsp.code, tNameGetTableName(pTableName));
D
dapan1121 已提交
324
    CTG_ERR_RET(rpcRsp.code);
D
dapan1121 已提交
325 326
  }

D
dapan1121 已提交
327 328
  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_VND_TABLE_META)](output, rpcRsp.pCont, rpcRsp.contLen);
  if (code) {
329
    ctgError("Process vnode tablemeta rsp failed, code:%x, tbName:%s", code, tNameGetTableName(pTableName));
D
dapan1121 已提交
330 331 332
    CTG_ERR_RET(code);
  }

333
  ctgDebug("Got table meta from vnode, db:%s, tbName:%s", dbFullName, tNameGetTableName(pTableName));
D
dapan1121 已提交
334 335 336 337 338

  return TSDB_CODE_SUCCESS;
}


339 340
int32_t ctgGetHashFunction(int8_t hashMethod, tableNameHashFp *fp) {
  switch (hashMethod) {
D
dapan1121 已提交
341 342 343 344 345 346 347 348
    default:
      *fp = MurmurHash3_32;
      break;
  }

  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
349
int32_t ctgGetVgInfoFromDB(struct SCatalog *pCatalog, void *pRpc, const SEpSet *pMgmtEps, SDBVgroupInfo *dbInfo, SArray** vgroupList) {
D
dapan1121 已提交
350
  SHashObj *vgroupHash = NULL;
351
  SVgroupInfo *vgInfo = NULL;
D
dapan1121 已提交
352 353
  SArray *vgList = NULL;
  int32_t code = 0;
D
dapan1121 已提交
354
  int32_t vgNum = taosHashGetSize(dbInfo->vgInfo);
355

D
dapan1121 已提交
356
  vgList = taosArrayInit(vgNum, sizeof(SVgroupInfo));
D
dapan1121 已提交
357
  if (NULL == vgList) {
D
dapan1121 已提交
358
    ctgError("taosArrayInit failed, num:%d", vgNum);
D
dapan 已提交
359 360 361
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);    
  }

362 363 364
  void *pIter = taosHashIterate(dbInfo->vgInfo, NULL);
  while (pIter) {
    vgInfo = pIter;
D
dapan1121 已提交
365

D
dapan1121 已提交
366
    if (NULL == taosArrayPush(vgList, vgInfo)) {
D
dapan1121 已提交
367
      ctgError("taosArrayPush failed, vgId:%d", vgInfo->vgId);
D
dapan1121 已提交
368
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
369 370 371 372
    }
    
    pIter = taosHashIterate(dbInfo->vgInfo, pIter);
    vgInfo = NULL;
D
dapan1121 已提交
373 374
  }

D
dapan1121 已提交
375 376 377
  *vgroupList = vgList;
  vgList = NULL;

D
dapan1121 已提交
378 379
  ctgDebug("Got vg list from DB, vgNum:%d", vgNum);

D
dapan1121 已提交
380
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
381 382 383 384 385 386 387 388

_return:

  if (vgList) {
    taosArrayDestroy(vgList);
  }

  CTG_RET(code);
D
dapan1121 已提交
389 390
}

D
dapan1121 已提交
391
int32_t ctgGetVgInfoFromHashValue(struct SCatalog *pCatalog, SDBVgroupInfo *dbInfo, const SName *pTableName, SVgroupInfo *pVgroup) {
D
dapan1121 已提交
392 393
  int32_t code = 0;
  
394
  int32_t vgNum = taosHashGetSize(dbInfo->vgInfo);
H
Haojun Liao 已提交
395 396 397
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);

398
  if (vgNum <= 0) {
D
dapan1121 已提交
399
    ctgError("db vgroup cache invalid, db:%s, vgroup number:%d", db, vgNum);
D
dapan1121 已提交
400
    CTG_ERR_RET(TSDB_CODE_TSC_DB_NOT_SELECTED);
D
dapan1121 已提交
401 402
  }

403 404
  tableNameHashFp fp = NULL;
  SVgroupInfo *vgInfo = NULL;
D
dapan1121 已提交
405

D
dapan1121 已提交
406
  CTG_ERR_JRET(ctgGetHashFunction(dbInfo->hashMethod, &fp));
407 408

  char tbFullName[TSDB_TABLE_FNAME_LEN];
H
Haojun Liao 已提交
409
  tNameExtractFullName(pTableName, tbFullName);
410 411 412 413 414 415 416

  uint32_t hashValue = (*fp)(tbFullName, (uint32_t)strlen(tbFullName));

  void *pIter = taosHashIterate(dbInfo->vgInfo, NULL);
  while (pIter) {
    vgInfo = pIter;
    if (hashValue >= vgInfo->hashBegin && hashValue <= vgInfo->hashEnd) {
417
      taosHashCancelIterate(dbInfo->vgInfo, pIter);
418
      break;
D
dapan1121 已提交
419
    }
420 421 422
    
    pIter = taosHashIterate(dbInfo->vgInfo, pIter);
    vgInfo = NULL;
D
dapan1121 已提交
423 424
  }

425
  if (NULL == vgInfo) {
426
    ctgError("no hash range found for hash value [%u], db:%s, numOfVgId:%d", hashValue, db, taosHashGetSize(dbInfo->vgInfo));
D
dapan1121 已提交
427
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
428 429 430 431
  }

  *pVgroup = *vgInfo;

D
dapan1121 已提交
432
_return:
433
  CTG_RET(code);
D
dapan1121 已提交
434 435
}

D
dapan1121 已提交
436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452
int32_t ctgSTableVersionCompare(const void* key1, const void* key2) {
  if (((SSTableMetaVersion*)key1)->suid < ((SSTableMetaVersion*)key2)->suid) {
    return -1;
  } else if (((SSTableMetaVersion*)key1)->suid > ((SSTableMetaVersion*)key2)->suid) {
    return 1;
  } else {
    return 0;
  }
}

int32_t ctgDbVgVersionCompare(const void* key1, const void* key2) {
  if (((SDbVgVersion*)key1)->dbId < ((SDbVgVersion*)key2)->dbId) {
    return -1;
  } else if (((SDbVgVersion*)key1)->dbId > ((SDbVgVersion*)key2)->dbId) {
    return 1;
  } else {
    return 0;
D
dapan1121 已提交
453
  }
D
dapan1121 已提交
454 455 456 457 458 459 460 461 462
}


int32_t ctgMetaRentInit(SMetaRentMgmt *mgmt, uint32_t rentSec, int8_t type) {
  mgmt->slotRIdx = 0;
  mgmt->slotNum = rentSec / CTG_RENT_SLOT_SECOND;
  mgmt->type = type;

  size_t msgSize = sizeof(SRentSlotInfo) * mgmt->slotNum;
D
dapan1121 已提交
463
  
D
dapan1121 已提交
464 465 466 467 468
  mgmt->slots = calloc(1, msgSize);
  if (NULL == mgmt->slots) {
    qError("calloc %d failed", (int32_t)msgSize);
    return TSDB_CODE_CTG_MEM_ERROR;
  }
D
dapan1121 已提交
469

D
dapan1121 已提交
470 471 472 473
  qDebug("meta rent initialized, type:%d, slotNum:%d", type, mgmt->slotNum);
  
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
474

D
dapan1121 已提交
475 476

int32_t ctgMetaRentAdd(SMetaRentMgmt *mgmt, void *meta, int64_t id, int32_t size) {
D
dapan1121 已提交
477
  int16_t widx = abs(id % mgmt->slotNum);
D
dapan1121 已提交
478 479 480 481 482 483 484 485 486 487

  SRentSlotInfo *slot = &mgmt->slots[widx];
  int32_t code = 0;
  
  CTG_LOCK(CTG_WRITE, &slot->lock);
  if (NULL == slot->meta) {
    slot->meta = taosArrayInit(CTG_DEFAULT_RENT_SLOT_SIZE, size);
    if (NULL == slot->meta) {
      qError("taosArrayInit %d failed, id:%"PRIx64", slot idx:%d, type:%d", CTG_DEFAULT_RENT_SLOT_SIZE, id, widx, mgmt->type);
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
488
    }
D
dapan1121 已提交
489
  }
D
dapan1121 已提交
490

D
dapan1121 已提交
491 492 493
  if (NULL == taosArrayPush(slot->meta, meta)) {
    qError("taosArrayPush meta to rent failed, id:%"PRIx64", slot idx:%d, type:%d", id, widx, mgmt->type);
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
494 495
  }

D
dapan1121 已提交
496
  slot->needSort = true;
D
dapan1121 已提交
497

D
dapan1121 已提交
498
  qDebug("add meta to rent, id:%"PRIx64", slot idx:%d, type:%d", id, widx, mgmt->type);
D
dapan1121 已提交
499

D
dapan1121 已提交
500 501 502 503 504 505 506
_return:

  CTG_UNLOCK(CTG_WRITE, &slot->lock);
  CTG_RET(code);
}

int32_t ctgMetaRentUpdate(SMetaRentMgmt *mgmt, void *meta, int64_t id, int32_t size, __compar_fn_t compare) {
D
dapan1121 已提交
507
  int16_t widx = abs(id % mgmt->slotNum);
D
dapan1121 已提交
508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524

  SRentSlotInfo *slot = &mgmt->slots[widx];
  int32_t code = 0;
  
  CTG_LOCK(CTG_WRITE, &slot->lock);
  if (NULL == slot->meta) {
    qError("meta in slot is empty, id:%"PRIx64", slot idx:%d, type:%d", id, widx, mgmt->type);
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
  }

  if (slot->needSort) {
    taosArraySort(slot->meta, compare);
    slot->needSort = false;
    qDebug("slot meta sorted, slot idx:%d, type:%d", widx, mgmt->type);
  }

  void *orig = taosArraySearch(slot->meta, &id, compare, TD_EQ);
D
dapan1121 已提交
525
  if (NULL == orig) {
D
dapan1121 已提交
526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550
    qError("meta not found in slot, id:%"PRIx64", slot idx:%d, type:%d", id, widx, mgmt->type);
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }

  memcpy(orig, meta, size);

  qDebug("meta in rent updated, id:%"PRIx64", slot idx:%d, type:%d", id, widx, mgmt->type);

_return:

  CTG_UNLOCK(CTG_WRITE, &slot->lock);

  if (code) {
    qWarn("meta in rent update failed, will try to add it, code:%x, id:%"PRIx64", slot idx:%d, type:%d", code, id, widx, mgmt->type);
    CTG_RET(ctgMetaRentAdd(mgmt, meta, id, size));
  }

  CTG_RET(code);
}

int32_t ctgMetaRentGetImpl(SMetaRentMgmt *mgmt, void **res, uint32_t *num, int32_t size) {
  int16_t ridx = atomic_add_fetch_16(&mgmt->slotRIdx, 1);
  if (ridx >= mgmt->slotNum) {
    ridx %= mgmt->slotNum;
    atomic_store_16(&mgmt->slotRIdx, ridx);
D
dapan1121 已提交
551
  }
D
dapan1121 已提交
552 553 554

  SRentSlotInfo *slot = &mgmt->slots[ridx];
  int32_t code = 0;
D
dapan1121 已提交
555
  
D
dapan1121 已提交
556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611
  CTG_LOCK(CTG_READ, &slot->lock);
  if (NULL == slot->meta) {
    qDebug("empty meta in slot:%d, type:%d", ridx, mgmt->type);
    *num = 0;
    goto _return;
  }

  size_t metaNum = taosArrayGetSize(slot->meta);
  if (metaNum <= 0) {
    qDebug("no meta in slot:%d, type:%d", ridx, mgmt->type);
    *num = 0;
    goto _return;
  }

  size_t msize = metaNum * size;
  *res = malloc(msize);
  if (NULL == *res) {
    qError("malloc %d failed", (int32_t)msize);
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
  }

  void *meta = taosArrayGet(slot->meta, 0);

  memcpy(*res, meta, msize);

  *num = (uint32_t)metaNum;

  qDebug("Got %d meta from rent, type:%d", (int32_t)metaNum, mgmt->type);

_return:

  CTG_UNLOCK(CTG_READ, &slot->lock);

  CTG_RET(code);
}

int32_t ctgMetaRentGet(SMetaRentMgmt *mgmt, void **res, uint32_t *num, int32_t size) {
  while (true) {
    int64_t msec = taosGetTimestampMs();
    int64_t lsec = atomic_load_64(&mgmt->lastReadMsec);
    if ((msec - lsec) < CTG_RENT_SLOT_SECOND * 1000) {
      *res = NULL;
      *num = 0;
      qDebug("too short time period to get expired meta, type:%d", mgmt->type);
      return TSDB_CODE_SUCCESS;
    }

    if (lsec != atomic_val_compare_exchange_64(&mgmt->lastReadMsec, lsec, msec)) {
      continue;
    }

    break;
  }

  CTG_ERR_RET(ctgMetaRentGetImpl(mgmt, res, num, size));

D
dapan1121 已提交
612 613 614 615
  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
616

D
dapan1121 已提交
617
int32_t ctgUpdateTableMetaCache(struct SCatalog *pCatalog, STableMetaOutput *output) {
D
dapan1121 已提交
618
  int32_t code = 0;
D
dapan1121 已提交
619 620

  if (NULL == output->tbMeta) {
D
dapan1121 已提交
621
    ctgError("no valid table meta got from meta rsp, tbName:%s", output->tbFname);
D
dapan1121 已提交
622
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
623 624 625
  }

  if (NULL == pCatalog->tableCache.cache) {
D
dapan1121 已提交
626 627
    SHashObj *cache = taosHashInit(ctgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
    if (NULL == cache) {
D
dapan1121 已提交
628
      ctgError("taosHashInit failed, num:%d", ctgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
629
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
630
    }
D
dapan1121 已提交
631 632 633 634

    if (NULL != atomic_val_compare_exchange_ptr(&pCatalog->tableCache.cache, NULL, cache)) {
      taosHashCleanup(cache);
    }
D
dapan1121 已提交
635
  }
D
dapan1121 已提交
636

D
dapan1121 已提交
637
  if (NULL == pCatalog->tableCache.stableCache) {
D
dapan1121 已提交
638 639
    SHashObj *cache = taosHashInit(ctgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), true, HASH_ENTRY_LOCK);
    if (NULL == cache) {
D
dapan1121 已提交
640
      ctgError("taosHashInit failed, num:%d", ctgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
641
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
642
    }
D
dapan1121 已提交
643 644 645 646

    if (NULL != atomic_val_compare_exchange_ptr(&pCatalog->tableCache.stableCache, NULL, cache)) {
      taosHashCleanup(cache);
    }
D
dapan1121 已提交
647 648
  }

D
dapan1121 已提交
649
  if (CTG_IS_META_CTABLE(output->metaType) || CTG_IS_META_BOTH(output->metaType)) {
D
dapan1121 已提交
650
    if (taosHashPut(pCatalog->tableCache.cache, output->ctbFname, strlen(output->ctbFname), &output->ctbMeta, sizeof(output->ctbMeta)) != 0) {
D
dapan1121 已提交
651
      ctgError("taosHashPut ctablemeta to cache failed, ctbName:%s", output->ctbFname);
D
dapan1121 已提交
652
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
653 654
    }

D
dapan1121 已提交
655 656
    ctgDebug("update child tablemeta to cache, tbName:%s", output->ctbFname);
  }
D
dapan1121 已提交
657

D
dapan1121 已提交
658 659
  if (CTG_IS_META_CTABLE(output->metaType)) {
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
660
  }
D
dapan1121 已提交
661 662 663 664 665
  
  if (CTG_IS_META_BOTH(output->metaType) && TSDB_SUPER_TABLE != output->tbMeta->tableType) {
    ctgError("table type error, expected:%d, actual:%d", TSDB_SUPER_TABLE, output->tbMeta->tableType);
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }    
D
dapan1121 已提交
666

D
dapan1121 已提交
667
  int32_t tbSize = sizeof(*output->tbMeta) + sizeof(SSchema) * (output->tbMeta->tableInfo.numOfColumns + output->tbMeta->tableInfo.numOfTags);
D
dapan1121 已提交
668 669

  if (TSDB_SUPER_TABLE == output->tbMeta->tableType) {
D
dapan1121 已提交
670 671 672
    bool newAdded = false;
    SSTableMetaVersion metaRent = {.suid = output->tbMeta->suid, .sversion = output->tbMeta->sversion, .tversion = output->tbMeta->tversion};
    
D
dapan1121 已提交
673 674 675
    CTG_LOCK(CTG_WRITE, &pCatalog->tableCache.stableLock);
    if (taosHashPut(pCatalog->tableCache.cache, output->tbFname, strlen(output->tbFname), output->tbMeta, tbSize) != 0) {
      CTG_UNLOCK(CTG_WRITE, &pCatalog->tableCache.stableLock);
D
dapan1121 已提交
676
      ctgError("taosHashPut tablemeta to cache failed, tbName:%s", output->tbFname);
D
dapan1121 已提交
677
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
678 679 680
    }

    STableMeta *tbMeta = taosHashGet(pCatalog->tableCache.cache, output->tbFname, strlen(output->tbFname));
D
dapan1121 已提交
681
    if (taosHashPutExt(pCatalog->tableCache.stableCache, &output->tbMeta->suid, sizeof(output->tbMeta->suid), &tbMeta, POINTER_BYTES, &newAdded) != 0) {
D
dapan1121 已提交
682
      CTG_UNLOCK(CTG_WRITE, &pCatalog->tableCache.stableLock);
D
dapan1121 已提交
683
      ctgError("taosHashPutExt stable to stable cache failed, suid:%"PRIx64, output->tbMeta->suid);
D
dapan1121 已提交
684
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
685 686
    }
    CTG_UNLOCK(CTG_WRITE, &pCatalog->tableCache.stableLock);
D
dapan1121 已提交
687 688 689 690 691 692 693 694

    ctgDebug("update stable to cache, suid:%"PRIx64, output->tbMeta->suid);

    if (newAdded) {
      CTG_ERR_RET(ctgMetaRentAdd(&pCatalog->stableRent, &metaRent, metaRent.suid, sizeof(SSTableMetaVersion)));
    } else {
      CTG_ERR_RET(ctgMetaRentUpdate(&pCatalog->stableRent, &metaRent, metaRent.suid, sizeof(SSTableMetaVersion), ctgSTableVersionCompare));
    }
D
dapan1121 已提交
695 696
  } else {
    if (taosHashPut(pCatalog->tableCache.cache, output->tbFname, strlen(output->tbFname), output->tbMeta, tbSize) != 0) {
D
dapan1121 已提交
697
      ctgError("taosHashPut tablemeta to cache failed, tbName:%s", output->tbFname);
D
dapan1121 已提交
698
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
699 700
    }
  }
D
dapan1121 已提交
701

D
dapan1121 已提交
702 703
  ctgDebug("update tablemeta to cache, tbName:%s", output->tbFname);

D
dapan1121 已提交
704
  CTG_RET(code);
D
dapan1121 已提交
705 706
}

707
int32_t ctgGetDBVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* dbName, bool forceUpdate, SDBVgroupInfo** dbInfo) {
D
dapan1121 已提交
708
  bool inCache = false;
709
  if (!forceUpdate) {
D
dapan1121 已提交
710 711
    CTG_ERR_RET(ctgGetDBVgroupFromCache(pCatalog, dbName, dbInfo, &inCache));
    if (inCache) {
D
dapan1121 已提交
712 713
      return TSDB_CODE_SUCCESS;
    }
714 715

    ctgDebug("failed to get DB vgroupInfo from cache, dbName:%s, load it from mnode, update:%d", dbName, forceUpdate);
D
dapan1121 已提交
716 717 718 719 720
  }

  SUseDbOutput DbOut = {0};
  SBuildUseDBInput input = {0};

721
  tstrncpy(input.db, dbName, tListLen(input.db));
D
dapan1121 已提交
722
  input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
H
Haojun Liao 已提交
723

D
dapan1121 已提交
724 725 726 727 728 729
  while (true) {
    CTG_ERR_RET(ctgGetDBVgroupFromMnode(pCatalog, pRpc, pMgmtEps, &input, &DbOut));
    CTG_ERR_RET(catalogUpdateDBVgroup(pCatalog, dbName, &DbOut.dbVgroup));
    CTG_ERR_RET(ctgGetDBVgroupFromCache(pCatalog, dbName, dbInfo, &inCache));

    if (!inCache) {
D
dapan1121 已提交
730
      ctgWarn("can't get db vgroup from cache, will retry, db:%s", dbName);
D
dapan1121 已提交
731 732 733 734 735 736 737 738 739 740 741 742 743 744 745
      continue;
    }

    break;
  }

  return TSDB_CODE_SUCCESS;
}


int32_t ctgValidateAndRemoveDb(struct SCatalog* pCatalog, const char* dbName, SDBVgroupInfo* dbInfo) {
  SDBVgroupInfo *oldInfo = (SDBVgroupInfo *)taosHashAcquire(pCatalog->dbCache.cache, dbName, strlen(dbName));
  if (oldInfo) {
    CTG_LOCK(CTG_WRITE, &oldInfo->lock);
    if (dbInfo->vgVersion <= oldInfo->vgVersion) {
D
dapan1121 已提交
746
      ctgInfo("db vgVersion is not new, db:%s, vgVersion:%d, current:%d", dbName, dbInfo->vgVersion, oldInfo->vgVersion);
D
dapan1121 已提交
747 748 749 750 751 752 753
      CTG_UNLOCK(CTG_WRITE, &oldInfo->lock);
      taosHashRelease(pCatalog->dbCache.cache, oldInfo);
      
      return TSDB_CODE_SUCCESS;
    }
    
    if (oldInfo->vgInfo) {
D
dapan1121 已提交
754
      ctgInfo("cleanup db vgInfo, db:%s", dbName);
D
dapan1121 已提交
755 756 757 758 759 760 761 762
      taosHashCleanup(oldInfo->vgInfo);
      oldInfo->vgInfo = NULL;
    }
    
    CTG_UNLOCK(CTG_WRITE, &oldInfo->lock);
  
    taosHashRelease(pCatalog->dbCache.cache, oldInfo);
  }
D
dapan1121 已提交
763 764 765 766

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
767 768 769 770 771 772 773 774 775 776 777 778 779 780 781
int32_t ctgRenewTableMetaImpl(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, int32_t isSTable) {
  if (NULL == pCatalog || NULL == pTransporter || NULL == pMgmtEps || NULL == pTableName) {
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }

  SVgroupInfo vgroupInfo = {0};
  int32_t code = 0;

  CTG_ERR_RET(catalogGetTableHashVgroup(pCatalog, pTransporter, pMgmtEps, pTableName, &vgroupInfo));

  STableMetaOutput voutput = {0};
  STableMetaOutput moutput = {0};
  STableMetaOutput *output = &voutput;

  if (CTG_IS_STABLE(isSTable)) {
782
    ctgDebug("will renew table meta, supposed to be stable, tbName:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
783 784

    // if get from mnode failed, will not try vnode
D
dapan1121 已提交
785 786
    CTG_ERR_JRET(ctgGetTableMetaFromMnode(pCatalog, pTransporter, pMgmtEps, pTableName, &moutput));

D
dapan1121 已提交
787
    if (CTG_IS_META_NONE(moutput.metaType)) {
D
dapan1121 已提交
788 789 790 791 792
      CTG_ERR_JRET(ctgGetTableMetaFromVnode(pCatalog, pTransporter, pMgmtEps, pTableName, &vgroupInfo, &voutput));
    } else {
      output = &moutput;
    }
  } else {
793
    ctgDebug("will renew table meta, not supposed to be stable, tbName:%s, isStable:%d", tNameGetTableName(pTableName), isSTable);
D
dapan1121 已提交
794 795

    // if get from vnode failed or no table meta, will not try mnode
D
dapan1121 已提交
796 797
    CTG_ERR_JRET(ctgGetTableMetaFromVnode(pCatalog, pTransporter, pMgmtEps, pTableName, &vgroupInfo, &voutput));

D
dapan1121 已提交
798
    if (CTG_IS_META_TABLE(voutput.metaType) && TSDB_SUPER_TABLE == voutput.tbMeta->tableType) {
799
      ctgDebug("will continue to renew table meta since got stable, tbName:%s, metaType:%d", tNameGetTableName(pTableName), voutput.metaType);
D
dapan1121 已提交
800
      
D
dapan1121 已提交
801 802 803 804 805
      CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCatalog, pTransporter, pMgmtEps, voutput.tbFname, &moutput));

      tfree(voutput.tbMeta);
      voutput.tbMeta = moutput.tbMeta;
      moutput.tbMeta = NULL;
D
dapan1121 已提交
806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821
    } else if (CTG_IS_META_BOTH(voutput.metaType)) {
      int32_t exist = 0;
      CTG_ERR_JRET(ctgIsTableMetaExistInCache(pCatalog, voutput.tbFname, &exist));
      if (0 == exist) {
        CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCatalog, pTransporter, pMgmtEps, voutput.tbFname, &moutput));

        if (CTG_IS_META_NONE(moutput.metaType)) {
          SET_META_TYPE_NONE(voutput.metaType);
        }
        
        tfree(voutput.tbMeta);
        voutput.tbMeta = moutput.tbMeta;
        moutput.tbMeta = NULL;
      } else {
        SET_META_TYPE_CTABLE(voutput.metaType); 
      }
D
dapan1121 已提交
822 823 824
    }
  }

D
dapan1121 已提交
825
  if (CTG_IS_META_NONE(output->metaType)) {
826
    ctgError("no tablemeta got, tbNmae:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
827 828 829
    CTG_ERR_JRET(CTG_ERR_CODE_TABLE_NOT_EXIST);
  }

D
dapan1121 已提交
830 831 832 833 834 835 836 837 838 839
  CTG_ERR_JRET(ctgUpdateTableMetaCache(pCatalog, output));

_return:

  tfree(voutput.tbMeta);
  tfree(moutput.tbMeta);
  
  CTG_RET(code);
}

D
dapan1121 已提交
840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865
int32_t ctgGetTableMeta(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, bool forceUpdate, STableMeta** pTableMeta, int32_t isSTable) {
  if (NULL == pCatalog || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == pTableMeta) {
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }
  
  int32_t exist = 0;

  if (!forceUpdate) {
    CTG_ERR_RET(ctgGetTableMetaFromCache(pCatalog, pTableName, pTableMeta, &exist));

    if (exist && CTG_TBTYPE_MATCH(isSTable, (*pTableMeta)->tableType)) {
      return TSDB_CODE_SUCCESS;
    }
  } else if (CTG_IS_UNKNOWN_STABLE(isSTable)) {
    int32_t tbType = 0;
    
    CTG_ERR_RET(ctgGetTableTypeFromCache(pCatalog, pTableName, &tbType));

    CTG_SET_STABLE(isSTable, tbType);
  }

  CTG_ERR_RET(ctgRenewTableMetaImpl(pCatalog, pRpc, pMgmtEps, pTableName, isSTable));

  CTG_ERR_RET(ctgGetTableMetaFromCache(pCatalog, pTableName, pTableMeta, &exist));

  if (0 == exist) {
866
    ctgError("renew tablemeta succeed but get from cache failed, may be deleted, tbName:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
867 868 869 870 871 872
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }
  
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922
void ctgFreeMetaRent(SMetaRentMgmt *mgmt) {
  if (NULL == mgmt->slots) {
    return;
  }

  for (int32_t i = 0; i < mgmt->slotNum; ++i) {
    SRentSlotInfo *slot = &mgmt->slots[i];
    if (slot->meta) {
      taosArrayDestroy(slot->meta);
      slot->meta = NULL;
    }
  }

  tfree(mgmt->slots);
}

void ctgFreeDbCache(SDBVgroupCache *db) {
  if (NULL == db->cache) {
    return;
  }

  SDBVgroupInfo *dbInfo = NULL;
  void *pIter = taosHashIterate(db->cache, NULL);
  while (pIter) {
    dbInfo = pIter;

    if (dbInfo->vgInfo) {
      taosHashCleanup(dbInfo->vgInfo);
      dbInfo->vgInfo = NULL;
    }
    
    pIter = taosHashIterate(db->cache, pIter);
  }

  taosHashCleanup(db->cache);
  db->cache = NULL;
}

void ctgFreeTableMetaCache(STableMetaCache *table) {
  if (table->stableCache) {
    taosHashCleanup(table->stableCache);
    table->stableCache = NULL;
  }

  if (table->cache) {
    taosHashCleanup(table->cache);
    table->cache = NULL;
  }
}

D
dapan1121 已提交
923
void ctgFreeHandle(struct SCatalog* pCatalog) {
D
dapan1121 已提交
924 925 926 927 928 929
  ctgFreeMetaRent(&pCatalog->dbRent);
  ctgFreeMetaRent(&pCatalog->stableRent);
  ctgFreeDbCache(&pCatalog->dbCache);
  ctgFreeTableMetaCache(&pCatalog->tableCache);
  
  free(pCatalog);
D
dapan1121 已提交
930 931
}

D
dapan1121 已提交
932
int32_t catalogInit(SCatalogCfg *cfg) {
D
dapan1121 已提交
933
  if (ctgMgmt.pCluster) {
D
dapan1121 已提交
934
    qError("catalog already init");
D
dapan1121 已提交
935
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
936 937 938 939
  }

  if (cfg) {
    memcpy(&ctgMgmt.cfg, cfg, sizeof(*cfg));
H
Haojun Liao 已提交
940

D
dapan1121 已提交
941 942 943 944 945 946 947
    if (ctgMgmt.cfg.maxDBCacheNum == 0) {
      ctgMgmt.cfg.maxDBCacheNum = CTG_DEFAULT_CACHE_DB_NUMBER;
    }

    if (ctgMgmt.cfg.maxTblCacheNum == 0) {
      ctgMgmt.cfg.maxTblCacheNum = CTG_DEFAULT_CACHE_TABLEMETA_NUMBER;
    }
D
dapan1121 已提交
948 949 950 951 952 953 954 955

    if (ctgMgmt.cfg.dbRentSec == 0) {
      ctgMgmt.cfg.dbRentSec = CTG_DEFAULT_RENT_SECOND;
    }

    if (ctgMgmt.cfg.stableRentSec == 0) {
      ctgMgmt.cfg.stableRentSec = CTG_DEFAULT_RENT_SECOND;
    }
D
dapan1121 已提交
956 957 958
  } else {
    ctgMgmt.cfg.maxDBCacheNum = CTG_DEFAULT_CACHE_DB_NUMBER;
    ctgMgmt.cfg.maxTblCacheNum = CTG_DEFAULT_CACHE_TABLEMETA_NUMBER;
D
dapan1121 已提交
959 960
    ctgMgmt.cfg.dbRentSec = CTG_DEFAULT_RENT_SECOND;
    ctgMgmt.cfg.stableRentSec = CTG_DEFAULT_RENT_SECOND;
D
dapan 已提交
961 962
  }

D
dapan1121 已提交
963
  ctgMgmt.pCluster = taosHashInit(CTG_DEFAULT_CACHE_CLUSTER_NUMBER, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
D
dapan1121 已提交
964
  if (NULL == ctgMgmt.pCluster) {
D
dapan1121 已提交
965 966
    qError("taosHashInit %d cluster cache failed", CTG_DEFAULT_CACHE_CLUSTER_NUMBER);
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
967 968
  }

D
dapan1121 已提交
969 970
  qDebug("catalog initialized, maxDb:%u, maxTbl:%u, dbRentSec:%u, stableRentSec:%u", ctgMgmt.cfg.maxDBCacheNum, ctgMgmt.cfg.maxTblCacheNum, ctgMgmt.cfg.dbRentSec, ctgMgmt.cfg.stableRentSec);

D
dapan 已提交
971
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
972 973
}

974 975
int32_t catalogGetHandle(uint64_t clusterId, struct SCatalog** catalogHandle) {
  if (NULL == catalogHandle) {
D
dapan1121 已提交
976
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
977 978 979
  }

  if (NULL == ctgMgmt.pCluster) {
D
dapan1121 已提交
980
    qError("cluster cache are not ready, clusterId:%"PRIx64, clusterId);
D
dapan1121 已提交
981
    CTG_ERR_RET(TSDB_CODE_CTG_NOT_READY);
D
dapan 已提交
982 983
  }

D
dapan1121 已提交
984 985
  int32_t code = 0;
  SCatalog *clusterCtg = NULL;
D
dapan 已提交
986

D
dapan1121 已提交
987 988
  while (true) {
    SCatalog **ctg = (SCatalog **)taosHashGet(ctgMgmt.pCluster, (char*)&clusterId, sizeof(clusterId));
D
dapan 已提交
989

D
dapan1121 已提交
990 991 992 993 994
    if (ctg && (*ctg)) {
      *catalogHandle = *ctg;
      qDebug("got catalog handle from cache, clusterId:%"PRIx64", CTG:%p", clusterId, *ctg);
      return TSDB_CODE_SUCCESS;
    }
D
dapan 已提交
995

D
dapan1121 已提交
996 997 998 999 1000 1001
    clusterCtg = calloc(1, sizeof(SCatalog));
    if (NULL == clusterCtg) {
      qError("calloc %d failed", (int32_t)sizeof(SCatalog));
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }

D
dapan1121 已提交
1002 1003
    clusterCtg->clusterId = clusterId;

D
dapan1121 已提交
1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020
    CTG_ERR_JRET(ctgMetaRentInit(&clusterCtg->dbRent, ctgMgmt.cfg.dbRentSec, CTG_RENT_DB));
    CTG_ERR_JRET(ctgMetaRentInit(&clusterCtg->stableRent, ctgMgmt.cfg.stableRentSec, CTG_RENT_STABLE));

    code = taosHashPut(ctgMgmt.pCluster, &clusterId, sizeof(clusterId), &clusterCtg, POINTER_BYTES);
    if (code) {
      if (HASH_NODE_EXIST(code)) {
        ctgFreeHandle(clusterCtg);
        continue;
      }
      
      qError("taosHashPut CTG to cache failed, clusterId:%"PRIx64, clusterId);
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
    }

    qDebug("add CTG to cache, clusterId:%"PRIx64", CTG:%p", clusterId, clusterCtg);

    break;
D
dapan 已提交
1021
  }
D
dapan1121 已提交
1022 1023

  *catalogHandle = clusterCtg;
D
dapan 已提交
1024
  
D
dapan1121 已提交
1025
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037

_return:

  ctgFreeHandle(clusterCtg);
  
  CTG_RET(code);
}

void catalogFreeHandle(struct SCatalog* pCatalog) {
  if (NULL == pCatalog) {
    return;
  }
D
dapan1121 已提交
1038 1039 1040 1041 1042 1043 1044

  if (taosHashRemove(ctgMgmt.pCluster, &pCatalog->clusterId, sizeof(pCatalog->clusterId))) {
    ctgWarn("taosHashRemove from cluster failed, may already be freed, clusterId:%"PRIx64, pCatalog->clusterId);
    return;
  }

  uint64_t clusterId = pCatalog->clusterId;
D
dapan1121 已提交
1045 1046
  
  ctgFreeHandle(pCatalog);
D
dapan1121 已提交
1047 1048

  ctgInfo("handle freed, culsterId:%"PRIx64, clusterId);
D
dapan 已提交
1049 1050
}

D
dapan1121 已提交
1051 1052
int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName, int32_t* version) {
  if (NULL == pCatalog || NULL == dbName || NULL == version) {
D
dapan1121 已提交
1053
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
1054 1055 1056 1057
  }

  if (NULL == pCatalog->dbCache.cache) {
    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
1058
    ctgInfo("empty db cache, dbName:%s", dbName);
D
dapan1121 已提交
1059 1060 1061
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
1062
  SDBVgroupInfo * dbInfo = taosHashAcquire(pCatalog->dbCache.cache, dbName, strlen(dbName));
D
dapan1121 已提交
1063 1064
  if (NULL == dbInfo) {
    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
1065
    ctgInfo("db not in cache, dbName:%s", dbName);
D
dapan1121 已提交
1066 1067 1068
    return TSDB_CODE_SUCCESS;
  }

1069
  *version = dbInfo->vgVersion;
D
dapan1121 已提交
1070
  taosHashRelease(pCatalog->dbCache.cache, dbInfo);
D
dapan1121 已提交
1071

D
dapan1121 已提交
1072 1073
  ctgDebug("Got db vgVersion from cache, dbName:%s, vgVersion:%d", dbName, *version);

D
dapan1121 已提交
1074 1075 1076
  return TSDB_CODE_SUCCESS;
}

1077
int32_t catalogGetDBVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* dbName, bool forceUpdate, SArray** vgroupList) {
D
dapan1121 已提交
1078 1079 1080 1081
  if (NULL == pCatalog || NULL == dbName || NULL == pRpc || NULL == pMgmtEps || NULL == vgroupList) {
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }

H
Haojun Liao 已提交
1082
  SDBVgroupInfo* db   = NULL;
D
dapan1121 已提交
1083
  SVgroupInfo *vgInfo = NULL;
1084 1085

  int32_t code = 0;
D
dapan1121 已提交
1086 1087 1088 1089 1090
  SArray *vgList = NULL;
  CTG_ERR_JRET(ctgGetDBVgroup(pCatalog, pRpc, pMgmtEps, dbName, forceUpdate, &db));

  vgList = taosArrayInit(taosHashGetSize(db->vgInfo), sizeof(SVgroupInfo));
  if (NULL == vgList) {
D
dapan1121 已提交
1091
    ctgError("taosArrayInit %d failed", taosHashGetSize(db->vgInfo));
D
dapan1121 已提交
1092 1093 1094 1095 1096 1097 1098 1099
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);    
  }

  void *pIter = taosHashIterate(db->vgInfo, NULL);
  while (pIter) {
    vgInfo = pIter;

    if (NULL == taosArrayPush(vgList, vgInfo)) {
D
dapan1121 已提交
1100
      ctgError("taosArrayPush failed, vgId:%d", vgInfo->vgId);
D
dapan1121 已提交
1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
    }
    
    pIter = taosHashIterate(db->vgInfo, pIter);
    vgInfo = NULL;
  }

  *vgroupList = vgList;
  vgList = NULL;

_return:
  if (db) {
    CTG_UNLOCK(CTG_READ, &db->lock);
    taosHashRelease(pCatalog->dbCache.cache, db);
  }

  if (vgList) {
    taosArrayDestroy(vgList);
    vgList = NULL;
  }

  CTG_RET(code);  
}


D
dapan1121 已提交
1126
int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, SDBVgroupInfo* dbInfo) {
D
dapan1121 已提交
1127 1128
  int32_t code = 0;
  
D
dapan1121 已提交
1129
  if (NULL == pCatalog || NULL == dbName || NULL == dbInfo) {
D
dapan1121 已提交
1130 1131 1132 1133
    CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
  }

  if (NULL == dbInfo->vgInfo || dbInfo->vgVersion < 0 || taosHashGetSize(dbInfo->vgInfo) <= 0) {
D
dapan1121 已提交
1134
    ctgError("invalid db vgInfo, dbName:%s, vgInfo:%p, vgVersion:%d", dbName, dbInfo->vgInfo, dbInfo->vgVersion);
D
dapan1121 已提交
1135
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1136 1137
  }

1138
  if (dbInfo->vgVersion < 0) {
D
dapan1121 已提交
1139
    ctgWarn("db vgVersion less than 0, dbName:%s, vgVersion:%d", dbName, dbInfo->vgVersion);
D
dapan1121 已提交
1140

D
dapan1121 已提交
1141
    if (pCatalog->dbCache.cache) {
D
dapan1121 已提交
1142 1143 1144
      CTG_ERR_JRET(ctgValidateAndRemoveDb(pCatalog, dbName, dbInfo));
      
      CTG_ERR_JRET(taosHashRemove(pCatalog->dbCache.cache, dbName, strlen(dbName)));
D
dapan1121 已提交
1145 1146
    }
    
D
dapan1121 已提交
1147
    ctgWarn("db removed from cache, db:%s", dbName);
D
dapan1121 已提交
1148
    goto _return;
D
dapan1121 已提交
1149
  }
D
dapan1121 已提交
1150

D
dapan1121 已提交
1151
  if (NULL == pCatalog->dbCache.cache) {
D
dapan1121 已提交
1152 1153
    SHashObj *cache = taosHashInit(ctgMgmt.cfg.maxDBCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
    if (NULL == cache) {
D
dapan1121 已提交
1154
      ctgError("taosHashInit %d failed", CTG_DEFAULT_CACHE_DB_NUMBER);
D
dapan1121 已提交
1155
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1156
    }
D
dapan1121 已提交
1157 1158 1159 1160

    if (NULL != atomic_val_compare_exchange_ptr(&pCatalog->dbCache.cache, NULL, cache)) {
      taosHashCleanup(cache);
    }
1161
  } else {
D
dapan1121 已提交
1162
    CTG_ERR_JRET(ctgValidateAndRemoveDb(pCatalog, dbName, dbInfo));
D
dapan1121 已提交
1163 1164
  }

D
dapan1121 已提交
1165 1166 1167
  bool newAdded = false;
  if (taosHashPutExt(pCatalog->dbCache.cache, dbName, strlen(dbName), dbInfo, sizeof(*dbInfo), &newAdded) != 0) {
    ctgError("taosHashPutExt db vgroup to cache failed, db:%s", dbName);
D
dapan1121 已提交
1168
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1169 1170
  }

D
dapan1121 已提交
1171 1172 1173 1174 1175 1176 1177 1178 1179
  dbInfo->vgInfo = NULL;

  SDbVgVersion vgVersion = {.dbId = dbInfo->dbId, .vgVersion = dbInfo->vgVersion};
  if (newAdded) {
    CTG_ERR_JRET(ctgMetaRentAdd(&pCatalog->dbRent, &vgVersion, dbInfo->dbId, sizeof(SDbVgVersion)));
  } else {
    CTG_ERR_JRET(ctgMetaRentUpdate(&pCatalog->dbRent, &vgVersion, dbInfo->dbId, sizeof(SDbVgVersion), ctgDbVgVersionCompare));
  }
  
D
dapan1121 已提交
1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190
  ctgDebug("dbName:%s vgroup updated, vgVersion:%d", dbName, dbInfo->vgVersion);


_return:

  if (dbInfo && dbInfo->vgInfo) {
    taosHashCleanup(dbInfo->vgInfo);
    dbInfo->vgInfo = NULL;
  }
  
  CTG_RET(code);
D
dapan1121 已提交
1191 1192
}

H
Haojun Liao 已提交
1193
int32_t catalogGetTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta) {
D
dapan1121 已提交
1194
  return ctgGetTableMeta(pCatalog, pTransporter, pMgmtEps, pTableName, false, pTableMeta, -1);
D
dapan1121 已提交
1195
}
D
dapan1121 已提交
1196

D
dapan1121 已提交
1197
int32_t catalogGetSTableMeta(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta) {
D
dapan1121 已提交
1198
  return ctgGetTableMeta(pCatalog, pTransporter, pMgmtEps, pTableName, false, pTableMeta, 1);
D
dapan1121 已提交
1199 1200 1201
}

int32_t catalogRenewTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, int32_t isSTable) {
1202
  if (NULL == pCatalog || NULL == pTransporter || NULL == pMgmtEps || NULL == pTableName) {
D
dapan1121 已提交
1203
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
1204 1205
  }

D
dapan1121 已提交
1206
  return ctgRenewTableMetaImpl(pCatalog, pTransporter, pMgmtEps, pTableName, isSTable);
1207
}
1208

D
dapan1121 已提交
1209
int32_t catalogRenewAndGetTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t isSTable) {
D
dapan1121 已提交
1210
  return ctgGetTableMeta(pCatalog, pTransporter, pMgmtEps, pTableName, true, pTableMeta, isSTable);
D
dapan1121 已提交
1211 1212
}

H
Haojun Liao 已提交
1213 1214
int32_t catalogGetTableDistVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgroupList) {
  if (NULL == pCatalog || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == pVgroupList) {
D
dapan1121 已提交
1215
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
1216 1217 1218 1219 1220
  }
  
  STableMeta *tbMeta = NULL;
  int32_t code = 0;
  SVgroupInfo vgroupInfo = {0};
D
dapan1121 已提交
1221 1222 1223 1224
  SDBVgroupInfo* dbVgroup = NULL;
  SArray *vgList = NULL;

  *pVgroupList = NULL;
D
dapan1121 已提交
1225
  
D
dapan1121 已提交
1226
  CTG_ERR_JRET(ctgGetTableMeta(pCatalog, pRpc, pMgmtEps, pTableName, false, &tbMeta, -1));
D
dapan1121 已提交
1227

H
Haojun Liao 已提交
1228 1229
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);
H
Haojun Liao 已提交
1230
  CTG_ERR_JRET(ctgGetDBVgroup(pCatalog, pRpc, pMgmtEps, db, false, &dbVgroup));
D
dapan 已提交
1231

D
dapan 已提交
1232 1233 1234 1235 1236 1237 1238 1239 1240 1241
  // REMOEV THIS ....
  if (0 == tbMeta->vgId) {
    SVgroupInfo vgroup = {0};
    
    catalogGetTableHashVgroup(pCatalog, pRpc, pMgmtEps, pTableName, &vgroup);

    tbMeta->vgId = vgroup.vgId;
  }
  // REMOVE THIS ....

1242
  if (tbMeta->tableType == TSDB_SUPER_TABLE) {
D
dapan1121 已提交
1243
    CTG_ERR_JRET(ctgGetVgInfoFromDB(pCatalog, pRpc, pMgmtEps, dbVgroup, pVgroupList));
D
dapan1121 已提交
1244
  } else {
1245
    int32_t vgId = tbMeta->vgId;
D
dapan1121 已提交
1246
    if (NULL == taosHashGetClone(dbVgroup->vgInfo, &vgId, sizeof(vgId), &vgroupInfo)) {
1247
      ctgError("table's vgId not found in vgroup list, vgId:%d, tbName:%s", vgId, tNameGetTableName(pTableName));
D
dapan 已提交
1248
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);    
1249
    }
D
dapan1121 已提交
1250

D
dapan1121 已提交
1251 1252
    vgList = taosArrayInit(1, sizeof(SVgroupInfo));
    if (NULL == vgList) {
D
dapan1121 已提交
1253
      ctgError("taosArrayInit %d failed", (int32_t)sizeof(SVgroupInfo));
D
dapan 已提交
1254 1255 1256
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);    
    }

D
dapan1121 已提交
1257
    if (NULL == taosArrayPush(vgList, &vgroupInfo)) {
1258
      ctgError("taosArrayPush vgroupInfo to array failed, vgId:%d, tbName:%s", vgId, tNameGetTableName(pTableName));
D
dapan1121 已提交
1259 1260
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
    }
D
dapan 已提交
1261

D
dapan1121 已提交
1262 1263 1264
    *pVgroupList = vgList;
    vgList = NULL;
  }
D
dapan 已提交
1265

D
dapan1121 已提交
1266 1267
_return:
  tfree(tbMeta);
D
dapan 已提交
1268

D
dapan1121 已提交
1269 1270 1271 1272 1273 1274 1275 1276 1277
  if (dbVgroup) {
    CTG_UNLOCK(CTG_READ, &dbVgroup->lock);
    taosHashRelease(pCatalog->dbCache.cache, dbVgroup);
  }

  if (vgList) {
    taosArrayDestroy(vgList);
    vgList = NULL;
  }
D
dapan1121 已提交
1278
  
D
dapan1121 已提交
1279
  CTG_RET(code);
D
dapan1121 已提交
1280 1281 1282
}


H
Haojun Liao 已提交
1283
int32_t catalogGetTableHashVgroup(struct SCatalog *pCatalog, void *pTransporter, const SEpSet *pMgmtEps, const SName *pTableName, SVgroupInfo *pVgroup) {
D
dapan1121 已提交
1284
  SDBVgroupInfo* dbInfo = NULL;
D
dapan1121 已提交
1285 1286
  int32_t code = 0;

H
Haojun Liao 已提交
1287 1288
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);
D
dapan1121 已提交
1289

H
Haojun Liao 已提交
1290
  CTG_ERR_RET(ctgGetDBVgroup(pCatalog, pTransporter, pMgmtEps, db, false, &dbInfo));
D
dapan1121 已提交
1291

D
dapan1121 已提交
1292
  CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCatalog, dbInfo, pTableName, pVgroup));
D
dapan1121 已提交
1293

D
dapan1121 已提交
1294 1295 1296 1297 1298
_return:
  if (dbInfo) {
    CTG_UNLOCK(CTG_READ, &dbInfo->lock);  
    taosHashRelease(pCatalog->dbCache.cache, dbInfo);
  }
D
dapan1121 已提交
1299

D
dapan1121 已提交
1300
  CTG_RET(code);
D
dapan1121 已提交
1301 1302 1303
}


H
Haojun Liao 已提交
1304 1305
int32_t catalogGetAllMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp) {
  if (NULL == pCatalog || NULL == pTransporter || NULL == pMgmtEps || NULL == pReq || NULL == pRsp) {
D
dapan1121 已提交
1306
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
1307
  }
D
dapan1121 已提交
1308 1309 1310 1311 1312

  int32_t code = 0;

  if (pReq->pTableName) {
    int32_t tbNum = (int32_t)taosArrayGetSize(pReq->pTableName);
D
dapan1121 已提交
1313
    if (tbNum <= 0) {
D
dapan1121 已提交
1314
      ctgError("empty table name list, tbNum:%d", tbNum);
D
dapan1121 已提交
1315 1316
      CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
    }
H
Haojun Liao 已提交
1317

D
dapan1121 已提交
1318 1319
    pRsp->pTableMeta = taosArrayInit(tbNum, POINTER_BYTES);
    if (NULL == pRsp->pTableMeta) {
D
dapan1121 已提交
1320
      ctgError("taosArrayInit %d failed", tbNum);
D
dapan1121 已提交
1321
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1322 1323 1324 1325 1326 1327
    }
    
    for (int32_t i = 0; i < tbNum; ++i) {
      SName *name = taosArrayGet(pReq->pTableName, i);
      STableMeta *pTableMeta = NULL;
      
H
Haojun Liao 已提交
1328
      CTG_ERR_JRET(ctgGetTableMeta(pCatalog, pTransporter, pMgmtEps, name, false, &pTableMeta, -1));
D
dapan1121 已提交
1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348

      if (NULL == taosArrayPush(pRsp->pTableMeta, &pTableMeta)) {
        ctgError("taosArrayPush failed, idx:%d", i);
        tfree(pTableMeta);
        CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
      }
    }
  }

  return TSDB_CODE_SUCCESS;

_return:  
  if (pRsp->pTableMeta) {
    int32_t aSize = taosArrayGetSize(pRsp->pTableMeta);
    for (int32_t i = 0; i < aSize; ++i) {
      STableMeta *pMeta = taosArrayGetP(pRsp->pTableMeta, i);
      tfree(pMeta);
    }
    
    taosArrayDestroy(pRsp->pTableMeta);
D
dapan1121 已提交
1349
    pRsp->pTableMeta = NULL;
D
dapan1121 已提交
1350
  }
D
dapan 已提交
1351
  
D
dapan1121 已提交
1352
  CTG_RET(code);
1353
}
D
dapan 已提交
1354

D
dapan1121 已提交
1355 1356
int32_t catalogGetQnodeList(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, SArray* pQnodeList) {
  if (NULL == pCatalog || NULL == pRpc  || NULL == pMgmtEps || NULL == pQnodeList) {
D
dapan 已提交
1357 1358 1359
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
1360
  //TODO
D
dapan 已提交
1361 1362 1363 1364

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377 1378 1379 1380
int32_t catalogGetExpiredSTables(struct SCatalog* pCatalog, SSTableMetaVersion **stables, uint32_t *num) {
  if (NULL == pCatalog || NULL == stables || NULL == num) {
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }

  CTG_RET(ctgMetaRentGet(&pCatalog->stableRent, (void **)stables, num, sizeof(SSTableMetaVersion)));
}

int32_t catalogGetExpiredDBs(struct SCatalog* pCatalog, SDbVgVersion **dbs, uint32_t *num) {
  if (NULL == pCatalog || NULL == dbs || NULL == num) {
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }

  CTG_RET(ctgMetaRentGet(&pCatalog->dbRent, (void **)dbs, num, sizeof(SDbVgVersion)));
}

D
dapan 已提交
1381

D
dapan 已提交
1382
void catalogDestroy(void) {
D
dapan1121 已提交
1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396
  if (NULL == ctgMgmt.pCluster) {
    return;
  }

  SCatalog *pCatalog = NULL;
  void *pIter = taosHashIterate(ctgMgmt.pCluster, NULL);
  while (pIter) {
    pCatalog = *(SCatalog **)pIter;

    if (pCatalog) {
      catalogFreeHandle(pCatalog);
    }
    
    pIter = taosHashIterate(ctgMgmt.pCluster, pIter);
D
dapan 已提交
1397
  }
D
dapan1121 已提交
1398 1399 1400
  
  taosHashCleanup(ctgMgmt.pCluster);
  ctgMgmt.pCluster = NULL;
D
dapan1121 已提交
1401 1402

  qInfo("catalog destroyed");
D
dapan 已提交
1403 1404 1405 1406
}