catalog.c 45.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) {
H
Haojun Liao 已提交
89
    ctgError("error rsp for use db, code:%s, db:%s", tstrerror(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_NULL(output->metaType);
D
dapan1121 已提交
257
      ctgDebug("stablemeta not exist in mnode, tbName:%s", tbFullName);
D
dapan1121 已提交
258 259 260
      return TSDB_CODE_SUCCESS;
    }
    
H
Haojun Liao 已提交
261
    ctgError("error rsp for stablemeta from mnode, code:%s, tbName:%s", tstrerror(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_NULL(output->metaType);
319
      ctgDebug("tablemeta not exist in vnode, tbName:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
320 321 322
      return TSDB_CODE_SUCCESS;
    }
  
H
Haojun Liao 已提交
323
    ctgError("error rsp for table meta from vnode, code:%s, tbName:%s", tstrerror(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) {
H
Haojun Liao 已提交
329
    ctgError("Process vnode tablemeta rsp failed, code:%s, tbName:%s", tstrerror(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
  return TSDB_CODE_SUCCESS;
}


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

  return TSDB_CODE_SUCCESS;
}

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

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

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

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

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

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

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

_return:

  if (vgList) {
    taosArrayDestroy(vgList);
  }

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

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

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

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

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

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

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

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

  *pVgroup = *vgInfo;

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

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


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

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

D
dapan1121 已提交
474 475

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

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

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

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

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

D
dapan1121 已提交
499 500 501 502 503 504 505
_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 已提交
506
  int16_t widx = abs(id % mgmt->slotNum);
D
dapan1121 已提交
507 508 509 510 511 512

  SRentSlotInfo *slot = &mgmt->slots[widx];
  int32_t code = 0;
  
  CTG_LOCK(CTG_WRITE, &slot->lock);
  if (NULL == slot->meta) {
D
dapan1121 已提交
513 514
    qError("empty meta slot, id:%"PRIx64", slot idx:%d, type:%d", id, widx, mgmt->type);
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
515 516 517 518 519
  }

  if (slot->needSort) {
    taosArraySort(slot->meta, compare);
    slot->needSort = false;
D
dapan1121 已提交
520
    qDebug("meta slot sorted, slot idx:%d, type:%d", widx, mgmt->type);
D
dapan1121 已提交
521 522 523
  }

  void *orig = taosArraySearch(slot->meta, &id, compare, TD_EQ);
D
dapan1121 已提交
524
  if (NULL == orig) {
D
dapan1121 已提交
525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544
    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);
}

D
dapan1121 已提交
545 546 547 548 549 550 551 552 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
int32_t ctgMetaRentRemove(SMetaRentMgmt *mgmt, int64_t id, __compar_fn_t compare) {
  int16_t widx = abs(id % mgmt->slotNum);

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

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

  int32_t idx = taosArraySearchIdx(slot->meta, &id, compare, TD_EQ);
  if (idx < 0) {
    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);
  }

  taosArrayRemove(slot->meta, idx);

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

_return:

  CTG_UNLOCK(CTG_WRITE, &slot->lock);

  CTG_RET(code);
}


D
dapan1121 已提交
581 582 583 584 585
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 已提交
586
  }
D
dapan1121 已提交
587 588 589

  SRentSlotInfo *slot = &mgmt->slots[ridx];
  int32_t code = 0;
D
dapan1121 已提交
590
  
D
dapan1121 已提交
591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646
  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 已提交
647 648 649 650
  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
651

D
dapan1121 已提交
652
int32_t ctgUpdateTableMetaCache(struct SCatalog *pCatalog, STableMetaOutput *output) {
D
dapan1121 已提交
653
  int32_t code = 0;
D
dapan1121 已提交
654 655

  if (NULL == output->tbMeta) {
D
dapan1121 已提交
656
    ctgError("no valid table meta got from meta rsp, tbName:%s", output->tbFname);
D
dapan1121 已提交
657
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
658 659 660
  }

  if (NULL == pCatalog->tableCache.cache) {
D
dapan1121 已提交
661 662
    SHashObj *cache = taosHashInit(ctgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
    if (NULL == cache) {
D
dapan1121 已提交
663
      ctgError("taosHashInit failed, num:%d", ctgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
664
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
665
    }
D
dapan1121 已提交
666 667 668 669

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

D
dapan1121 已提交
672
  if (NULL == pCatalog->tableCache.stableCache) {
D
dapan1121 已提交
673 674
    SHashObj *cache = taosHashInit(ctgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), true, HASH_ENTRY_LOCK);
    if (NULL == cache) {
D
dapan1121 已提交
675
      ctgError("taosHashInit failed, num:%d", ctgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
676
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
677
    }
D
dapan1121 已提交
678 679 680 681

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

D
dapan1121 已提交
684
  if (CTG_IS_META_CTABLE(output->metaType) || CTG_IS_META_BOTH(output->metaType)) {
D
dapan1121 已提交
685
    if (taosHashPut(pCatalog->tableCache.cache, output->ctbFname, strlen(output->ctbFname), &output->ctbMeta, sizeof(output->ctbMeta)) != 0) {
D
dapan1121 已提交
686
      ctgError("taosHashPut ctablemeta to cache failed, ctbName:%s", output->ctbFname);
D
dapan1121 已提交
687
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
688 689
    }

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

D
dapan1121 已提交
693 694
  if (CTG_IS_META_CTABLE(output->metaType)) {
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
695
  }
D
dapan1121 已提交
696 697 698 699 700
  
  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 已提交
701

D
dapan1121 已提交
702
  int32_t tbSize = sizeof(*output->tbMeta) + sizeof(SSchema) * (output->tbMeta->tableInfo.numOfColumns + output->tbMeta->tableInfo.numOfTags);
D
dapan1121 已提交
703 704

  if (TSDB_SUPER_TABLE == output->tbMeta->tableType) {
D
dapan1121 已提交
705 706 707
    bool newAdded = false;
    SSTableMetaVersion metaRent = {.suid = output->tbMeta->suid, .sversion = output->tbMeta->sversion, .tversion = output->tbMeta->tversion};
    
D
dapan1121 已提交
708 709 710
    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 已提交
711
      ctgError("taosHashPut tablemeta to cache failed, tbName:%s", output->tbFname);
D
dapan1121 已提交
712
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
713 714 715
    }

    STableMeta *tbMeta = taosHashGet(pCatalog->tableCache.cache, output->tbFname, strlen(output->tbFname));
D
dapan1121 已提交
716
    if (taosHashPutExt(pCatalog->tableCache.stableCache, &output->tbMeta->suid, sizeof(output->tbMeta->suid), &tbMeta, POINTER_BYTES, &newAdded) != 0) {
D
dapan1121 已提交
717
      CTG_UNLOCK(CTG_WRITE, &pCatalog->tableCache.stableLock);
D
dapan1121 已提交
718
      ctgError("taosHashPutExt stable to stable cache failed, suid:%"PRIx64, output->tbMeta->suid);
D
dapan1121 已提交
719
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
720 721
    }
    CTG_UNLOCK(CTG_WRITE, &pCatalog->tableCache.stableLock);
D
dapan1121 已提交
722 723 724 725 726 727 728 729

    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 已提交
730 731
  } else {
    if (taosHashPut(pCatalog->tableCache.cache, output->tbFname, strlen(output->tbFname), output->tbMeta, tbSize) != 0) {
D
dapan1121 已提交
732
      ctgError("taosHashPut tablemeta to cache failed, tbName:%s", output->tbFname);
D
dapan1121 已提交
733
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
734 735
    }
  }
D
dapan1121 已提交
736

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

D
dapan1121 已提交
739
  CTG_RET(code);
D
dapan1121 已提交
740 741
}

742
int32_t ctgGetDBVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* dbName, bool forceUpdate, SDBVgroupInfo** dbInfo) {
D
dapan1121 已提交
743
  bool inCache = false;
744
  if (!forceUpdate) {
D
dapan1121 已提交
745 746
    CTG_ERR_RET(ctgGetDBVgroupFromCache(pCatalog, dbName, dbInfo, &inCache));
    if (inCache) {
D
dapan1121 已提交
747 748
      return TSDB_CODE_SUCCESS;
    }
749 750

    ctgDebug("failed to get DB vgroupInfo from cache, dbName:%s, load it from mnode, update:%d", dbName, forceUpdate);
D
dapan1121 已提交
751 752 753 754 755
  }

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

756
  tstrncpy(input.db, dbName, tListLen(input.db));
D
dapan1121 已提交
757
  input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
H
Haojun Liao 已提交
758

D
dapan1121 已提交
759 760 761 762 763 764
  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 已提交
765
      ctgWarn("can't get db vgroup from cache, will retry, db:%s", dbName);
D
dapan1121 已提交
766 767 768 769 770 771 772 773 774 775
      continue;
    }

    break;
  }

  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
776
int32_t ctgValidateAndFreeDbInfo(struct SCatalog* pCatalog, const char* dbName, SDBVgroupInfo* dbInfo) {
D
dapan1121 已提交
777 778 779 780
  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 已提交
781
      ctgInfo("db vgVersion is not new, db:%s, vgVersion:%d, current:%d", dbName, dbInfo->vgVersion, oldInfo->vgVersion);
D
dapan1121 已提交
782 783 784 785 786 787 788
      CTG_UNLOCK(CTG_WRITE, &oldInfo->lock);
      taosHashRelease(pCatalog->dbCache.cache, oldInfo);
      
      return TSDB_CODE_SUCCESS;
    }
    
    if (oldInfo->vgInfo) {
D
dapan1121 已提交
789
      ctgInfo("cleanup db vgInfo, db:%s", dbName);
D
dapan1121 已提交
790 791 792 793 794 795 796 797
      taosHashCleanup(oldInfo->vgInfo);
      oldInfo->vgInfo = NULL;
    }
    
    CTG_UNLOCK(CTG_WRITE, &oldInfo->lock);
  
    taosHashRelease(pCatalog->dbCache.cache, oldInfo);
  }
D
dapan1121 已提交
798 799 800 801

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
802 803
int32_t ctgValidateAndRemoveDbInfo(struct SCatalog* pCatalog, SDbVgVersion* target, bool *removed) {
  *removed = false;
D
dapan1121 已提交
804

D
dapan1121 已提交
805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820
  SDBVgroupInfo *info = (SDBVgroupInfo *)taosHashAcquire(pCatalog->dbCache.cache, target->dbName, strlen(target->dbName));
  if (NULL == info) {
    ctgInfo("db not exist in dbCache, may be removed, db:%s", target->dbName);
    return TSDB_CODE_SUCCESS;
  }
  
  CTG_LOCK(CTG_WRITE, &info->lock);
  if (info->dbId != target->dbId) {
    ctgInfo("db id already updated, db:%s, dbId:%"PRIx64 ", targetId:%"PRIx64, target->dbName, info->dbId, target->dbId);
    CTG_UNLOCK(CTG_WRITE, &info->lock);
    taosHashRelease(pCatalog->dbCache.cache, info);
    return TSDB_CODE_SUCCESS;
  }
  
  if (info->vgVersion > target->vgVersion) {
    ctgInfo("db vgVersion already updated, db:%s, version:%d, targetVer:%d", target->dbName, info->vgVersion, target->vgVersion);
D
dapan1121 已提交
821
    CTG_UNLOCK(CTG_WRITE, &info->lock);
D
dapan1121 已提交
822 823 824
    taosHashRelease(pCatalog->dbCache.cache, info);
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
825
  
D
dapan1121 已提交
826 827 828 829 830 831 832 833 834
  if (info->vgInfo) {
    ctgInfo("cleanup db vgInfo, db:%s", target->dbName);
    taosHashCleanup(info->vgInfo);
    info->vgInfo = NULL;
  }

  if (taosHashRemove(pCatalog->dbCache.cache, target->dbName, strlen(target->dbName))) {
    ctgError("taosHashRemove from dbCache failed, db:%s", target->dbName);
    CTG_UNLOCK(CTG_WRITE, &info->lock);      
D
dapan1121 已提交
835
    taosHashRelease(pCatalog->dbCache.cache, info);
D
dapan1121 已提交
836
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
837
  }
D
dapan1121 已提交
838 839 840 841
  
  CTG_UNLOCK(CTG_WRITE, &info->lock);

  taosHashRelease(pCatalog->dbCache.cache, info);
D
dapan1121 已提交
842

D
dapan1121 已提交
843 844
  *removed = true;
  
D
dapan1121 已提交
845 846 847 848
  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
849 850 851 852 853 854 855 856 857 858 859 860 861 862 863
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)) {
864
    ctgDebug("will renew table meta, supposed to be stable, tbName:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
865 866

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

D
dapan1121 已提交
869
    if (CTG_IS_META_NULL(moutput.metaType)) {
D
dapan1121 已提交
870 871 872 873 874
      CTG_ERR_JRET(ctgGetTableMetaFromVnode(pCatalog, pTransporter, pMgmtEps, pTableName, &vgroupInfo, &voutput));
    } else {
      output = &moutput;
    }
  } else {
875
    ctgDebug("will renew table meta, not supposed to be stable, tbName:%s, isStable:%d", tNameGetTableName(pTableName), isSTable);
D
dapan1121 已提交
876 877

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

D
dapan1121 已提交
880
    if (CTG_IS_META_TABLE(voutput.metaType) && TSDB_SUPER_TABLE == voutput.tbMeta->tableType) {
881
      ctgDebug("will continue to renew table meta since got stable, tbName:%s, metaType:%d", tNameGetTableName(pTableName), voutput.metaType);
D
dapan1121 已提交
882
      
D
dapan1121 已提交
883 884
      CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCatalog, pTransporter, pMgmtEps, voutput.tbFname, &moutput));

D
dapan1121 已提交
885 886
      voutput.metaType = moutput.metaType;
      
D
dapan1121 已提交
887 888 889
      tfree(voutput.tbMeta);
      voutput.tbMeta = moutput.tbMeta;
      moutput.tbMeta = NULL;
D
dapan1121 已提交
890 891 892 893 894 895
    } 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));

D
dapan1121 已提交
896 897
        if (CTG_IS_META_NULL(moutput.metaType)) {
          SET_META_TYPE_NULL(voutput.metaType);
D
dapan1121 已提交
898 899 900 901 902 903
        }
        
        tfree(voutput.tbMeta);
        voutput.tbMeta = moutput.tbMeta;
        moutput.tbMeta = NULL;
      } else {
D
dapan1121 已提交
904 905
        tfree(voutput.tbMeta);
        
D
dapan1121 已提交
906 907
        SET_META_TYPE_CTABLE(voutput.metaType); 
      }
D
dapan1121 已提交
908 909 910
    }
  }

