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

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

D
dapan1121 已提交
30
  SDBVgroupInfo *info = NULL;
D
dapan1121 已提交
31

D
dapan1121 已提交
32 33
  while (true) {
    info = taosHashAcquire(pCatalog->dbCache.cache, dbName, strlen(dbName));
D
dapan1121 已提交
34

D
dapan1121 已提交
35 36
    if (NULL == info) {
      *inCache = false;
D
dapan1121 已提交
37
      ctgWarn("not in db vgroup cache, dbName:%s", dbName);
D
dapan1121 已提交
38 39 40 41 42 43 44 45 46 47 48 49 50
      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 已提交
51
  }
D
dapan1121 已提交
52

D
dapan1121 已提交
53 54
  *dbInfo = info;
  *inCache = true;
D
dapan1121 已提交
55 56

  ctgDebug("Got db vgroup from cache, dbName:%s", dbName);
D
dapan1121 已提交
57 58 59 60 61 62 63 64 65 66 67
  
  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 已提交
68 69 70 71 72 73 74
  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 已提交
75
  
D
dapan1121 已提交
76
  SRpcMsg rpcMsg = {
H
Hongze Cheng 已提交
77
      .msgType = TDMT_MND_USE_DB,
D
catalog  
dapan1121 已提交
78
      .pCont   = msg,
D
dapan1121 已提交
79 80 81 82 83 84
      .contLen = msgLen,
  };

  SRpcMsg rpcRsp = {0};

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

D
dapan1121 已提交
90 91 92 93 94
  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 已提交
95

D
dapan1121 已提交
96 97
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
98

D
dapan1121 已提交
99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121
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 已提交
122

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

D
dapan1121 已提交
130
  char tbFullName[TSDB_TABLE_FNAME_LEN];
H
Haojun Liao 已提交
131
  tNameExtractFullName(pTableName, tbFullName);
D
dapan1121 已提交
132

D
dapan1121 已提交
133 134 135 136
  *pTableMeta = NULL;

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

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

D
dapan1121 已提交
144
  *exist = 1;
D
dapan1121 已提交
145 146
  
  tbMeta = *pTableMeta;
D
dapan1121 已提交
147

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

D
dapan1121 已提交
151 152 153 154 155 156 157 158
    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 已提交
159
    ctgError("stable not in stableCache, suid:%"PRIx64, tbMeta->suid);
D
dapan1121 已提交
160 161 162 163
    tfree(*pTableMeta);
    *exist = 0;
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
164

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

D
dapan1121 已提交
172 173 174 175
  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 已提交
176
    ctgError("realloc size[%d] failed", metaSize);
D
dapan1121 已提交
177
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
178 179
  }

D
dapan1121 已提交
180 181 182
  memcpy(&(*pTableMeta)->sversion, &(*stbMeta)->sversion, metaSize - sizeof(SCTableMeta));

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

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

D
dapan1121 已提交
189 190
int32_t ctgGetTableTypeFromCache(struct SCatalog* pCatalog, const SName* pTableName, int32_t *tbType) {
  if (NULL == pCatalog->tableCache.cache) {
D
dapan1121 已提交
191
    ctgWarn("empty tablemeta cache, tbName:%s", pTableName->tname);  
D
dapan1121 已提交
192 193 194 195 196 197 198 199 200 201 202 203
    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 已提交
204 205
    ctgWarn("tablemeta not in cache, tbName:%s", tbFullName);  
  
D
dapan1121 已提交
206 207 208 209
    return TSDB_CODE_SUCCESS;
  }

  *tbType = pTableMeta->tableType;
D
dapan1121 已提交
210 211

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


D
dapan1121 已提交
217 218 219 220 221 222 223 224 225 226
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 已提交
227
int32_t ctgGetTableMetaFromMnodeImpl(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, char* tbFullName, STableMetaOutput* output) {
D
dapan1121 已提交
228
  SBuildTableMetaInput bInput = {.vgId = 0, .dbName = NULL, .tableFullName = tbFullName};
D
dapan1121 已提交
229 230 231 232
  char *msg = NULL;
  SEpSet *pVnodeEpSet = NULL;
  int32_t msgLen = 0;

D
dapan1121 已提交
233 234 235 236 237 238 239
  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 已提交
240 241 242 243 244 245

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

D
dapan1121 已提交
247 248
  SRpcMsg rpcRsp = {0};

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

D
dapan1121 已提交
262 263 264 265 266 267 268
  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 已提交
269 270 271 272

  return TSDB_CODE_SUCCESS;
}

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

D
dapan1121 已提交
277
  return ctgGetTableMetaFromMnodeImpl(pCatalog, pTransporter, pMgmtEps, tbFullName, output);
D
dapan1121 已提交
278
}
D
dapan1121 已提交
279

D
dapan1121 已提交
280 281
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 已提交
282
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
283 284
  }

D
dapan1121 已提交
285 286
  char dbFullName[TSDB_DB_FNAME_LEN];
  tNameGetFullDbName(pTableName, dbFullName);
D
dapan1121 已提交
287

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

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

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

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

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

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

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

  return TSDB_CODE_SUCCESS;
}


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

  return TSDB_CODE_SUCCESS;
}

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

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

359 360 361
  void *pIter = taosHashIterate(dbInfo->vgInfo, NULL);
  while (pIter) {
    vgInfo = pIter;
D
dapan1121 已提交
362

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

D
dapan1121 已提交
372 373 374
  *vgroupList = vgList;
  vgList = NULL;

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

D
dapan1121 已提交
377
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
378 379 380 381 382 383 384 385

_return:

  if (vgList) {
    taosArrayDestroy(vgList);
  }

  CTG_RET(code);
D
dapan1121 已提交
386 387
}

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

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

400 401
  tableNameHashFp fp = NULL;
  SVgroupInfo *vgInfo = NULL;
D
dapan1121 已提交
402

D
dapan1121 已提交
403
  CTG_ERR_JRET(ctgGetHashFunction(dbInfo->hashMethod, &fp));
404 405

  char tbFullName[TSDB_TABLE_FNAME_LEN];
H
Haojun Liao 已提交
406
  tNameExtractFullName(pTableName, tbFullName);
407 408 409 410 411 412 413

  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) {
414
      taosHashCancelIterate(dbInfo->vgInfo, pIter);
415
      break;
D
dapan1121 已提交
416
    }
417 418 419
    
    pIter = taosHashIterate(dbInfo->vgInfo, pIter);
    vgInfo = NULL;
D
dapan1121 已提交
420 421
  }

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

  *pVgroup = *vgInfo;

D
dapan1121 已提交
429
_return:
430
  CTG_RET(code);
D
dapan1121 已提交
431 432
}

D
dapan1121 已提交
433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449
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 已提交
450
  }
D
dapan1121 已提交
451 452 453 454 455 456 457 458 459
}


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 已提交
460
  
D
dapan1121 已提交
461 462 463 464 465
  mgmt->slots = calloc(1, msgSize);
  if (NULL == mgmt->slots) {
    qError("calloc %d failed", (int32_t)msgSize);
    return TSDB_CODE_CTG_MEM_ERROR;
  }
D
dapan1121 已提交
466

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

D
dapan1121 已提交
472 473

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

  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 已提交
485
    }
D
dapan1121 已提交
486
  }
D
dapan1121 已提交
487

D
dapan1121 已提交
488 489 490
  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 已提交
491 492
  }

D
dapan1121 已提交
493
  slot->needSort = true;
D
dapan1121 已提交
494

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

D
dapan1121 已提交
497 498 499 500 501 502 503
_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 已提交
504
  int16_t widx = abs(id % mgmt->slotNum);
D
dapan1121 已提交
505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521

  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 已提交
522
  if (NULL == orig) {
D
dapan1121 已提交
523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547
    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 已提交
548
  }
D
dapan1121 已提交
549 550 551

  SRentSlotInfo *slot = &mgmt->slots[ridx];
  int32_t code = 0;
D
dapan1121 已提交
552
  
D
dapan1121 已提交
553 554 555 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
  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 已提交
609 610 611 612
  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
613

D
dapan1121 已提交
614
int32_t ctgUpdateTableMetaCache(struct SCatalog *pCatalog, STableMetaOutput *output) {
D
dapan1121 已提交
615
  int32_t code = 0;
D
dapan1121 已提交
616 617

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

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

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

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

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

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

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

D
dapan1121 已提交
655 656
  if (CTG_IS_META_CTABLE(output->metaType)) {
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
657
  }
D
dapan1121 已提交
658 659 660 661 662
  
  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 已提交
663

D
dapan1121 已提交
664
  int32_t tbSize = sizeof(*output->tbMeta) + sizeof(SSchema) * (output->tbMeta->tableInfo.numOfColumns + output->tbMeta->tableInfo.numOfTags);
D
dapan1121 已提交
665 666

  if (TSDB_SUPER_TABLE == output->tbMeta->tableType) {
D
dapan1121 已提交
667 668 669
    bool newAdded = false;
    SSTableMetaVersion metaRent = {.suid = output->tbMeta->suid, .sversion = output->tbMeta->sversion, .tversion = output->tbMeta->tversion};
    
D
dapan1121 已提交
670 671 672
    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 已提交
673
      ctgError("taosHashPut tablemeta to cache failed, tbName:%s", output->tbFname);
D
dapan1121 已提交
674
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
675 676 677
    }

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

    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 已提交
692 693
  } else {
    if (taosHashPut(pCatalog->tableCache.cache, output->tbFname, strlen(output->tbFname), output->tbMeta, tbSize) != 0) {
D
dapan1121 已提交
694
      ctgError("taosHashPut tablemeta to cache failed, tbName:%s", output->tbFname);
D
dapan1121 已提交
695
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
696 697
    }
  }