D
dapan1121 已提交
911
  if (CTG_IS_META_NULL(output->metaType)) {
912
    ctgError("no tablemeta got, tbNmae:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
913 914 915
    CTG_ERR_JRET(CTG_ERR_CODE_TABLE_NOT_EXIST);
  }

D
dapan1121 已提交
916 917 918 919 920 921 922 923 924 925
  CTG_ERR_JRET(ctgUpdateTableMetaCache(pCatalog, output));

_return:

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

D
dapan1121 已提交
926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951
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) {
952
    ctgError("renew tablemeta succeed but get from cache failed, may be deleted, tbName:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
953 954 955 956 957 958
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }
  
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008
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 已提交
1009
void ctgFreeHandle(struct SCatalog* pCatalog) {
D
dapan1121 已提交
1010 1011 1012 1013 1014 1015
  ctgFreeMetaRent(&pCatalog->dbRent);
  ctgFreeMetaRent(&pCatalog->stableRent);
  ctgFreeDbCache(&pCatalog->dbCache);
  ctgFreeTableMetaCache(&pCatalog->tableCache);
  
  free(pCatalog);
D
dapan1121 已提交
1016 1017
}

D
dapan1121 已提交
1018
int32_t catalogInit(SCatalogCfg *cfg) {
D
dapan1121 已提交
1019
  if (ctgMgmt.pCluster) {
D
dapan1121 已提交
1020
    qError("catalog already init");
D
dapan1121 已提交
1021
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
1022 1023 1024 1025
  }

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

D
dapan1121 已提交
1027 1028 1029 1030 1031 1032 1033
    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 已提交
1034 1035 1036 1037 1038 1039 1040 1041

    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 已提交
1042 1043 1044
  } else {
    ctgMgmt.cfg.maxDBCacheNum = CTG_DEFAULT_CACHE_DB_NUMBER;
    ctgMgmt.cfg.maxTblCacheNum = CTG_DEFAULT_CACHE_TABLEMETA_NUMBER;
D
dapan1121 已提交
1045 1046
    ctgMgmt.cfg.dbRentSec = CTG_DEFAULT_RENT_SECOND;
    ctgMgmt.cfg.stableRentSec = CTG_DEFAULT_RENT_SECOND;
D
dapan 已提交
1047 1048
  }

D
dapan1121 已提交
1049
  ctgMgmt.pCluster = taosHashInit(CTG_DEFAULT_CACHE_CLUSTER_NUMBER, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
D
dapan1121 已提交
1050
  if (NULL == ctgMgmt.pCluster) {
D
dapan1121 已提交
1051 1052
    qError("taosHashInit %d cluster cache failed", CTG_DEFAULT_CACHE_CLUSTER_NUMBER);
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
1053 1054
  }

D
dapan1121 已提交
1055 1056
  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 已提交
1057
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1058 1059
}

1060 1061
int32_t catalogGetHandle(uint64_t clusterId, struct SCatalog** catalogHandle) {
  if (NULL == catalogHandle) {
D
dapan1121 已提交
1062
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
1063 1064 1065
  }

  if (NULL == ctgMgmt.pCluster) {
D
dapan1121 已提交
1066
    qError("cluster cache are not ready, clusterId:%"PRIx64, clusterId);
D
dapan1121 已提交
1067
    CTG_ERR_RET(TSDB_CODE_CTG_NOT_READY);
D
dapan 已提交
1068 1069
  }

D
dapan1121 已提交
1070 1071
  int32_t code = 0;
  SCatalog *clusterCtg = NULL;
D
dapan 已提交
1072

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

D
dapan1121 已提交
1076 1077 1078 1079 1080
    if (ctg && (*ctg)) {
      *catalogHandle = *ctg;
      qDebug("got catalog handle from cache, clusterId:%"PRIx64", CTG:%p", clusterId, *ctg);
      return TSDB_CODE_SUCCESS;
    }
D
dapan 已提交
1081

D
dapan1121 已提交
1082 1083 1084 1085 1086 1087
    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 已提交
1088 1089
    clusterCtg->clusterId = clusterId;

D
dapan1121 已提交
1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106
    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 已提交
1107
  }
D
dapan1121 已提交
1108 1109

  *catalogHandle = clusterCtg;
D
dapan 已提交
1110
  
D
dapan1121 已提交
1111
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123

_return:

  ctgFreeHandle(clusterCtg);
  
  CTG_RET(code);
}

void catalogFreeHandle(struct SCatalog* pCatalog) {
  if (NULL == pCatalog) {
    return;
  }
D
dapan1121 已提交
1124 1125 1126 1127 1128 1129 1130

  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 已提交
1131 1132
  
  ctgFreeHandle(pCatalog);
D
dapan1121 已提交
1133 1134

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

D
dapan1121 已提交
1137 1138
int32_t catalogGetDBVgroupVersion(struct SCatalog* pCatalog, const char* dbName, int32_t* version) {
  if (NULL == pCatalog || NULL == dbName || NULL == version) {
D
dapan1121 已提交
1139
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
1140 1141 1142 1143
  }

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

D
dapan1121 已提交
1148
  SDBVgroupInfo * dbInfo = taosHashAcquire(pCatalog->dbCache.cache, dbName, strlen(dbName));
D
dapan1121 已提交
1149 1150
  if (NULL == dbInfo) {
    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
1151
    ctgInfo("db not in cache, dbName:%s", dbName);
D
dapan1121 已提交
1152 1153 1154
    return TSDB_CODE_SUCCESS;
  }

1155
  *version = dbInfo->vgVersion;
D
dapan1121 已提交
1156
  taosHashRelease(pCatalog->dbCache.cache, dbInfo);
D
dapan1121 已提交
1157

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

D
dapan1121 已提交
1160 1161 1162
  return TSDB_CODE_SUCCESS;
}

1163
int32_t catalogGetDBVgroup(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, const char* dbName, bool forceUpdate, SArray** vgroupList) {
D
dapan1121 已提交
1164 1165 1166 1167
  if (NULL == pCatalog || NULL == dbName || NULL == pRpc || NULL == pMgmtEps || NULL == vgroupList) {
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }

H
Haojun Liao 已提交
1168
  SDBVgroupInfo* db   = NULL;
D
dapan1121 已提交
1169
  SVgroupInfo *vgInfo = NULL;
1170 1171

  int32_t code = 0;
D
dapan1121 已提交
1172 1173 1174 1175 1176
  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 已提交
1177
    ctgError("taosArrayInit %d failed", taosHashGetSize(db->vgInfo));
D
dapan1121 已提交
1178 1179 1180 1181 1182 1183 1184 1185
    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 已提交
1186
      ctgError("taosArrayPush failed, vgId:%d", vgInfo->vgId);
D
dapan1121 已提交
1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211
      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 已提交
1212
int32_t catalogUpdateDBVgroup(struct SCatalog* pCatalog, const char* dbName, SDBVgroupInfo* dbInfo) {
D
dapan1121 已提交
1213 1214
  int32_t code = 0;
  
D
dapan1121 已提交
1215
  if (NULL == pCatalog || NULL == dbName || NULL == dbInfo) {
D
dapan1121 已提交
1216 1217 1218 1219
    CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
  }

  if (NULL == dbInfo->vgInfo || dbInfo->vgVersion < 0 || taosHashGetSize(dbInfo->vgInfo) <= 0) {
D
dapan1121 已提交
1220
    ctgError("invalid db vgInfo, dbName:%s, vgInfo:%p, vgVersion:%d", dbName, dbInfo->vgInfo, dbInfo->vgVersion);
D
dapan1121 已提交
1221
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1222 1223 1224
  }

  if (NULL == pCatalog->dbCache.cache) {
D
dapan1121 已提交
1225 1226
    SHashObj *cache = taosHashInit(ctgMgmt.cfg.maxDBCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
    if (NULL == cache) {
D
dapan1121 已提交
1227
      ctgError("taosHashInit %d failed", CTG_DEFAULT_CACHE_DB_NUMBER);
D
dapan1121 已提交
1228
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1229
    }
D
dapan1121 已提交
1230 1231 1232 1233

    if (NULL != atomic_val_compare_exchange_ptr(&pCatalog->dbCache.cache, NULL, cache)) {
      taosHashCleanup(cache);
    }
1234
  } else {
D
dapan1121 已提交
1235 1236 1237 1238
    CTG_ERR_JRET(ctgValidateAndFreeDbInfo(pCatalog, dbName, dbInfo));
  }

  bool newAdded = false;
D
dapan1121 已提交
1239 1240
  
  dbInfo->lock = 0;
D
dapan1121 已提交
1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262
  if (taosHashPutExt(pCatalog->dbCache.cache, dbName, strlen(dbName), dbInfo, sizeof(*dbInfo), &newAdded) != 0) {
    ctgError("taosHashPutExt db vgroup to cache failed, db:%s", dbName);
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
  }

  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));
  }
  
  ctgDebug("dbName:%s vgroup updated, vgVersion:%d", dbName, dbInfo->vgVersion);


_return:

  if (dbInfo && dbInfo->vgInfo) {
    taosHashCleanup(dbInfo->vgInfo);
    dbInfo->vgInfo = NULL;
D
dapan1121 已提交
1263
  }
D
dapan1121 已提交
1264 1265 1266 1267 1268 1269 1270
  
  CTG_RET(code);
}