D
dapan1121 已提交
698

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

D
dapan1121 已提交
701
  CTG_RET(code);
D
dapan1121 已提交
702 703
}

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

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

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

718
  tstrncpy(input.db, dbName, tListLen(input.db));
D
dapan1121 已提交
719
  input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
H
Haojun Liao 已提交
720

D
dapan1121 已提交
721 722 723 724 725 726
  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 已提交
727
      ctgWarn("can't get db vgroup from cache, will retry, db:%s", dbName);
D
dapan1121 已提交
728 729 730 731 732 733 734 735 736 737 738 739 740 741 742
      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 已提交
743
      ctgInfo("db vgVersion is not new, db:%s, vgVersion:%d, current:%d", dbName, dbInfo->vgVersion, oldInfo->vgVersion);
D
dapan1121 已提交
744 745 746 747 748 749 750
      CTG_UNLOCK(CTG_WRITE, &oldInfo->lock);
      taosHashRelease(pCatalog->dbCache.cache, oldInfo);
      
      return TSDB_CODE_SUCCESS;
    }
    
    if (oldInfo->vgInfo) {
D
dapan1121 已提交
751
      ctgInfo("cleanup db vgInfo, db:%s", dbName);
D
dapan1121 已提交
752 753 754 755 756 757 758 759
      taosHashCleanup(oldInfo->vgInfo);
      oldInfo->vgInfo = NULL;
    }
    
    CTG_UNLOCK(CTG_WRITE, &oldInfo->lock);
  
    taosHashRelease(pCatalog->dbCache.cache, oldInfo);
  }
D
dapan1121 已提交
760 761 762 763

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
764 765 766 767 768 769 770 771 772 773 774 775 776 777 778
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)) {
779
    ctgDebug("will renew table meta, supposed to be stable, tbName:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
780 781

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

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

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

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

      tfree(voutput.tbMeta);
      voutput.tbMeta = moutput.tbMeta;
      moutput.tbMeta = NULL;
D
dapan1121 已提交
803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818
    } 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 已提交
819 820 821
    }
  }

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

D
dapan1121 已提交
827 828 829 830 831 832 833 834 835 836
  CTG_ERR_JRET(ctgUpdateTableMetaCache(pCatalog, output));

_return:

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

D
dapan1121 已提交
837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862
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) {
863
    ctgError("renew tablemeta succeed but get from cache failed, may be deleted, tbName:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
864 865 866 867 868 869
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }
  
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
870 871 872 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
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 已提交
920
void ctgFreeHandle(struct SCatalog* pCatalog) {
D
dapan1121 已提交
921 922 923 924 925 926
  ctgFreeMetaRent(&pCatalog->dbRent);
  ctgFreeMetaRent(&pCatalog->stableRent);
  ctgFreeDbCache(&pCatalog->dbCache);
  ctgFreeTableMetaCache(&pCatalog->tableCache);
  
  free(pCatalog);
D
dapan1121 已提交
927 928
}

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

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

D
dapan1121 已提交
938 939 940 941 942 943 944
    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 已提交
945 946 947 948 949 950 951 952

    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 已提交
953 954 955
  } else {
    ctgMgmt.cfg.maxDBCacheNum = CTG_DEFAULT_CACHE_DB_NUMBER;
    ctgMgmt.cfg.maxTblCacheNum = CTG_DEFAULT_CACHE_TABLEMETA_NUMBER;
D
dapan1121 已提交
956 957
    ctgMgmt.cfg.dbRentSec = CTG_DEFAULT_RENT_SECOND;
    ctgMgmt.cfg.stableRentSec = CTG_DEFAULT_RENT_SECOND;
D
dapan 已提交
958 959
  }

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

D
dapan1121 已提交
966 967
  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 已提交
968
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
969 970
}

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

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

D
dapan1121 已提交
981 982
  int32_t code = 0;
  SCatalog *clusterCtg = NULL;
D
dapan 已提交
983

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

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

D
dapan1121 已提交
993 994 995 996 997 998
    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 已提交