int32_t catalogRemoveDBVgroup(struct SCatalog* pCatalog, SDbVgVersion* dbInfo) {
  int32_t code = 0;
D
dapan1121 已提交
1271
  bool removed = false;
D
dapan1121 已提交
1272 1273
  
  if (NULL == pCatalog || NULL == dbInfo) {
D
dapan1121 已提交
1274
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
1275 1276
  }

D
dapan1121 已提交
1277 1278
  if (NULL == pCatalog->dbCache.cache) {
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1279 1280
  }
  
D
dapan1121 已提交
1281 1282 1283
  CTG_ERR_RET(ctgValidateAndRemoveDbInfo(pCatalog, dbInfo, &removed));
  if (!removed) {
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1284 1285
  }
  
D
dapan1121 已提交
1286
  ctgInfo("db removed from cache, db:%s", dbInfo->dbName);
D
dapan1121 已提交
1287

D
dapan1121 已提交
1288 1289 1290
  CTG_ERR_RET(ctgMetaRentRemove(&pCatalog->dbRent, dbInfo->dbId, ctgDbVgVersionCompare));
  
  ctgDebug("db removed from rent, db:%s", dbInfo->dbName);
D
dapan1121 已提交
1291 1292
  
  CTG_RET(code);
D
dapan1121 已提交
1293 1294
}

D
dapan1121 已提交
1295

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

D
dapan1121 已提交
1300
int32_t catalogGetSTableMeta(struct SCatalog* pCatalog, void * pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta) {
D
dapan1121 已提交
1301
  return ctgGetTableMeta(pCatalog, pTransporter, pMgmtEps, pTableName, false, pTableMeta, 1);
D
dapan1121 已提交
1302 1303 1304
}

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

D
dapan1121 已提交
1309
  return ctgRenewTableMetaImpl(pCatalog, pTransporter, pMgmtEps, pTableName, isSTable);
1310
}
1311

D
dapan1121 已提交
1312
int32_t catalogRenewAndGetTableMeta(struct SCatalog* pCatalog, void *pTransporter, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t isSTable) {
D
dapan1121 已提交
1313
  return ctgGetTableMeta(pCatalog, pTransporter, pMgmtEps, pTableName, true, pTableMeta, isSTable);
D
dapan1121 已提交
1314 1315
}

H
Haojun Liao 已提交
1316 1317
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 已提交
1318
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
1319 1320 1321 1322 1323
  }
  
  STableMeta *tbMeta = NULL;
  int32_t code = 0;
  SVgroupInfo vgroupInfo = {0};
D
dapan1121 已提交
1324 1325 1326 1327
  SDBVgroupInfo* dbVgroup = NULL;
  SArray *vgList = NULL;

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