999 1000
    clusterCtg->clusterId = clusterId;

D
dapan1121 已提交
1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017
    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 已提交
1018
  }
D
dapan1121 已提交
1019 1020

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

_return:

  ctgFreeHandle(clusterCtg);
  
  CTG_RET(code);
}

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

  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 已提交
1042 1043
  
  ctgFreeHandle(pCatalog);
D
dapan1121 已提交
1044 1045

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

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

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

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

1066
  *version = dbInfo->vgVersion;
D
dapan1121 已提交
1067
  taosHashRelease(pCatalog->dbCache.cache, dbInfo);
D
dapan1121 已提交
1068

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

D
dapan1121 已提交
1071 1072 1073
  return TSDB_CODE_SUCCESS;
}

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

H
Haojun Liao 已提交
1079
  SDBVgroupInfo* db   = NULL;
D
dapan1121 已提交
1080
  SVgroupInfo *vgInfo = NULL;
1081 1082

  int32_t code = 0;
D
dapan1121 已提交
1083 1084 1085 1086 1087
  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 已提交
1088
    ctgError("taosArrayInit %d failed", taosHashGetSize(db->vgInfo));
D
dapan1121 已提交
1089 1090 1091 1092 1093 1094 1095 1096
    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 已提交
1097
      ctgError("taosArrayPush failed, vgId:%d", vgInfo->vgId);
D
dapan1121 已提交
1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122
      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 已提交
1123
int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, SDBVgroupInfo* dbInfo) {
D
dapan1121 已提交
1124 1125
  int32_t code = 0;
  
D
dapan1121 已提交
1126
  if (NULL == pCatalog || NULL == dbName || NULL == dbInfo) {
D
dapan1121 已提交
1127 1128 1129 1130
    CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
  }

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

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

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

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

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

D
dapan1121 已提交
1162 1163 1164
  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 已提交
1165
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1166 1167
  }

D
dapan1121 已提交
1168 1169 1170 1171 1172 1173 1174 1175 1176
  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 已提交
1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187
  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 已提交
1188 1189
}

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

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

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

D
dapan1121 已提交
1203
  return ctgRenewTableMetaImpl(pCatalog, pTransporter, pMgmtEps, pTableName, isSTable);
1204
}
1205

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

H
Haojun Liao 已提交
1210 1211
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 已提交
1212
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
1213 1214 1215 1216 1217
  }
  
  STableMeta *tbMeta = NULL;
  int32_t code = 0;
  SVgroupInfo vgroupInfo = {0};
D
dapan1121 已提交
1218 1219 1220 1221
  SDBVgroupInfo* dbVgroup = NULL;
  SArray *vgList = NULL;

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

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

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

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

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

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

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

D
dapan1121 已提交
1259 1260 1261
    *pVgroupList = vgList;
    vgList = NULL;
  }
D
dapan 已提交
1262

D
dapan1121 已提交
1263 1264
_return:
  tfree(tbMeta);
D
dapan 已提交
1265

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

  if (vgList) {
    taosArrayDestroy(vgList);
    vgList = NULL;
  }
D
dapan1121 已提交
1275
  
D
dapan1121 已提交
1276
  CTG_RET(code);
D
dapan1121 已提交
1277 1278 1279
}


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

H
Haojun Liao 已提交
1284 1285
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);
D
dapan1121 已提交
1286

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

D
dapan1121 已提交
1289
  CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCatalog, dbInfo, pTableName, pVgroup));
D
dapan1121 已提交
1290

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

D
dapan1121 已提交
1297
  CTG_RET(code);
D
dapan1121 已提交
1298 1299 1300
}


H
Haojun Liao 已提交
1301 1302
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 已提交
1303
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
1304
  }
D
dapan1121 已提交
1305 1306 1307 1308 1309

  int32_t code = 0;

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

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

      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 已提交
1346
    pRsp->pTableMeta = NULL;
D
dapan1121 已提交
1347
  }
D
dapan 已提交
1348
  
D
dapan1121 已提交
1349
  CTG_RET(code);
1350
}
D
dapan 已提交
1351

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

D
dapan1121 已提交
1357
  //TODO
D
dapan 已提交
1358 1359 1360 1361

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377
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 已提交
1378

D
dapan 已提交
1379
void catalogDestroy(void) {
D
dapan1121 已提交
1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393
  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 已提交
1394
  }
D
dapan1121 已提交
1395 1396 1397
  
  taosHashCleanup(ctgMgmt.pCluster);
  ctgMgmt.pCluster = NULL;
D
dapan1121 已提交
1398 1399

  qInfo("catalog destroyed");
D
dapan 已提交
1400 1401 1402 1403
}