H
Haojun Liao 已提交
1331 1332
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);
H
Haojun Liao 已提交
1333
  CTG_ERR_JRET(ctgGetDBVgroup(pCatalog, pRpc, pMgmtEps, db, false, &dbVgroup));
D
dapan 已提交
1334

D
dapan 已提交
1335 1336 1337 1338 1339 1340 1341 1342 1343 1344
  // REMOEV THIS ....
  if (0 == tbMeta->vgId) {
    SVgroupInfo vgroup = {0};
    
    catalogGetTableHashVgroup(pCatalog, pRpc, pMgmtEps, pTableName, &vgroup);

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

1345
  if (tbMeta->tableType == TSDB_SUPER_TABLE) {
D
dapan1121 已提交
1346
    CTG_ERR_JRET(ctgGetVgInfoFromDB(pCatalog, pRpc, pMgmtEps, dbVgroup, pVgroupList));
D
dapan1121 已提交
1347
  } else {
1348
    int32_t vgId = tbMeta->vgId;
D
dapan1121 已提交
1349
    if (NULL == taosHashGetClone(dbVgroup->vgInfo, &vgId, sizeof(vgId), &vgroupInfo)) {
1350
      ctgError("table's vgId not found in vgroup list, vgId:%d, tbName:%s", vgId, tNameGetTableName(pTableName));
D
dapan 已提交
1351
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);    
1352
    }
D
dapan1121 已提交
1353

D
dapan1121 已提交
1354 1355
    vgList = taosArrayInit(1, sizeof(SVgroupInfo));
    if (NULL == vgList) {
D
dapan1121 已提交
1356
      ctgError("taosArrayInit %d failed", (int32_t)sizeof(SVgroupInfo));
D
dapan 已提交
1357 1358 1359
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);    
    }

D
dapan1121 已提交
1360
    if (NULL == taosArrayPush(vgList, &vgroupInfo)) {
1361
      ctgError("taosArrayPush vgroupInfo to array failed, vgId:%d, tbName:%s", vgId, tNameGetTableName(pTableName));
D
dapan1121 已提交
1362 1363
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
    }
D
dapan 已提交
1364

D
dapan1121 已提交
1365 1366 1367
    *pVgroupList = vgList;
    vgList = NULL;
  }
D
dapan 已提交
1368

D
dapan1121 已提交
1369 1370
_return:
  tfree(tbMeta);
D
dapan 已提交
1371

D
dapan1121 已提交
1372 1373 1374 1375 1376 1377 1378 1379 1380
  if (dbVgroup) {
    CTG_UNLOCK(CTG_READ, &dbVgroup->lock);
    taosHashRelease(pCatalog->dbCache.cache, dbVgroup);
  }

  if (vgList) {
    taosArrayDestroy(vgList);
    vgList = NULL;
  }
D
dapan1121 已提交
1381
  
D
dapan1121 已提交
1382
  CTG_RET(code);
D
dapan1121 已提交
1383 1384 1385
}


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

H
Haojun Liao 已提交
1390 1391
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);
D
dapan1121 已提交
1392

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

D
dapan1121 已提交
1395
  CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCatalog, dbInfo, pTableName, pVgroup));
D
dapan1121 已提交
1396

D
dapan1121 已提交
1397
_return:
D
dapan1121 已提交
1398

D
dapan1121 已提交
1399 1400 1401 1402
  if (dbInfo) {
    CTG_UNLOCK(CTG_READ, &dbInfo->lock);  
    taosHashRelease(pCatalog->dbCache.cache, dbInfo);
  }
D
dapan1121 已提交
1403

D
dapan1121 已提交
1404
  CTG_RET(code);
D
dapan1121 已提交
1405 1406 1407
}


H
Haojun Liao 已提交
1408 1409
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 已提交
1410
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
1411
  }
D
dapan1121 已提交
1412 1413 1414 1415 1416

  int32_t code = 0;

  if (pReq->pTableName) {
    int32_t tbNum = (int32_t)taosArrayGetSize(pReq->pTableName);
D
dapan1121 已提交
1417
    if (tbNum <= 0) {
D
dapan1121 已提交
1418
      ctgError("empty table name list, tbNum:%d", tbNum);
D
dapan1121 已提交
1419 1420
      CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
    }
H
Haojun Liao 已提交
1421

D
dapan1121 已提交
1422 1423
    pRsp->pTableMeta = taosArrayInit(tbNum, POINTER_BYTES);
    if (NULL == pRsp->pTableMeta) {
D
dapan1121 已提交
1424
      ctgError("taosArrayInit %d failed", tbNum);
D
dapan1121 已提交
1425
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1426 1427 1428 1429 1430 1431
    }
    
    for (int32_t i = 0; i < tbNum; ++i) {
      SName *name = taosArrayGet(pReq->pTableName, i);
      STableMeta *pTableMeta = NULL;
      
H
Haojun Liao 已提交
1432
      CTG_ERR_JRET(ctgGetTableMeta(pCatalog, pTransporter, pMgmtEps, name, false, &pTableMeta, -1));
D
dapan1121 已提交
1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452

      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 已提交
1453
    pRsp->pTableMeta = NULL;
D
dapan1121 已提交
1454
  }
D
dapan 已提交
1455
  
D
dapan1121 已提交
1456
  CTG_RET(code);
1457
}
D
dapan 已提交
1458

D
dapan1121 已提交
1459 1460
int32_t catalogGetQnodeList(struct SCatalog* pCatalog, void *pRpc, const SEpSet* pMgmtEps, SArray* pQnodeList) {
  if (NULL == pCatalog || NULL == pRpc  || NULL == pMgmtEps || NULL == pQnodeList) {
D
dapan 已提交
1461 1462 1463
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
1464
  //TODO
D
dapan 已提交
1465 1466 1467 1468

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484
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 已提交
1485

D
dapan 已提交
1486
void catalogDestroy(void) {
D
dapan1121 已提交
1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500
  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 已提交
1501
  }
D
dapan1121 已提交
1502 1503 1504
  
  taosHashCleanup(ctgMgmt.pCluster);
  ctgMgmt.pCluster = NULL;
D
dapan1121 已提交
1505 1506

  qInfo("catalog destroyed");
D
dapan 已提交
1507 1508 1509 1510
}