catalog.c 74.7 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
dapan 已提交
21 22 23 24 25 26
int32_t ctgActUpdateVg(SCtgMetaAction *action);
int32_t ctgActUpdateTbl(SCtgMetaAction *action);
int32_t ctgActRemoveDB(SCtgMetaAction *action);
int32_t ctgActRemoveStb(SCtgMetaAction *action);
int32_t ctgActRemoveTbl(SCtgMetaAction *action);

D
dapan 已提交
27
SCatalogMgmt gCtgMgmt = {0};
D
dapan1121 已提交
28
SCtgDebug gCTGDebug = {0};
D
dapan 已提交
29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74
SCtgAction gCtgAction[CTG_ACT_MAX] = {{
                            CTG_ACT_UPDATE_VG,
                            "update vgInfo",
                            ctgActUpdateVg
                          },
                          {
                            CTG_ACT_UPDATE_TBL,
                            "update tbMeta",
                            ctgActUpdateTbl
                          },
                          {
                            CTG_ACT_REMOVE_DB,
                            "remove DB",
                            ctgActRemoveDB
                          },
                          {
                            CTG_ACT_REMOVE_STB,
                            "remove stbMeta",
                            ctgActRemoveStb
                          },
                          {
                            CTG_ACT_REMOVE_TBL,
                            "remove tbMeta",
                            ctgActRemoveTbl
                          }
};

int32_t ctgDbgEnableDebug(char *option) {
  if (0 == strcasecmp(option, "lock")) {
    gCTGDebug.lockDebug = true;
    qDebug("lock debug enabled");
    return TSDB_CODE_SUCCESS;
  }

  if (0 == strcasecmp(option, "cache")) {
    gCTGDebug.cacheDebug = true;
    qDebug("cache debug enabled");
    return TSDB_CODE_SUCCESS;
  }

  if (0 == strcasecmp(option, "api")) {
    gCTGDebug.apiDebug = true;
    qDebug("api debug enabled");
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
75 76 77 78 79 80
  if (0 == strcasecmp(option, "meta")) {
    gCTGDebug.metaDebug = true;
    qDebug("api debug enabled");
    return TSDB_CODE_SUCCESS;
  }

D
dapan 已提交
81 82 83 84 85 86 87 88 89 90
  qError("invalid debug option:%s", option);
  
  return TSDB_CODE_CTG_INTERNAL_ERROR;
}

int32_t ctgDbgGetStatNum(char *option, void *res) {
  if (0 == strcasecmp(option, "runtime.qDoneNum")) {
    *(uint64_t *)res = atomic_load_64(&gCtgMgmt.stat.runtime.qDoneNum);
    return TSDB_CODE_SUCCESS;
  }
91

D
dapan 已提交
92 93 94 95
  qError("invalid stat option:%s", option);
  
  return TSDB_CODE_CTG_INTERNAL_ERROR;
}
D
dapan1121 已提交
96

D
dapan1121 已提交
97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118
int32_t ctgDbgGetTbMetaNum(SCtgDBCache *dbCache) {
  return dbCache->tbCache.metaCache ? (int32_t)taosHashGetSize(dbCache->tbCache.metaCache) : 0;
}

int32_t ctgDbgGetStbNum(SCtgDBCache *dbCache) {
  return dbCache->tbCache.stbCache ? (int32_t)taosHashGetSize(dbCache->tbCache.stbCache) : 0;
}

int32_t ctgDbgGetRentNum(SCtgRentMgmt *rent) {
  int32_t num = 0;
  for (uint16_t i = 0; i < rent->slotNum; ++i) {
    SCtgRentSlot *slot = &rent->slots[i];
    if (NULL == slot->meta) {
      continue;
    }

    num += taosArrayGetSize(slot->meta);
  }

  return num;
}

D
dapan1121 已提交
119 120
int32_t ctgDbgGetClusterCacheNum(SCatalog* pCtg, int32_t type) {
  if (NULL == pCtg || NULL == pCtg->dbCache) {
D
dapan1121 已提交
121 122 123 124 125
    return 0;
  }

  switch (type) {
    case CTG_DBG_DB_NUM:
D
dapan1121 已提交
126
      return (int32_t)taosHashGetSize(pCtg->dbCache);
D
dapan1121 已提交
127
    case CTG_DBG_DB_RENT_NUM:
D
dapan1121 已提交
128
      return ctgDbgGetRentNum(&pCtg->dbRent);
D
dapan1121 已提交
129
    case CTG_DBG_STB_RENT_NUM:
D
dapan1121 已提交
130
      return ctgDbgGetRentNum(&pCtg->stbRent);
D
dapan1121 已提交
131 132 133 134 135 136
    default:
      break;
  }

  SCtgDBCache *dbCache = NULL;
  int32_t num = 0;
D
dapan1121 已提交
137
  void *pIter = taosHashIterate(pCtg->dbCache, NULL);
D
dapan1121 已提交
138 139 140 141 142 143 144 145 146 147 148 149 150
  while (pIter) {
    dbCache = (SCtgDBCache *)pIter;
    switch (type) {
      case CTG_DBG_META_NUM:
        num += ctgDbgGetTbMetaNum(dbCache);
        break;
      case CTG_DBG_STB_NUM:
        num += ctgDbgGetStbNum(dbCache);
        break;
      default:
        ctgError("invalid type:%d", type);
        break;
    }
D
dapan1121 已提交
151
    pIter = taosHashIterate(pCtg->dbCache, pIter);
D
dapan1121 已提交
152 153 154 155 156
  }

  return num;
}

D
dapan1121 已提交
157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177
void ctgDbgShowTableMeta(SCatalog* pCtg, const char *tbName, STableMeta* p) {
  if (!gCTGDebug.metaDebug) {
    return;
  }

  STableComInfo *c = &p->tableInfo;

  if (TSDB_CHILD_TABLE == p->tableType) {
    ctgDebug("table [%s] meta: type:%d, vgId:%d, uid:%" PRIx64 ",suid:%" PRIx64, tbName, p->tableType, p->vgId, p->uid, p->suid);
    return;
  } else {
    ctgDebug("table [%s] meta: type:%d, vgId:%d, uid:%" PRIx64 ",suid:%" PRIx64 ",sv:%d, tv:%d, tagNum:%d, precision:%d, colNum:%d, rowSize:%d",
     tbName, p->tableType, p->vgId, p->uid, p->suid, p->sversion, p->tversion, c->numOfTags, c->precision, c->numOfColumns, c->rowSize);
  }
  
  int32_t colNum = c->numOfColumns + c->numOfTags;
  for (int32_t i = 0; i < colNum; ++i) {
    SSchema *s = &p->schema[i];
    ctgDebug("[%d] name:%s, type:%d, colId:%d, bytes:%d", i, s->name, s->type, s->colId, s->bytes);
  }
}
D
dapan1121 已提交
178

D
dapan1121 已提交
179 180
void ctgDbgShowDBCache(SCatalog* pCtg, SHashObj *dbHash) {
  if (NULL == dbHash || !gCTGDebug.cacheDebug) {
D
dapan1121 已提交
181 182 183 184 185 186 187 188 189 190 191 192
    return;
  }

  int32_t i = 0;
  SCtgDBCache *dbCache = NULL;
  void *pIter = taosHashIterate(dbHash, NULL);
  while (pIter) {
    char *dbFName = NULL;
    size_t len = 0;
    
    dbCache = (SCtgDBCache *)pIter;

D
dapan1121 已提交
193
    taosHashGetKey(dbCache, (void **)&dbFName, &len);
D
dapan1121 已提交
194

D
dapan1121 已提交
195 196 197 198 199 200
    int32_t metaNum = dbCache->tbCache.metaCache ? taosHashGetSize(dbCache->tbCache.metaCache) : 0;
    int32_t stbNum = dbCache->tbCache.stbCache ? taosHashGetSize(dbCache->tbCache.stbCache) : 0;
    int32_t vgVersion = CTG_DEFAULT_INVALID_VERSION;
    int32_t hashMethod = -1;
    int32_t vgNum = 0;

D
dapan1121 已提交
201
    if (dbCache->vgInfo) {
D
dapan1121 已提交
202 203
      vgVersion = dbCache->vgInfo->vgVersion;
      hashMethod = dbCache->vgInfo->hashMethod;
D
dapan1121 已提交
204
      if (dbCache->vgInfo->vgHash) {
D
dapan1121 已提交
205
        vgNum = taosHashGetSize(dbCache->vgInfo->vgHash);
D
dapan1121 已提交
206
      }
D
dapan1121 已提交
207
    }    
D
dapan1121 已提交
208
    
D
dapan1121 已提交
209 210 211
    ctgDebug("[%d] db [%.*s][%"PRIx64"] %s: metaNum:%d, stbNum:%d, vgVersion:%d, hashMethod:%d, vgNum:%d", 
      i, (int32_t)len, dbFName, dbCache->dbId, dbCache->deleted?"deleted":"", metaNum, stbNum, vgVersion, hashMethod, vgNum);

D
dapan1121 已提交
212 213 214 215
    pIter = taosHashIterate(dbHash, pIter);
  }
}

D
dapan1121 已提交
216 217 218



D
dapan1121 已提交
219
void ctgDbgShowClusterCache(SCatalog* pCtg) {
D
dapan1121 已提交
220
  if (!gCTGDebug.cacheDebug || NULL == pCtg) {
D
dapan1121 已提交
221 222 223
    return;
  }

D
dapan1121 已提交
224 225
  ctgDebug("## cluster %"PRIx64" %p cache Info ##", pCtg->clusterId, pCtg);
  ctgDebug("db:%d meta:%d stb:%d dbRent:%d stbRent:%d", ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_DB_NUM), ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_META_NUM), 
D
dapan1121 已提交
226
    ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_STB_NUM), ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_DB_RENT_NUM), ctgDbgGetClusterCacheNum(pCtg, CTG_DBG_STB_RENT_NUM));    
D
dapan1121 已提交
227
  
D
dapan1121 已提交
228
  ctgDbgShowDBCache(pCtg, pCtg->dbCache);
D
dapan1121 已提交
229 230
}

D
dapan 已提交
231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261


void ctgPopAction(SCtgMetaAction **action) {
  SCtgQNode *orig = gCtgMgmt.head;
  
  SCtgQNode *node = gCtgMgmt.head->next;
  gCtgMgmt.head = gCtgMgmt.head->next;

  CTG_QUEUE_SUB();
  
  tfree(orig);

  *action = &node->action;
}


int32_t ctgPushAction(SCtgMetaAction *action) {
  SCtgQNode *node = calloc(1, sizeof(SCtgQNode));
  if (NULL == node) {
    qError("calloc %d failed", (int32_t)sizeof(SCtgQNode));
    CTG_RET(TSDB_CODE_CTG_MEM_ERROR);
  }
  
  node->action = *action;

  CTG_LOCK(CTG_WRITE, &gCtgMgmt.qlock);
  gCtgMgmt.tail->next = node;
  gCtgMgmt.tail = node;
  CTG_UNLOCK(CTG_WRITE, &gCtgMgmt.qlock);

  CTG_QUEUE_ADD();
D
dapan1121 已提交
262
  CTG_STAT_ADD(gCtgMgmt.stat.runtime.qNum);
D
dapan 已提交
263 264 265 266 267 268 269

  tsem_post(&gCtgMgmt.sem);

  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286
void ctgFreeMetaRent(SCtgRentMgmt *mgmt) {
  if (NULL == mgmt->slots) {
    return;
  }

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

  tfree(mgmt->slots);
}


D
dapan1121 已提交
287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314
int32_t ctgPushRmDBMsgInQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId) {
  int32_t code = 0;
  SCtgMetaAction action= {.act = CTG_ACT_REMOVE_DB};
  SCtgRemoveDBMsg *msg = malloc(sizeof(SCtgRemoveDBMsg));
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgRemoveDBMsg));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

  msg->pCtg = pCtg;
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  msg->dbId = dbId;

  action.data = msg;

  CTG_ERR_JRET(ctgPushAction(&action));

  ctgDebug("action [%s] added into queue", gCtgAction[action.act].name);

  return TSDB_CODE_SUCCESS;

_return:
  
  tfree(action.data);
  CTG_RET(code);  
}


D
dapan1121 已提交
315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374
int32_t ctgPushRmStbMsgInQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid) {
  int32_t code = 0;
  SCtgMetaAction action= {.act = CTG_ACT_REMOVE_STB};
  SCtgRemoveStbMsg *msg = malloc(sizeof(SCtgRemoveStbMsg));
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgRemoveStbMsg));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

  msg->pCtg = pCtg;
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  strncpy(msg->stbName, stbName, sizeof(msg->stbName));
  msg->dbId = dbId;
  msg->suid = suid;

  action.data = msg;

  CTG_ERR_JRET(ctgPushAction(&action));

  ctgDebug("action [%s] added into queue", gCtgAction[action.act].name);

  return TSDB_CODE_SUCCESS;

_return:
  
  tfree(action.data);
  CTG_RET(code);  
}



int32_t ctgPushRmTblMsgInQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *tbName) {
  int32_t code = 0;
  SCtgMetaAction action= {.act = CTG_ACT_REMOVE_TBL};
  SCtgRemoveTblMsg *msg = malloc(sizeof(SCtgRemoveTblMsg));
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgRemoveTblMsg));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

  msg->pCtg = pCtg;
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  strncpy(msg->tbName, tbName, sizeof(msg->tbName));
  msg->dbId = dbId;

  action.data = msg;

  CTG_ERR_JRET(ctgPushAction(&action));

  ctgDebug("action [%s] added into queue", gCtgAction[action.act].name);

  return TSDB_CODE_SUCCESS;

_return:
  
  tfree(action.data);
  CTG_RET(code);  
}


D
dapan1121 已提交
375 376 377 378 379
void ctgFreeTableMetaCache(SCtgTbMetaCache *cache) {
  CTG_LOCK(CTG_WRITE, &cache->stbLock);
  if (cache->stbCache) {
    taosHashCleanup(cache->stbCache);
    cache->stbCache = NULL;
D
dapan1121 已提交
380
  }
D
dapan1121 已提交
381
  CTG_UNLOCK(CTG_WRITE, &cache->stbLock);
D
dapan1121 已提交
382

D
dapan1121 已提交
383 384 385 386
  CTG_LOCK(CTG_WRITE, &cache->metaLock);
  if (cache->metaCache) {
    taosHashCleanup(cache->metaCache);
    cache->metaCache = NULL;
D
dapan1121 已提交
387
  }
D
dapan1121 已提交
388
  CTG_UNLOCK(CTG_WRITE, &cache->metaLock);
D
dapan1121 已提交
389 390
}

D
dapan1121 已提交
391 392 393 394 395 396 397 398 399 400 401 402 403
void ctgFreeVgInfo(SDBVgInfo *vgInfo) {
  if (NULL == vgInfo) {
    return;
  }

  if (vgInfo->vgHash) {
    taosHashCleanup(vgInfo->vgHash);
    vgInfo->vgHash = NULL;
  }
  
  tfree(vgInfo);
}

D
dapan1121 已提交
404 405 406 407 408
void ctgFreeDbCache(SCtgDBCache *dbCache) {
  if (NULL == dbCache) {
    return;
  }

D
dapan1121 已提交
409
  CTG_LOCK(CTG_WRITE, &dbCache->vgLock);
D
dapan1121 已提交
410
  ctgFreeVgInfo (dbCache->vgInfo);
D
dapan1121 已提交
411
  CTG_UNLOCK(CTG_WRITE, &dbCache->vgLock);
D
dapan1121 已提交
412 413 414 415 416

  ctgFreeTableMetaCache(&dbCache->tbCache);
}


D
dapan1121 已提交
417 418 419 420 421 422
void ctgFreeHandle(SCatalog* pCtg) {
  ctgFreeMetaRent(&pCtg->dbRent);
  ctgFreeMetaRent(&pCtg->stbRent);
  
  if (pCtg->dbCache) {
    void *pIter = taosHashIterate(pCtg->dbCache, NULL);
D
dapan1121 已提交
423 424 425
    while (pIter) {
      SCtgDBCache *dbCache = pIter;

D
dapan1121 已提交
426 427
      atomic_store_8(&dbCache->deleted, 1);

D
dapan1121 已提交
428
      ctgFreeDbCache(dbCache);
D
dapan1121 已提交
429 430
            
      pIter = taosHashIterate(pCtg->dbCache, pIter);
D
dapan1121 已提交
431 432
    }  

D
dapan1121 已提交
433
    taosHashCleanup(pCtg->dbCache);
D
dapan1121 已提交
434 435
  }
  
D
dapan1121 已提交
436
  free(pCtg);
D
dapan1121 已提交
437 438
}

D
dapan1121 已提交
439

D
dapan 已提交
440
int32_t ctgAcquireVgInfo(SCatalog *pCtg, SCtgDBCache *dbCache, bool *inCache) {
D
dapan1121 已提交
441 442
  CTG_LOCK(CTG_READ, &dbCache->vgLock);
  
D
dapan 已提交
443 444 445 446 447 448 449 450 451 452
  if (dbCache->deleted) {
    CTG_UNLOCK(CTG_READ, &dbCache->vgLock);

    ctgDebug("db is dropping, dbId:%"PRIx64, dbCache->dbId);
    
    *inCache = false;
    return TSDB_CODE_SUCCESS;
  }

  
D
dapan1121 已提交
453 454 455
  if (NULL == dbCache->vgInfo) {
    CTG_UNLOCK(CTG_READ, &dbCache->vgLock);

D
dapan 已提交
456
    *inCache = false;
D
dapan1121 已提交
457
    ctgDebug("db vgInfo is empty, dbId:%"PRIx64, dbCache->dbId);
D
dapan1121 已提交
458 459 460
    return TSDB_CODE_SUCCESS;
  }

D
dapan 已提交
461 462
  *inCache = true;
  
D
dapan1121 已提交
463 464 465 466 467 468 469 470 471 472 473 474 475 476
  return TSDB_CODE_SUCCESS;
}

int32_t ctgWAcquireVgInfo(SCatalog *pCtg, SCtgDBCache *dbCache) {
  CTG_LOCK(CTG_WRITE, &dbCache->vgLock);

  if (dbCache->deleted) {
    ctgDebug("db is dropping, dbId:%"PRIx64, dbCache->dbId);
    CTG_UNLOCK(CTG_WRITE, &dbCache->vgLock);
    CTG_ERR_RET(TSDB_CODE_CTG_DB_DROPPED);
  }

  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
477

D
dapan1121 已提交
478 479 480
void ctgReleaseDBCache(SCatalog *pCtg, SCtgDBCache *dbCache) {
  taosHashRelease(pCtg->dbCache, dbCache);
}
D
dapan1121 已提交
481

D
dapan1121 已提交
482 483 484
void ctgReleaseVgInfo(SCtgDBCache *dbCache) {
  CTG_UNLOCK(CTG_READ, &dbCache->vgLock);
}
D
dapan1121 已提交
485

D
dapan1121 已提交
486 487 488
void ctgWReleaseVgInfo(SCtgDBCache *dbCache) {
  CTG_UNLOCK(CTG_WRITE, &dbCache->vgLock);
}
D
dapan1121 已提交
489

D
dapan1121 已提交
490

D
dapan 已提交
491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528
int32_t ctgAcquireDBCacheImpl(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache, bool acquire) {
  SCtgDBCache *dbCache = NULL;
  if (acquire) {
    dbCache = (SCtgDBCache *)taosHashAcquire(pCtg->dbCache, dbFName, strlen(dbFName));
  } else {
    dbCache = (SCtgDBCache *)taosHashGet(pCtg->dbCache, dbFName, strlen(dbFName));
  }
  
  if (NULL == dbCache) {
    *pCache = NULL;
    ctgDebug("db not in cache, dbFName:%s", dbFName);
    return TSDB_CODE_SUCCESS;
  }

  if (dbCache->deleted) {
    if (acquire) {
      ctgReleaseDBCache(pCtg, dbCache);
    }    
    
    *pCache = NULL;
    ctgDebug("db is removing from cache, dbFName:%s", dbFName);
    return TSDB_CODE_SUCCESS;
  }

  *pCache = dbCache;
    
  return TSDB_CODE_SUCCESS;
}

int32_t ctgAcquireDBCache(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache) {
  CTG_RET(ctgAcquireDBCacheImpl(pCtg, dbFName, pCache, true));
}

int32_t ctgGetDBCache(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache) {
  CTG_RET(ctgAcquireDBCacheImpl(pCtg, dbFName, pCache, false));
}


D
dapan1121 已提交
529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544
int32_t ctgAcquireVgInfoFromCache(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache, bool *inCache) {
  if (NULL == pCtg->dbCache) {
    *pCache = NULL;
    *inCache = false;
    ctgWarn("empty db cache, dbFName:%s", dbFName);
    return TSDB_CODE_SUCCESS;
  }

  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
  if (NULL == dbCache) {  
    *pCache = NULL;
    *inCache = false;
    return TSDB_CODE_SUCCESS;
  }
  
D
dapan 已提交
545 546
  ctgAcquireVgInfo(pCtg, dbCache, inCache);
  if (!(*inCache)) {
D
dapan1121 已提交
547 548 549 550
    ctgReleaseDBCache(pCtg, dbCache);
  
    *pCache = NULL;
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
551
  }
D
dapan1121 已提交
552

D
dapan1121 已提交
553
  *pCache = dbCache;
D
dapan1121 已提交
554
  *inCache = true;
D
dapan1121 已提交
555

D
dapan1121 已提交
556
  ctgDebug("Got db vgInfo from cache, dbFName:%s", dbFName);
D
dapan1121 已提交
557 558 559 560 561 562
  
  return TSDB_CODE_SUCCESS;
}



D
dapan1121 已提交
563
int32_t ctgGetDBVgInfoFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SBuildUseDBInput *input, SUseDbOutput *out) {
D
dapan1121 已提交
564 565 566
  char *msg = NULL;
  int32_t msgLen = 0;

D
dapan 已提交
567
  ctgDebug("try to get db vgInfo from mnode, dbFName:%s", input->db);
D
dapan1121 已提交
568 569 570 571 572 573

  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 已提交
574
  
D
dapan1121 已提交
575
  SRpcMsg rpcMsg = {
H
Hongze Cheng 已提交
576
      .msgType = TDMT_MND_USE_DB,
D
catalog  
dapan1121 已提交
577
      .pCont   = msg,
D
dapan1121 已提交
578 579 580 581 582 583
      .contLen = msgLen,
  };

  SRpcMsg rpcRsp = {0};

  rpcSendRecv(pRpc, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
D
dapan1121 已提交
584
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
D
dapan1121 已提交
585
    ctgError("error rsp for use db, error:%s, db:%s", tstrerror(rpcRsp.code), input->db);
D
dapan1121 已提交
586
    CTG_ERR_RET(rpcRsp.code);
D
dapan1121 已提交
587
  }
D
dapan1121 已提交
588

D
dapan1121 已提交
589 590 591 592 593
  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 已提交
594

D
dapan 已提交
595 596
  ctgDebug("Got db vgInfo from mnode, dbFName:%s", input->db);

D
dapan1121 已提交
597 598
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
599

D
dapan1121 已提交
600 601
int32_t ctgIsTableMetaExistInCache(SCatalog* pCtg, char *dbFName, char* tbName, int32_t *exist) {
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
602
    *exist = 0;
D
dapan1121 已提交
603 604 605 606
    ctgWarn("empty db cache, dbFName:%s, tbName:%s", dbFName, tbName);
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
607 608
  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
609 610
  if (NULL == dbCache) {
    *exist = 0;
D
dapan1121 已提交
611 612 613
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
614
  size_t sz = 0;
D
dapan1121 已提交
615 616 617 618
  CTG_LOCK(CTG_READ, &dbCache->tbCache.metaLock);  
  STableMeta *tbMeta = taosHashGet(dbCache->tbCache.metaCache, tbName, strlen(tbName));
  CTG_UNLOCK(CTG_READ, &dbCache->tbCache.metaLock);
  
D
dapan1121 已提交
619
  if (NULL == tbMeta) {
D
dapan 已提交
620
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
621
    
D
dapan1121 已提交
622
    *exist = 0;
D
dapan1121 已提交
623
    ctgDebug("tbmeta not in cache, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
624 625 626 627
    return TSDB_CODE_SUCCESS;
  }

  *exist = 1;
D
dapan1121 已提交
628

D
dapan 已提交
629
  ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
630
  
D
dapan1121 已提交
631
  ctgDebug("tbmeta is in cache, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
632 633 634 635
  
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
636

D
dapan1121 已提交
637
int32_t ctgGetTableMetaFromCache(SCatalog* pCtg, const SName* pTableName, STableMeta** pTableMeta, int32_t *exist, int32_t flag, uint64_t *dbId) {
D
dapan1121 已提交
638
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
639
    *exist = 0;
D
dapan1121 已提交
640
    ctgWarn("empty tbmeta cache, tbName:%s", pTableName->tname);
D
dapan1121 已提交
641 642 643
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
644
  char dbFName[TSDB_DB_FNAME_LEN] = {0};
D
dapan1121 已提交
645
  if (CTG_FLAG_IS_INF_DB(flag)) {
D
dapan1121 已提交
646 647 648 649
    strcpy(dbFName, pTableName->dbname);
  } else {
    tNameGetFullDbName(pTableName, dbFName);
  }
D
dapan1121 已提交
650

D
dapan1121 已提交
651 652
  *pTableMeta = NULL;

D
dapan1121 已提交
653 654
  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
655 656 657 658 659
  if (NULL == dbCache) {
    *exist = 0;
    return TSDB_CODE_SUCCESS;
  }
  
D
dapan1121 已提交
660 661 662 663 664
  size_t sz = 0;  
  CTG_LOCK(CTG_READ, &dbCache->tbCache.metaLock);
  STableMeta *tbMeta = taosHashGetCloneExt(dbCache->tbCache.metaCache, pTableName->tname, strlen(pTableName->tname), NULL, (void **)pTableMeta, &sz);
  CTG_UNLOCK(CTG_READ, &dbCache->tbCache.metaLock);

D
dapan1121 已提交
665
  if (NULL == *pTableMeta) {
D
dapan1121 已提交
666
    *exist = 0;
D
dapan1121 已提交
667 668
    ctgReleaseDBCache(pCtg, dbCache);
    ctgDebug("tbl not in cache, dbFName:%s, tbName:%s", dbFName, pTableName->tname);
D
dapan1121 已提交
669 670 671
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
672
  *exist = 1;
D
dapan1121 已提交
673 674 675
  if (dbId) {
    *dbId = dbCache->dbId;
  }
D
dapan1121 已提交
676 677
  
  tbMeta = *pTableMeta;
D
dapan1121 已提交
678

D
dapan1121 已提交
679
  if (tbMeta->tableType != TSDB_CHILD_TABLE) {
D
dapan1121 已提交
680
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
681
    ctgDebug("Got meta from cache, type:%d, dbFName:%s, tbName:%s", tbMeta->tableType, dbFName, pTableName->tname);
D
dapan1121 已提交
682 683 684
    return TSDB_CODE_SUCCESS;
  }
  
D
dapan1121 已提交
685
  CTG_LOCK(CTG_READ, &dbCache->tbCache.stbLock);
D
dapan1121 已提交
686
  
D
dapan1121 已提交
687
  STableMeta **stbMeta = taosHashGet(dbCache->tbCache.stbCache, &tbMeta->suid, sizeof(tbMeta->suid));
D
dapan1121 已提交
688
  if (NULL == stbMeta || NULL == *stbMeta) {
D
dapan1121 已提交
689
    CTG_UNLOCK(CTG_READ, &dbCache->tbCache.stbLock);
D
dapan1121 已提交
690 691
    ctgReleaseDBCache(pCtg, dbCache);
    ctgError("stb not in stbCache, suid:%"PRIx64, tbMeta->suid);
D
dapan1121 已提交
692 693 694 695
    tfree(*pTableMeta);
    *exist = 0;
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
696

D
dapan1121 已提交
697
  if ((*stbMeta)->suid != tbMeta->suid) {    
D
dapan1121 已提交
698
    CTG_UNLOCK(CTG_READ, &dbCache->tbCache.stbLock);
D
dapan1121 已提交
699
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
700
    tfree(*pTableMeta);
D
dapan1121 已提交
701
    ctgError("stable suid in stbCache mis-match, expected suid:%"PRIx64 ",actual suid:%"PRIx64, tbMeta->suid, (*stbMeta)->suid);
D
dapan1121 已提交
702 703
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }
D
dapan1121 已提交
704

D
dapan1121 已提交
705
  int32_t metaSize = CTG_META_SIZE(*stbMeta);
D
dapan1121 已提交
706 707
  *pTableMeta = realloc(*pTableMeta, metaSize);
  if (NULL == *pTableMeta) {    
D
dapan1121 已提交
708
    CTG_UNLOCK(CTG_READ, &dbCache->tbCache.stbLock);
D
dapan1121 已提交
709
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
710
    ctgError("realloc size[%d] failed", metaSize);
D
dapan1121 已提交
711
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
712 713
  }

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

D
dapan1121 已提交
716
  CTG_UNLOCK(CTG_READ, &dbCache->tbCache.stbLock);
D
dapan1121 已提交
717

D
dapan1121 已提交
718
  ctgReleaseDBCache(pCtg, dbCache);
D
dapan 已提交
719

D
dapan1121 已提交
720
  ctgDebug("Got tbmeta from cache, dbFName:%s, tbName:%s", dbFName, pTableName->tname);
D
dapan1121 已提交
721 722 723 724
  
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
725
int32_t ctgGetTableTypeFromCache(SCatalog* pCtg, const SName* pTableName, int32_t *tbType, int32_t flag) {
D
dapan1121 已提交
726
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
727
    ctgWarn("empty db cache, tbName:%s", pTableName->tname);  
D
dapan1121 已提交
728 729 730
    return TSDB_CODE_SUCCESS;
  }

D
dapan 已提交
731
  char dbFName[TSDB_DB_FNAME_LEN] = {0};
D
dapan1121 已提交
732
  if (CTG_FLAG_IS_INF_DB(flag)) {
D
dapan1121 已提交
733 734 735 736 737
    strcpy(dbFName, pTableName->dbname);
  } else {
    tNameGetFullDbName(pTableName, dbFName);
  }
  
D
dapan 已提交
738 739
  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
740 741 742
  if (NULL == dbCache) {
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
743

D
dapan1121 已提交
744 745 746
  CTG_LOCK(CTG_READ, &dbCache->tbCache.metaLock);
  STableMeta *pTableMeta = (STableMeta *)taosHashAcquire(dbCache->tbCache.metaCache, pTableName->tname, strlen(pTableName->tname));

D
dapan1121 已提交
747
  if (NULL == pTableMeta) {
D
dapan1121 已提交
748
    CTG_UNLOCK(CTG_READ, &dbCache->tbCache.metaLock);
D
dapan 已提交
749 750
    ctgWarn("tbl not in cache, dbFName:%s, tbName:%s", dbFName, pTableName->tname);  
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
751
    
D
dapan1121 已提交
752 753 754
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
755 756
  *tbType = atomic_load_8(&pTableMeta->tableType);

D
dapan1121 已提交
757 758 759 760
  taosHashRelease(dbCache->tbCache.metaCache, pTableMeta);

  CTG_UNLOCK(CTG_READ, &dbCache->tbCache.metaLock);

D
dapan 已提交
761
  ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
762

D
dapan 已提交
763
  ctgDebug("Got tbtype from cache, dbFName:%s, tbName:%s, type:%d", dbFName, pTableName->tname, *tbType);  
D
dapan1121 已提交
764 765 766 767
  
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
768
int32_t ctgGetTableMetaFromMnodeImpl(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, char *dbFName, char* tbName, STableMetaOutput* output) {
D
dapan1121 已提交
769
  SBuildTableMetaInput bInput = {.vgId = 0, .dbFName = dbFName, .tbName = tbName};
D
dapan1121 已提交
770 771 772 773
  char *msg = NULL;
  SEpSet *pVnodeEpSet = NULL;
  int32_t msgLen = 0;

D
dapan1121 已提交
774
  ctgDebug("try to get table meta from mnode, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
775

D
dapan1121 已提交
776
  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_MND_TABLE_META)](&bInput, &msg, 0, &msgLen);
D
dapan1121 已提交
777 778 779 780
  if (code) {
    ctgError("Build mnode stablemeta msg failed, code:%x", code);
    CTG_ERR_RET(code);
  }
D
dapan1121 已提交
781 782

  SRpcMsg rpcMsg = {
D
dapan1121 已提交
783
      .msgType = TDMT_MND_TABLE_META,
D
dapan1121 已提交
784 785 786
      .pCont   = msg,
      .contLen = msgLen,
  };
D
dapan1121 已提交
787

D
dapan1121 已提交
788 789
  SRpcMsg rpcRsp = {0};

D
dapan1121 已提交
790
  rpcSendRecv(pTrans, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
D
dapan1121 已提交
791 792
  
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
D
dapan1121 已提交
793
    if (CTG_TABLE_NOT_EXIST(rpcRsp.code)) {
D
dapan1121 已提交
794
      SET_META_TYPE_NULL(output->metaType);
D
dapan1121 已提交
795
      ctgDebug("stablemeta not exist in mnode, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
796 797 798
      return TSDB_CODE_SUCCESS;
    }
    
D
dapan1121 已提交
799
    ctgError("error rsp for stablemeta from mnode, code:%s, dbFName:%s, tbName:%s", tstrerror(rpcRsp.code), dbFName, tbName);
D
dapan1121 已提交
800 801 802
    CTG_ERR_RET(rpcRsp.code);
  }

D
dapan1121 已提交
803
  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_TABLE_META)](output, rpcRsp.pCont, rpcRsp.contLen);
D
dapan1121 已提交
804
  if (code) {
D
dapan1121 已提交
805
    ctgError("Process mnode stablemeta rsp failed, code:%x, dbFName:%s, tbName:%s", code, dbFName, tbName);
D
dapan1121 已提交
806 807 808
    CTG_ERR_RET(code);
  }

D
dapan1121 已提交
809
  ctgDebug("Got table meta from mnode, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
810 811 812 813

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
814
int32_t ctgGetTableMetaFromMnode(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, STableMetaOutput* output) {
D
dapan1121 已提交
815 816
  char dbFName[TSDB_DB_FNAME_LEN];
  tNameGetFullDbName(pTableName, dbFName);
D
dapan1121 已提交
817

D
dapan1121 已提交
818
  return ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, dbFName, (char *)pTableName->tname, output);
D
dapan1121 已提交
819
}
D
dapan1121 已提交
820

D
dapan1121 已提交
821 822
int32_t ctgGetTableMetaFromVnode(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, SVgroupInfo *vgroupInfo, STableMetaOutput* output) {
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pTableName || NULL == vgroupInfo || NULL == output) {
D
dapan1121 已提交
823
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
824 825
  }

D
dapan1121 已提交
826 827
  char dbFName[TSDB_DB_FNAME_LEN];
  tNameGetFullDbName(pTableName, dbFName);
D
dapan1121 已提交
828

D
dapan1121 已提交
829
  ctgDebug("try to get table meta from vnode, dbFName:%s, tbName:%s", dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
830

D
dapan1121 已提交
831
  SBuildTableMetaInput bInput = {.vgId = vgroupInfo->vgId, .dbFName = dbFName, .tbName = (char *)tNameGetTableName(pTableName)};
D
dapan1121 已提交
832 833 834
  char *msg = NULL;
  int32_t msgLen = 0;

D
dapan1121 已提交
835 836
  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_VND_TABLE_META)](&bInput, &msg, 0, &msgLen);
  if (code) {
D
dapan1121 已提交
837
    ctgError("Build vnode tablemeta msg failed, code:%x, dbFName:%s, tbName:%s", code, dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
838 839
    CTG_ERR_RET(code);
  }
D
dapan1121 已提交
840 841

  SRpcMsg rpcMsg = {
H
Hongze Cheng 已提交
842
      .msgType = TDMT_VND_TABLE_META,
D
dapan1121 已提交
843 844 845 846 847
      .pCont   = msg,
      .contLen = msgLen,
  };

  SRpcMsg rpcRsp = {0};
D
dapan1121 已提交
848
  rpcSendRecv(pTrans, &vgroupInfo->epset, &rpcMsg, &rpcRsp);
D
dapan1121 已提交
849
  
D
dapan1121 已提交
850
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
D
dapan1121 已提交
851
    if (CTG_TABLE_NOT_EXIST(rpcRsp.code)) {
D
dapan1121 已提交
852
      SET_META_TYPE_NULL(output->metaType);
D
dapan1121 已提交
853
      ctgDebug("tablemeta not exist in vnode, dbFName:%s, tbName:%s", dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
854 855 856
      return TSDB_CODE_SUCCESS;
    }
  
D
dapan1121 已提交
857
    ctgError("error rsp for table meta from vnode, code:%s, dbFName:%s, tbName:%s", tstrerror(rpcRsp.code), dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
858
    CTG_ERR_RET(rpcRsp.code);
D
dapan1121 已提交
859 860
  }

D
dapan1121 已提交
861 862
  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_VND_TABLE_META)](output, rpcRsp.pCont, rpcRsp.contLen);
  if (code) {
D
dapan1121 已提交
863
    ctgError("Process vnode tablemeta rsp failed, code:%s, dbFName:%s, tbName:%s", tstrerror(code), dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
864 865 866
    CTG_ERR_RET(code);
  }

D
dapan1121 已提交
867
  ctgDebug("Got table meta from vnode, dbFName:%s, tbName:%s", dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
868 869 870 871
  return TSDB_CODE_SUCCESS;
}


872 873
int32_t ctgGetHashFunction(int8_t hashMethod, tableNameHashFp *fp) {
  switch (hashMethod) {
D
dapan1121 已提交
874 875 876 877 878 879 880 881
    default:
      *fp = MurmurHash3_32;
      break;
  }

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
882
int32_t ctgGenerateVgList(SCatalog *pCtg, SHashObj *vgHash, SArray** pList) {
D
dapan1121 已提交
883
  SHashObj *vgroupHash = NULL;
884
  SVgroupInfo *vgInfo = NULL;
D
dapan1121 已提交
885 886
  SArray *vgList = NULL;
  int32_t code = 0;
D
dapan1121 已提交
887
  int32_t vgNum = taosHashGetSize(vgHash);
888

D
dapan1121 已提交
889
  vgList = taosArrayInit(vgNum, sizeof(SVgroupInfo));
D
dapan1121 已提交
890
  if (NULL == vgList) {
D
dapan1121 已提交
891
    ctgError("taosArrayInit failed, num:%d", vgNum);
D
dapan 已提交
892 893 894
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);    
  }

D
dapan1121 已提交
895
  void *pIter = taosHashIterate(vgHash, NULL);
896 897
  while (pIter) {
    vgInfo = pIter;
D
dapan1121 已提交
898

D
dapan1121 已提交
899
    if (NULL == taosArrayPush(vgList, vgInfo)) {
D
dapan1121 已提交
900
      ctgError("taosArrayPush failed, vgId:%d", vgInfo->vgId);
D
dapan1121 已提交
901
      taosHashCancelIterate(vgHash, pIter);      
D
dapan1121 已提交
902
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
903 904
    }
    
D
dapan1121 已提交
905
    pIter = taosHashIterate(vgHash, pIter);
906
    vgInfo = NULL;
D
dapan1121 已提交
907 908
  }

D
dapan1121 已提交
909
  *pList = vgList;
D
dapan1121 已提交
910

D
dapan1121 已提交
911
  ctgDebug("Got vgList from cache, vgNum:%d", vgNum);
D
dapan1121 已提交
912

D
dapan1121 已提交
913
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
914 915 916 917 918 919 920 921

_return:

  if (vgList) {
    taosArrayDestroy(vgList);
  }

  CTG_RET(code);
D
dapan1121 已提交
922 923
}

D
dapan1121 已提交
924
int32_t ctgGetVgInfoFromHashValue(SCatalog *pCtg, SDBVgInfo *dbInfo, const SName *pTableName, SVgroupInfo *pVgroup) {
D
dapan1121 已提交
925 926
  int32_t code = 0;
  
D
dapan1121 已提交
927
  int32_t vgNum = taosHashGetSize(dbInfo->vgHash);
H
Haojun Liao 已提交
928 929 930
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);

931
  if (vgNum <= 0) {
D
dapan1121 已提交
932
    ctgError("db vgroup cache invalid, db:%s, vgroup number:%d", db, vgNum);
D
dapan1121 已提交
933
    CTG_ERR_RET(TSDB_CODE_TSC_DB_NOT_SELECTED);
D
dapan1121 已提交
934 935
  }

936 937
  tableNameHashFp fp = NULL;
  SVgroupInfo *vgInfo = NULL;
D
dapan1121 已提交
938

D
dapan1121 已提交
939
  CTG_ERR_RET(ctgGetHashFunction(dbInfo->hashMethod, &fp));
940 941

  char tbFullName[TSDB_TABLE_FNAME_LEN];
H
Haojun Liao 已提交
942
  tNameExtractFullName(pTableName, tbFullName);
943 944 945

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

D
dapan1121 已提交
946
  void *pIter = taosHashIterate(dbInfo->vgHash, NULL);
947 948 949
  while (pIter) {
    vgInfo = pIter;
    if (hashValue >= vgInfo->hashBegin && hashValue <= vgInfo->hashEnd) {
D
dapan1121 已提交
950
      taosHashCancelIterate(dbInfo->vgHash, pIter);
951
      break;
D
dapan1121 已提交
952
    }
953
    
D
dapan1121 已提交
954
    pIter = taosHashIterate(dbInfo->vgHash, pIter);
955
    vgInfo = NULL;
D
dapan1121 已提交
956 957
  }

958
  if (NULL == vgInfo) {
D
dapan1121 已提交
959 960
    ctgError("no hash range found for hash value [%u], db:%s, numOfVgId:%d", hashValue, db, taosHashGetSize(dbInfo->vgHash));
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
961 962 963 964
  }

  *pVgroup = *vgInfo;

965
  CTG_RET(code);
D
dapan1121 已提交
966 967
}

D
dapan1121 已提交
968
int32_t ctgStbVersionCompare(const void* key1, const void* key2) {
D
dapan 已提交
969
  if (*(uint64_t *)key1 < ((SSTableMetaVersion*)key2)->suid) {
D
dapan1121 已提交
970
    return -1;
D
dapan 已提交
971
  } else if (*(uint64_t *)key1 > ((SSTableMetaVersion*)key2)->suid) {
D
dapan1121 已提交
972 973 974 975 976 977 978
    return 1;
  } else {
    return 0;
  }
}

int32_t ctgDbVgVersionCompare(const void* key1, const void* key2) {
D
dapan1121 已提交
979
  if (*(int64_t *)key1 < ((SDbVgVersion*)key2)->dbId) {
D
dapan1121 已提交
980
    return -1;
D
dapan1121 已提交
981
  } else if (*(int64_t *)key1 > ((SDbVgVersion*)key2)->dbId) {
D
dapan1121 已提交
982 983 984
    return 1;
  } else {
    return 0;
D
dapan1121 已提交
985
  }
D
dapan1121 已提交
986 987
}

D
dapan1121 已提交
988
int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type) {
D
dapan1121 已提交
989 990 991 992
  mgmt->slotRIdx = 0;
  mgmt->slotNum = rentSec / CTG_RENT_SLOT_SECOND;
  mgmt->type = type;

D
dapan1121 已提交
993
  size_t msgSize = sizeof(SCtgRentSlot) * mgmt->slotNum;
D
dapan1121 已提交
994
  
D
dapan1121 已提交
995 996 997
  mgmt->slots = calloc(1, msgSize);
  if (NULL == mgmt->slots) {
    qError("calloc %d failed", (int32_t)msgSize);
D
dapan 已提交
998
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
999
  }
D
dapan1121 已提交
1000

D
dapan1121 已提交
1001 1002 1003 1004
  qDebug("meta rent initialized, type:%d, slotNum:%d", type, mgmt->slotNum);
  
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
1005

D
dapan1121 已提交
1006

D
dapan1121 已提交
1007
int32_t ctgMetaRentAdd(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size) {
D
dapan1121 已提交
1008
  int16_t widx = abs(id % mgmt->slotNum);
D
dapan1121 已提交
1009

D
dapan1121 已提交
1010
  SCtgRentSlot *slot = &mgmt->slots[widx];
D
dapan1121 已提交
1011 1012 1013 1014 1015 1016 1017 1018
  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 已提交
1019
    }
D
dapan1121 已提交
1020
  }
D
dapan1121 已提交
1021

D
dapan1121 已提交
1022 1023 1024
  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 已提交
1025 1026
  }

D
dapan1121 已提交
1027
  slot->needSort = true;
D
dapan1121 已提交
1028

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

D
dapan1121 已提交
1031 1032 1033 1034 1035 1036
_return:

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

D
dapan1121 已提交
1037
int32_t ctgMetaRentUpdate(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size, __compar_fn_t compare) {
D
dapan1121 已提交
1038
  int16_t widx = abs(id % mgmt->slotNum);
D
dapan1121 已提交
1039

D
dapan1121 已提交
1040
  SCtgRentSlot *slot = &mgmt->slots[widx];
D
dapan1121 已提交
1041 1042 1043 1044
  int32_t code = 0;
  
  CTG_LOCK(CTG_WRITE, &slot->lock);
  if (NULL == slot->meta) {
D
dapan1121 已提交
1045 1046
    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 已提交
1047 1048 1049
  }

  if (slot->needSort) {
D
dapan1121 已提交
1050
    qDebug("meta slot before sorte, slot idx:%d, type:%d, size:%d", widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta));
D
dapan1121 已提交
1051 1052
    taosArraySort(slot->meta, compare);
    slot->needSort = false;
D
dapan1121 已提交
1053
    qDebug("meta slot sorted, slot idx:%d, type:%d, size:%d", widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta));
D
dapan1121 已提交
1054 1055 1056
  }

  void *orig = taosArraySearch(slot->meta, &id, compare, TD_EQ);
D
dapan1121 已提交
1057
  if (NULL == orig) {
D
dapan1121 已提交
1058
    qError("meta not found in slot, id:%"PRIx64", slot idx:%d, type:%d, size:%d", id, widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta));
D
dapan1121 已提交
1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077
    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 已提交
1078
int32_t ctgMetaRentRemove(SCtgRentMgmt *mgmt, int64_t id, __compar_fn_t compare) {
D
dapan1121 已提交
1079 1080
  int16_t widx = abs(id % mgmt->slotNum);

D
dapan1121 已提交
1081
  SCtgRentSlot *slot = &mgmt->slots[widx];
D
dapan1121 已提交
1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113
  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 已提交
1114
int32_t ctgMetaRentGetImpl(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size) {
D
dapan1121 已提交
1115 1116 1117 1118
  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 已提交
1119
  }
D
dapan1121 已提交
1120

D
dapan1121 已提交
1121
  SCtgRentSlot *slot = &mgmt->slots[ridx];
D
dapan1121 已提交
1122
  int32_t code = 0;
D
dapan1121 已提交
1123
  
D
dapan1121 已提交
1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159
  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);
}

D
dapan1121 已提交
1160
int32_t ctgMetaRentGet(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size) {
D
dapan1121 已提交
1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179
  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 已提交
1180 1181 1182
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
1183
int32_t ctgAddNewDBCache(SCatalog *pCtg, const char *dbFName, uint64_t dbId) {
D
dapan1121 已提交
1184
  int32_t code = 0;
D
dapan1121 已提交
1185

D
dapan1121 已提交
1186 1187 1188
  SCtgDBCache newDBCache = {0};
  newDBCache.dbId = dbId;

D
dapan 已提交
1189
  newDBCache.tbCache.metaCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
D
dapan1121 已提交
1190
  if (NULL == newDBCache.tbCache.metaCache) {
D
dapan 已提交
1191
    ctgError("taosHashInit %d metaCache failed", gCtgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
1192 1193 1194
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

D
dapan 已提交
1195
  newDBCache.tbCache.stbCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), true, HASH_ENTRY_LOCK);
D
dapan1121 已提交
1196
  if (NULL == newDBCache.tbCache.stbCache) {
D
dapan 已提交
1197
    ctgError("taosHashInit %d stbCache failed", gCtgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
1198 1199 1200 1201
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
  }

  code = taosHashPut(pCtg->dbCache, dbFName, strlen(dbFName), &newDBCache, sizeof(SCtgDBCache));
D
dapan1121 已提交
1202 1203 1204
  if (code) {
    if (HASH_NODE_EXIST(code)) {
      ctgDebug("db already in cache, dbFName:%s", dbFName);
D
dapan1121 已提交
1205
      goto _return;
D
dapan1121 已提交
1206 1207 1208
    }
    
    ctgError("taosHashPut db to cache failed, dbFName:%s", dbFName);
D
dapan1121 已提交
1209 1210 1211
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
  }
  
D
dapan1121 已提交
1212
  SDbVgVersion vgVersion = {.dbId = newDBCache.dbId, .vgVersion = -1};
D
dapan1121 已提交
1213 1214
  strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));

D
dapan1121 已提交
1215
  ctgDebug("db added to cache, dbFName:%s, dbId:%"PRIx64, dbFName, dbId);
D
dapan1121 已提交
1216

D
dapan1121 已提交
1217 1218 1219 1220
  if (CTG_IS_INF_DBNAME(dbFName)) {
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
1221 1222 1223
  CTG_ERR_RET(ctgMetaRentAdd(&pCtg->dbRent, &vgVersion, dbId, sizeof(SDbVgVersion)));

  ctgDebug("db added to rent, dbFName:%s, vgVersion:%d, dbId:%"PRIx64, dbFName, vgVersion.vgVersion, dbId);
D
dapan1121 已提交
1224

D
dapan1121 已提交
1225
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1226

D
dapan1121 已提交
1227
_return:
D
dapan1121 已提交
1228

D
dapan1121 已提交
1229
  ctgFreeDbCache(&newDBCache);
D
dapan1121 已提交
1230

D
dapan1121 已提交
1231 1232
  CTG_RET(code);
}
D
dapan1121 已提交
1233

D
dapan1121 已提交
1234

D
dapan1121 已提交
1235
void ctgRemoveStbRent(SCatalog* pCtg, SCtgTbMetaCache *cache) {
D
dapan1121 已提交
1236 1237 1238 1239 1240 1241 1242
  CTG_LOCK(CTG_WRITE, &cache->stbLock);
  if (cache->stbCache) {
    void *pIter = taosHashIterate(cache->stbCache, NULL);
    while (pIter) {
      uint64_t *suid = NULL;
      taosHashGetKey(pIter, (void **)&suid, NULL);

D
dapan1121 已提交
1243
      if (TSDB_CODE_SUCCESS == ctgMetaRentRemove(&pCtg->stbRent, *suid, ctgStbVersionCompare)) {
D
dapan1121 已提交
1244 1245 1246 1247 1248 1249 1250 1251 1252 1253
        ctgDebug("stb removed from rent, suid:%"PRIx64, *suid);
      }
          
      pIter = taosHashIterate(cache->stbCache, pIter);
    }
  }
  CTG_UNLOCK(CTG_WRITE, &cache->stbLock);
}


D
dapan1121 已提交
1254
int32_t ctgRemoveDB(SCatalog* pCtg, SCtgDBCache *dbCache, const char* dbFName) {
D
dapan 已提交
1255 1256 1257 1258
  uint64_t dbId = dbCache->dbId;
  
  ctgInfo("start to remove db from cache, dbFName:%s, dbId:%"PRIx64, dbFName, dbCache->dbId);

D
dapan1121 已提交
1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271
  atomic_store_8(&dbCache->deleted, 1);

  ctgRemoveStbRent(pCtg, &dbCache->tbCache);

  ctgFreeDbCache(dbCache);

  ctgInfo("db removed from cache, dbFName:%s, dbId:%"PRIx64, dbFName, dbCache->dbId);

  CTG_ERR_RET(ctgMetaRentRemove(&pCtg->dbRent, dbCache->dbId, ctgDbVgVersionCompare));
  
  ctgDebug("db removed from rent, dbFName:%s, dbId:%"PRIx64, dbFName, dbCache->dbId);

  if (taosHashRemove(pCtg->dbCache, dbFName, strlen(dbFName))) {
D
dapan1121 已提交
1272 1273
    ctgInfo("taosHashRemove from dbCache failed, may be removed, dbFName:%s", dbFName);
    CTG_ERR_RET(TSDB_CODE_CTG_DB_DROPPED);
D
dapan1121 已提交
1274
  }
D
dapan 已提交
1275 1276

  ctgInfo("db removed from cache, dbFName:%s, dbId:%"PRIx64, dbFName, dbId);
D
dapan1121 已提交
1277 1278 1279
  
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
1280 1281


D
dapan1121 已提交
1282
int32_t ctgGetAddDBCache(SCatalog* pCtg, const char *dbFName, uint64_t dbId, SCtgDBCache **pCache) {
D
dapan1121 已提交
1283 1284
  int32_t code = 0;
  SCtgDBCache *dbCache = NULL;
D
dapan1121 已提交
1285
  ctgGetDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
1286
  
D
dapan1121 已提交
1287 1288
  if (dbCache) {
  // TODO OPEN IT
D
dapan1121 已提交
1289
#if 0    
D
dapan1121 已提交
1290 1291 1292 1293
    if (dbCache->dbId == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
D
dapan1121 已提交
1294
#else
D
dapan1121 已提交
1295 1296 1297
    if (0 == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1298 1299
    }

D
dapan1121 已提交
1300 1301 1302 1303 1304
    if (dbId && (dbCache->dbId == 0)) {
      dbCache->dbId = dbId;
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
D
dapan1121 已提交
1305
    
D
dapan1121 已提交
1306 1307 1308 1309 1310 1311
    if (dbCache->dbId == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
#endif
    CTG_ERR_RET(ctgRemoveDB(pCtg, dbCache, dbFName));
D
dapan1121 已提交
1312
  }
D
dapan1121 已提交
1313 1314
  
  CTG_ERR_RET(ctgAddNewDBCache(pCtg, dbFName, dbId));
D
dapan1121 已提交
1315

D
dapan1121 已提交
1316
  ctgGetDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
1317

D
dapan1121 已提交
1318
  *pCache = dbCache;
D
dapan1121 已提交
1319

D
dapan1121 已提交
1320
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1321 1322 1323
}


D
dapan1121 已提交
1324 1325 1326 1327 1328 1329
int32_t ctgUpdateDBVgInfo(SCatalog* pCtg, const char* dbFName, uint64_t dbId, SDBVgInfo** pDbInfo) {
  int32_t code = 0;
  SDBVgInfo* dbInfo = *pDbInfo;
  
  if (NULL == dbInfo->vgHash || dbInfo->vgVersion < 0 || taosHashGetSize(dbInfo->vgHash) <= 0) {
    ctgError("invalid db vgInfo, dbFName:%s, vgHash:%p, vgVersion:%d", dbFName, dbInfo->vgHash, dbInfo->vgVersion);
D
dapan1121 已提交
1330
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1331 1332
  }

D
dapan1121 已提交
1333
  bool newAdded = false;
D
dapan1121 已提交
1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344
  SDbVgVersion vgVersion = {.dbId = dbId, .vgVersion = dbInfo->vgVersion};

  SCtgDBCache *dbCache = NULL;
  CTG_ERR_RET(ctgGetAddDBCache(pCtg, dbFName, dbId, &dbCache));
  if (NULL == dbCache) {
    ctgInfo("conflict db update, ignore this update, dbFName:%s, dbId:%"PRIx64, dbFName, dbId);
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }

  SDBVgInfo *vgInfo = NULL;
  CTG_ERR_RET(ctgWAcquireVgInfo(pCtg, dbCache));
D
dapan1121 已提交
1345
  
D
dapan1121 已提交
1346 1347 1348 1349
  if (dbCache->vgInfo) {
    if (dbInfo->vgVersion <= dbCache->vgInfo->vgVersion) {
      ctgInfo("db vgVersion is old, dbFName:%s, vgVersion:%d, currentVersion:%d", dbFName, dbInfo->vgVersion, dbCache->vgInfo->vgVersion);
      ctgWReleaseVgInfo(dbCache);
D
dapan1121 已提交
1350
      
D
dapan1121 已提交
1351
      return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1352
    }
D
dapan1121 已提交
1353 1354

    ctgFreeVgInfo(dbCache->vgInfo);
D
dapan1121 已提交
1355 1356
  }

D
dapan1121 已提交
1357
  dbCache->vgInfo = dbInfo;
D
dapan1121 已提交
1358

D
dapan1121 已提交
1359
  *pDbInfo = NULL;
D
dapan1121 已提交
1360

D
dapan1121 已提交
1361
  ctgDebug("db vgInfo updated, dbFName:%s, vgVersion:%d, dbId:%"PRIx64, dbFName, vgVersion.vgVersion, vgVersion.dbId);
D
dapan1121 已提交
1362

D
dapan1121 已提交
1363
  ctgWReleaseVgInfo(dbCache);
D
dapan1121 已提交
1364

D
dapan1121 已提交
1365
  dbCache = NULL;
D
dapan1121 已提交
1366

D
dapan1121 已提交
1367 1368 1369
  strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));
  CTG_ERR_RET(ctgMetaRentUpdate(&pCtg->dbRent, &vgVersion, vgVersion.dbId, sizeof(SDbVgVersion), ctgDbVgVersionCompare));
  
D
dapan1121 已提交
1370 1371 1372 1373
  CTG_RET(code);
}


D
dapan1121 已提交
1374 1375
int32_t ctgUpdateTblMeta(SCatalog *pCtg, SCtgDBCache *dbCache, char *dbFName, uint64_t dbId, char *tbName, STableMeta *meta, int32_t metaSize) {
  SCtgTbMetaCache *tbCache = &dbCache->tbCache;
D
dapan1121 已提交
1376

D
dapan1121 已提交
1377 1378 1379 1380 1381
  CTG_LOCK(CTG_READ, &tbCache->metaLock);
  if (dbCache->deleted || NULL == tbCache->metaCache || NULL == tbCache->stbCache) {
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);    
    ctgError("db is dropping, dbId:%"PRIx64, dbCache->dbId);
    CTG_ERR_RET(TSDB_CODE_CTG_DB_DROPPED);
D
dapan1121 已提交
1382 1383
  }

D
dapan1121 已提交
1384 1385 1386 1387 1388 1389 1390
  int8_t origType = 0;
  uint64_t origSuid = 0;
  bool isStb = meta->tableType == TSDB_SUPER_TABLE;
  STableMeta *orig = taosHashGet(tbCache->metaCache, tbName, strlen(tbName));
  if (orig) {
    origType = orig->tableType;
    
D
dapan1121 已提交
1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401
    if (origType == TSDB_SUPER_TABLE) {
      if ((!isStb) || orig->suid != meta->suid) {
        CTG_LOCK(CTG_WRITE, &tbCache->stbLock);
        if (taosHashRemove(tbCache->stbCache, &orig->suid, sizeof(orig->suid))) {
          ctgError("stb not exist in stbCache, dbFName:%s, stb:%s, suid:%"PRIx64, dbFName, tbName, orig->suid);
        }
        CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);

        ctgDebug("stb removed from stbCache, dbFName:%s, stb:%s, suid:%"PRIx64, dbFName, tbName, orig->suid);
        
        ctgMetaRentRemove(&pCtg->stbRent, orig->suid, ctgStbVersionCompare);
D
dapan1121 已提交
1402
      }
D
dapan1121 已提交
1403

D
dapan1121 已提交
1404
      origSuid = orig->suid;
D
dapan1121 已提交
1405
    }
D
dapan1121 已提交
1406
  }
D
dapan1121 已提交
1407

D
dapan1121 已提交
1408 1409
  if (isStb) {
    CTG_LOCK(CTG_WRITE, &tbCache->stbLock);
D
dapan1121 已提交
1410
  }
D
dapan1121 已提交
1411
  
D
dapan1121 已提交
1412 1413 1414 1415 1416 1417 1418 1419 1420
  if (taosHashPut(tbCache->metaCache, tbName, strlen(tbName), meta, metaSize) != 0) {
    if (isStb) {
      CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);
    }
    
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);  
    ctgError("taosHashPut tbmeta to cache failed, dbFName:%s, tbName:%s, tbType:%d", dbFName, tbName, meta->tableType);
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }
D
dapan1121 已提交
1421

D
dapan 已提交
1422
  ctgDebug("tbmeta updated to cache, dbFName:%s, tbName:%s, tbType:%d", dbFName, tbName, meta->tableType);
D
dapan1121 已提交
1423
  ctgDbgShowTableMeta(pCtg, tbName, meta);
D
dapan 已提交
1424

D
dapan1121 已提交
1425 1426 1427
  if (!isStb) {
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);  
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1428
  }
D
dapan1121 已提交
1429

D
dapan1121 已提交
1430
  if (origType == TSDB_SUPER_TABLE && origSuid == meta->suid) {
D
dapan1121 已提交
1431 1432 1433 1434
    CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);  
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
1435

D
dapan1121 已提交
1436 1437 1438 1439 1440 1441
  STableMeta *tbMeta = taosHashGet(tbCache->metaCache, tbName, strlen(tbName));
  if (taosHashPut(tbCache->stbCache, &meta->suid, sizeof(meta->suid), &tbMeta, POINTER_BYTES) != 0) {
    CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);    
    ctgError("taosHashPutExt stable to stable cache failed, suid:%"PRIx64, meta->suid);
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1442
  }
D
dapan1121 已提交
1443 1444
  
  CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);
D
dapan1121 已提交
1445

D
dapan1121 已提交
1446 1447
  CTG_UNLOCK(CTG_READ, &tbCache->metaLock);

D
dapan 已提交
1448
  ctgDebug("stb updated to stbCache, dbFName:%s, tbName:%s, tbType:%d", dbFName, tbName, meta->tableType);
D
dapan1121 已提交
1449 1450 1451 1452 1453 1454 1455

  SSTableMetaVersion metaRent = {.dbId = dbId, .suid = meta->suid, .sversion = meta->sversion, .tversion = meta->tversion};
  strcpy(metaRent.dbFName, dbFName);
  strcpy(metaRent.stbName, tbName);
  CTG_ERR_RET(ctgMetaRentAdd(&pCtg->stbRent, &metaRent, metaRent.suid, sizeof(SSTableMetaVersion)));
  
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1456 1457
}

D
dapan 已提交
1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496
int32_t ctgCloneVgInfo(SDBVgInfo *src, SDBVgInfo **dst) {
  *dst = malloc(sizeof(SDBVgInfo));
  if (NULL == *dst) {
    qError("malloc %d failed", (int32_t)sizeof(SDBVgInfo));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

  memcpy(*dst, src, sizeof(SDBVgInfo));

  size_t hashSize = taosHashGetSize(src->vgHash);
  (*dst)->vgHash = taosHashInit(hashSize, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
  if (NULL == (*dst)->vgHash) {
    qError("taosHashInit %d failed", (int32_t)hashSize);
    tfree(*dst);
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

  int32_t *vgId = NULL;
  void *pIter = taosHashIterate(src->vgHash, NULL);
  while (pIter) {
    taosHashGetKey(pIter, (void **)&vgId, NULL);

    if (taosHashPut((*dst)->vgHash, (void *)vgId, sizeof(int32_t), pIter, sizeof(SVgroupInfo))) {
      qError("taosHashPut failed, hashSize:%d", (int32_t)hashSize);
      taosHashCancelIterate(src->vgHash, pIter);
      taosHashCleanup((*dst)->vgHash);
      tfree(*dst);
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }
    
    pIter = taosHashIterate(src->vgHash, pIter);
  }


  return TSDB_CODE_SUCCESS;
}



D
dapan1121 已提交
1497
int32_t ctgGetDBVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* dbFName, bool forceUpdate, SCtgDBCache** dbCache, SDBVgInfo **pInfo) {
D
dapan1121 已提交
1498
  bool inCache = false;
D
dapan1121 已提交
1499
  int32_t code = 0;
D
dapan1121 已提交
1500 1501 1502 1503 1504

  CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, dbCache, &inCache));

  if (inCache && !forceUpdate) {
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1505 1506 1507 1508 1509
  }

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

D
dapan1121 已提交
1510
  tstrncpy(input.db, dbFName, tListLen(input.db));
D
dapan1121 已提交
1511 1512 1513 1514 1515 1516
  if (inCache) {
    input.dbId = (*dbCache)->dbId;
    input.vgVersion = (*dbCache)->vgInfo->vgVersion;
  } else {
    input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
  }
H
Haojun Liao 已提交
1517

D
dapan1121 已提交
1518 1519 1520 1521 1522 1523 1524 1525 1526
  code = ctgGetDBVgInfoFromMnode(pCtg, pRpc, pMgmtEps, &input, &DbOut);
  if (code) {
    if (CTG_DB_NOT_EXIST(code) && input.vgVersion > CTG_DEFAULT_INVALID_VERSION) {
      ctgDebug("db no longer exist, dbFName:%s, dbId:%" PRIx64, input.db, input.dbId);
      ctgPushRmDBMsgInQueue(pCtg, input.db, input.dbId);
    }

    CTG_ERR_RET(code);
  }
D
dapan1121 已提交
1527

D
dapan 已提交
1528 1529 1530 1531 1532 1533 1534 1535
  CTG_ERR_JRET(ctgCloneVgInfo(DbOut.dbVgroup, pInfo));
  
  SCtgMetaAction action= {.act = CTG_ACT_UPDATE_VG};
  SCtgUpdateVgMsg *msg = malloc(sizeof(SCtgUpdateVgMsg));
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateVgMsg));
    ctgFreeVgInfo(DbOut.dbVgroup);
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1536 1537
  }

D
dapan 已提交
1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  msg->pCtg = pCtg;
  msg->dbId = DbOut.dbId;
  msg->dbInfo = DbOut.dbVgroup;

  action.data = msg;

  CTG_ERR_JRET(ctgPushAction(&action));

  ctgDebug("action [%s] added into queue", gCtgAction[action.act].name);

D
dapan1121 已提交
1549
  return TSDB_CODE_SUCCESS;
D
dapan 已提交
1550 1551 1552 1553 1554 1555 1556 1557 1558

_return:

  tfree(*pInfo);
  tfree(msg);
  
  *pInfo = DbOut.dbVgroup;
  
  CTG_RET(code);
D
dapan1121 已提交
1559 1560
}

D
dapan 已提交
1561

D
dapan1121 已提交
1562 1563 1564 1565 1566
int32_t ctgCloneMetaOutput(STableMetaOutput *output, STableMetaOutput **pOutput) {
  *pOutput = malloc(sizeof(STableMetaOutput));
  if (NULL == *pOutput) {
    qError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan 已提交
1567 1568
  }

D
dapan1121 已提交
1569 1570 1571 1572 1573 1574
  memcpy(*pOutput, output, sizeof(STableMetaOutput));

  if (output->tbMeta) {
    int32_t metaSize = CTG_META_SIZE(output->tbMeta);
    (*pOutput)->tbMeta = malloc(metaSize);
    if (NULL == (*pOutput)->tbMeta) {
D
dapan 已提交
1575
      qError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
D
dapan1121 已提交
1576 1577 1578 1579 1580
      tfree(*pOutput);
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }

    memcpy((*pOutput)->tbMeta, output->tbMeta, metaSize);
D
dapan 已提交
1581 1582
  }

D
dapan1121 已提交
1583 1584 1585
  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
1586 1587


D
dapan1121 已提交
1588 1589
int32_t ctgRefreshTblMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, int32_t flag, STableMetaOutput **pOutput) {
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pTableName) {
D
dapan1121 已提交
1590 1591 1592 1593 1594 1595
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }

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

D
dapan1121 已提交
1596
  if (!CTG_FLAG_IS_INF_DB(flag)) {
D
dapan1121 已提交
1597 1598
    CTG_ERR_RET(catalogGetTableHashVgroup(pCtg, pTrans, pMgmtEps, pTableName, &vgroupInfo));
  }
D
dapan1121 已提交
1599

D
dapan 已提交
1600
  SCtgUpdateTblMsg *msg = NULL;
D
dapan1121 已提交
1601
  STableMetaOutput  moutput = {0};
S
Shengliang Guan 已提交
1602
  STableMetaOutput *output = calloc(1, sizeof(STableMetaOutput));
D
dapan1121 已提交
1603 1604 1605 1606
  if (NULL == output) {
    ctgError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }
D
dapan1121 已提交
1607

D
dapan1121 已提交
1608
  if (CTG_FLAG_IS_INF_DB(flag)) {
D
dapan1121 已提交
1609 1610 1611
    ctgDebug("will refresh tbmeta, supposed in information_schema, tbName:%s", tNameGetTableName(pTableName));

    CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, (char *)pTableName->dbname, (char *)pTableName->tname, output));
D
dapan1121 已提交
1612
  } else if (CTG_FLAG_IS_STB(flag)) {
D
dapan 已提交
1613
    ctgDebug("will refresh tbmeta, supposed to be stb, tbName:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
1614 1615

    // if get from mnode failed, will not try vnode
D
dapan1121 已提交
1616
    CTG_ERR_JRET(ctgGetTableMetaFromMnode(pCtg, pTrans, pMgmtEps, pTableName, output));
D
dapan1121 已提交
1617

D
dapan1121 已提交
1618
    if (CTG_IS_META_NULL(output->metaType)) {
D
dapan1121 已提交
1619
      CTG_ERR_JRET(ctgGetTableMetaFromVnode(pCtg, pTrans, pMgmtEps, pTableName, &vgroupInfo, output));
D
dapan1121 已提交
1620 1621
    }
  } else {
D
dapan1121 已提交
1622
    ctgDebug("will refresh tbmeta, not supposed to be stb, tbName:%s, flag:%d", tNameGetTableName(pTableName), flag);
D
dapan1121 已提交
1623 1624

    // if get from vnode failed or no table meta, will not try mnode
D
dapan1121 已提交
1625
    CTG_ERR_JRET(ctgGetTableMetaFromVnode(pCtg, pTrans, pMgmtEps, pTableName, &vgroupInfo, output));
D
dapan1121 已提交
1626

D
dapan1121 已提交
1627
    if (CTG_IS_META_TABLE(output->metaType) && TSDB_SUPER_TABLE == output->tbMeta->tableType) {
D
dapan1121 已提交
1628
      ctgDebug("will continue to refresh tbmeta since got stb, tbName:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
1629

D
dapan1121 已提交
1630
      tfree(output->tbMeta);
D
dapan1121 已提交
1631
      
D
dapan1121 已提交
1632
      CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, output->dbFName, output->tbName, output));
D
dapan1121 已提交
1633
    } else if (CTG_IS_META_BOTH(output->metaType)) {
D
dapan1121 已提交
1634
      int32_t exist = 0;
D
dapan1121 已提交
1635 1636 1637 1638
      if (!CTG_FLAG_IS_FORCE_UPDATE(flag)) {
        CTG_ERR_JRET(ctgIsTableMetaExistInCache(pCtg, output->dbFName, output->tbName, &exist));
      }
      
D
dapan1121 已提交
1639
      if (0 == exist) {
D
dapan1121 已提交
1640
        CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, output->dbFName, output->tbName, &moutput));
D
dapan1121 已提交
1641

D
dapan1121 已提交
1642
        if (CTG_IS_META_NULL(moutput.metaType)) {
D
dapan1121 已提交
1643
          SET_META_TYPE_NULL(output->metaType);
D
dapan1121 已提交
1644 1645
        }
        
D
dapan1121 已提交
1646 1647
        tfree(output->tbMeta);
        output->tbMeta = moutput.tbMeta;
D
dapan1121 已提交
1648 1649
        moutput.tbMeta = NULL;
      } else {
D
dapan1121 已提交
1650
        tfree(output->tbMeta);
D
dapan1121 已提交
1651
        
D
dapan1121 已提交
1652
        SET_META_TYPE_CTABLE(output->metaType); 
D
dapan1121 已提交
1653
      }
D
dapan1121 已提交
1654 1655 1656
    }
  }

D
dapan1121 已提交
1657
  if (CTG_IS_META_NULL(output->metaType)) {
D
dapan 已提交
1658
    ctgError("no tbmeta got, tbNmae:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
1659 1660 1661
    CTG_ERR_JRET(CTG_ERR_CODE_TABLE_NOT_EXIST);
  }

D
dapan 已提交
1662 1663 1664 1665 1666 1667
  if (CTG_IS_META_TABLE(output->metaType)) {
    ctgDebug("tbmeta got, dbFName:%s, tbName:%s, tbType:%d", output->dbFName, output->tbName, output->tbMeta->tableType);
  } else {
    ctgDebug("tbmeta got, dbFName:%s, tbName:%s, tbType:%d, stbMetaGot:%d", output->dbFName, output->ctbName, output->ctbMeta.tableType, CTG_IS_META_BOTH(output->metaType));
  }

D
dapan1121 已提交
1668 1669 1670 1671 1672
  if (pOutput) {
    CTG_ERR_JRET(ctgCloneMetaOutput(output, pOutput));
  }

  SCtgMetaAction action= {.act = CTG_ACT_UPDATE_TBL};
D
dapan 已提交
1673
  msg = malloc(sizeof(SCtgUpdateTblMsg));
D
dapan1121 已提交
1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateTblMsg));
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
  }

  msg->pCtg = pCtg;
  msg->output = output;

  action.data = msg;

  CTG_ERR_JRET(ctgPushAction(&action));

D
dapan 已提交
1686 1687
  ctgDebug("action [%s] added into queue", gCtgAction[action.act].name);

D
dapan1121 已提交
1688
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1689 1690 1691

_return:

D
dapan1121 已提交
1692
  tfree(output->tbMeta);
D
dapan 已提交
1693
  tfree(output);
D
dapan1121 已提交
1694
  tfree(msg);
D
dapan1121 已提交
1695 1696 1697 1698
  
  CTG_RET(code);
}

D
dapan1121 已提交
1699
int32_t ctgGetTableMeta(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t flag) {
D
dapan1121 已提交
1700
  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == pTableMeta) {
D
dapan1121 已提交
1701 1702 1703 1704
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }
  
  int32_t exist = 0;
D
dapan1121 已提交
1705
  int32_t code = 0;
D
dapan1121 已提交
1706 1707 1708
  uint64_t dbId = 0;
  uint64_t suid = 0;
  STableMetaOutput *output = NULL;
D
dapan1121 已提交
1709

D
dapan1121 已提交
1710
  if (CTG_IS_INF_DBNAME(pTableName->dbname)) {
D
dapan1121 已提交
1711
    CTG_FLAG_SET_INF_DB(flag);
D
dapan1121 已提交
1712 1713
  }

D
dapan1121 已提交
1714
  CTG_ERR_RET(ctgGetTableMetaFromCache(pCtg, pTableName, pTableMeta, &exist, flag, &dbId));
D
dapan1121 已提交
1715

D
dapan1121 已提交
1716 1717 1718 1719 1720
  int32_t tbType = 0;

  if (exist) {
    if (CTG_FLAG_MATCH_STB(flag, (*pTableMeta)->tableType) && ((!CTG_FLAG_IS_FORCE_UPDATE(flag)) || (CTG_FLAG_IS_INF_DB(flag)))) {
      goto _return;
D
dapan1121 已提交
1721
    }
D
dapan1121 已提交
1722

D
dapan1121 已提交
1723 1724
    tbType = (*pTableMeta)->tableType;
    suid = (*pTableMeta)->suid;
D
dapan1121 已提交
1725

D
dapan1121 已提交
1726 1727 1728 1729 1730
    tfree(*pTableMeta);    
  }
    
  if (CTG_FLAG_IS_UNKNOWN_STB(flag)) {
    CTG_FLAG_SET_STB(flag, tbType);
D
dapan1121 已提交
1731 1732
  }

D
dapan1121 已提交
1733

D
dapan 已提交
1734
  while (true) {
D
dapan1121 已提交
1735
    CTG_ERR_JRET(ctgRefreshTblMeta(pCtg, pRpc, pMgmtEps, pTableName, flag, &output));
D
dapan1121 已提交
1736

D
dapan 已提交
1737 1738 1739 1740
    if (CTG_IS_META_TABLE(output->metaType)) {
      *pTableMeta = output->tbMeta;
      goto _return;
    }
D
dapan1121 已提交
1741

D
dapan 已提交
1742 1743 1744 1745 1746 1747
    if (CTG_IS_META_BOTH(output->metaType)) {
      memcpy(output->tbMeta, &output->ctbMeta, sizeof(output->ctbMeta));
      
      *pTableMeta = output->tbMeta;
      goto _return;
    }
D
dapan1121 已提交
1748

D
dapan 已提交
1749 1750 1751 1752 1753
    if ((!CTG_IS_META_CTABLE(output->metaType)) || output->tbMeta) {
      ctgError("invalid metaType:%d", output->metaType);
      tfree(output->tbMeta);
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
    }
D
dapan1121 已提交
1754

D
dapan 已提交
1755
    // HANDLE ONLY CHILD TABLE META
D
dapan1121 已提交
1756

D
dapan 已提交
1757 1758 1759
    SName stbName = *pTableName;
    strcpy(stbName.tname, output->tbName);
    
D
dapan1121 已提交
1760
    CTG_ERR_JRET(ctgGetTableMetaFromCache(pCtg, &stbName, pTableMeta, &exist, flag, NULL));
D
dapan 已提交
1761 1762 1763 1764
    if (0 == exist) {
      ctgDebug("stb no longer exist, dbFName:%s, tbName:%s", output->dbFName, pTableName->tname);
      continue;
    }
D
dapan1121 已提交
1765

D
dapan 已提交
1766 1767 1768 1769
    memcpy(*pTableMeta, &output->ctbMeta, sizeof(output->ctbMeta));

    break;
  }
D
dapan1121 已提交
1770 1771 1772

_return:

D
dapan1121 已提交
1773 1774 1775 1776 1777 1778 1779 1780 1781 1782 1783 1784 1785 1786 1787
  if (CTG_TABLE_NOT_EXIST(code) && exist) {
    char dbFName[TSDB_DB_FNAME_LEN] = {0};
    if (CTG_FLAG_IS_INF_DB(flag)) {
      strcpy(dbFName, pTableName->dbname);
    } else {
      tNameGetFullDbName(pTableName, dbFName);
    }

    if (TSDB_SUPER_TABLE == tbType) {
      ctgPushRmStbMsgInQueue(pCtg, dbFName, dbId, pTableName->tname, suid);
    } else {
      ctgPushRmTblMsgInQueue(pCtg, dbFName, dbId, pTableName->tname);
    }
  }

D
dapan1121 已提交
1788 1789
  tfree(output);

D
dapan 已提交
1790 1791
  if (*pTableMeta) {
    ctgDebug("tbmeta returned, tbName:%s, tbType:%d", pTableName->tname, (*pTableMeta)->tableType);
D
dapan1121 已提交
1792
    ctgDbgShowTableMeta(pCtg, pTableName->tname, *pTableMeta);
D
dapan 已提交
1793 1794
  }

D
dapan1121 已提交
1795 1796 1797 1798 1799 1800 1801 1802 1803 1804 1805 1806 1807
  CTG_RET(code);
}



int32_t ctgActUpdateVg(SCtgMetaAction *action) {
  int32_t code = 0;
  SCtgUpdateVgMsg *msg = action->data;
  
  CTG_ERR_JRET(ctgUpdateDBVgInfo(msg->pCtg, msg->dbFName, msg->dbId, &msg->dbInfo));

_return:

D
dapan1121 已提交
1808
  ctgFreeVgInfo(msg->dbInfo);
D
dapan1121 已提交
1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819
  tfree(msg);
  
  CTG_RET(code);
}

int32_t ctgActRemoveDB(SCtgMetaAction *action) {
  int32_t code = 0;
  SCtgRemoveDBMsg *msg = action->data;
  SCatalog* pCtg = msg->pCtg;

  SCtgDBCache *dbCache = NULL;
D
dapan 已提交
1820
  ctgGetDBCache(msg->pCtg, msg->dbFName, &dbCache);
D
dapan1121 已提交
1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848
  if (NULL == dbCache) {
    goto _return;
  }
  
  if (dbCache->dbId != msg->dbId) {
    ctgInfo("dbId already updated, dbFName:%s, dbId:%"PRIx64 ", targetId:%"PRIx64, msg->dbFName, dbCache->dbId, msg->dbId);
    goto _return;
  }
  
  CTG_ERR_JRET(ctgRemoveDB(pCtg, dbCache, msg->dbFName));

_return:

  tfree(msg);
  
  CTG_RET(code);
}


int32_t ctgActUpdateTbl(SCtgMetaAction *action) {
  int32_t code = 0;
  SCtgUpdateTblMsg *msg = action->data;
  SCatalog* pCtg = msg->pCtg;
  STableMetaOutput* output = msg->output;
  SCtgDBCache *dbCache = NULL;

  if ((!CTG_IS_META_CTABLE(output->metaType)) && NULL == output->tbMeta) {
    ctgError("no valid tbmeta got from meta rsp, dbFName:%s, tbName:%s", output->dbFName, output->tbName);
D
dapan 已提交
1849
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
1850 1851 1852 1853 1854 1855 1856
  }

  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_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }    

D
dapan1121 已提交
1857 1858 1859 1860 1861
  char *p = strchr(output->dbFName, '.');
  if (p && CTG_IS_INF_DBNAME(p + 1)) {
    memmove(output->dbFName, p + 1, strlen(p + 1));
  }
  
D
dapan1121 已提交
1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879
  CTG_ERR_JRET(ctgGetAddDBCache(pCtg, output->dbFName, output->dbId, &dbCache));
  if (NULL == dbCache) {
    ctgInfo("conflict db update, ignore this update, dbFName:%s, dbId:%"PRIx64, output->dbFName, output->dbId);
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }

  if (CTG_IS_META_TABLE(output->metaType) || CTG_IS_META_BOTH(output->metaType)) {
    int32_t metaSize = CTG_META_SIZE(output->tbMeta);
    
    CTG_ERR_JRET(ctgUpdateTblMeta(pCtg, dbCache, output->dbFName, output->dbId, output->tbName, output->tbMeta, metaSize));
  }

  if (CTG_IS_META_CTABLE(output->metaType) || CTG_IS_META_BOTH(output->metaType)) {
    CTG_ERR_JRET(ctgUpdateTblMeta(pCtg, dbCache, output->dbFName, output->dbId, output->ctbName, (STableMeta *)&output->ctbMeta, sizeof(output->ctbMeta)));
  }

_return:

D
dapan 已提交
1880 1881 1882
  if (output) {
    tfree(output->tbMeta);
    tfree(output);
D
dapan1121 已提交
1883
  }
D
dapan 已提交
1884
  
D
dapan1121 已提交
1885 1886 1887 1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908 1909 1910 1911 1912 1913 1914 1915 1916 1917 1918
  tfree(msg);
  
  CTG_RET(code);
}


int32_t ctgActRemoveStb(SCtgMetaAction *action) {
  int32_t code = 0;
  SCtgRemoveStbMsg *msg = action->data;
  SCatalog* pCtg = msg->pCtg;

  SCtgDBCache *dbCache = NULL;
  ctgGetDBCache(pCtg, msg->dbFName, &dbCache);
  if (NULL == dbCache) {
    return TSDB_CODE_SUCCESS;
  }

  if (dbCache->dbId != msg->dbId) {
    ctgDebug("dbId already modified, dbFName:%s, current:%"PRIx64", dbId:%"PRIx64", stb:%s, suid:%"PRIx64, msg->dbFName, dbCache->dbId, msg->dbId, msg->stbName, msg->suid);
    return TSDB_CODE_SUCCESS;
  }
  
  CTG_LOCK(CTG_WRITE, &dbCache->tbCache.stbLock);
  if (taosHashRemove(dbCache->tbCache.stbCache, &msg->suid, sizeof(msg->suid))) {
    CTG_UNLOCK(CTG_WRITE, &dbCache->tbCache.stbLock);
    ctgDebug("stb not exist in stbCache, may be removed, dbFName:%s, stb:%s, suid:%"PRIx64, msg->dbFName, msg->stbName, msg->suid);
    return TSDB_CODE_SUCCESS;
  }

  CTG_LOCK(CTG_READ, &dbCache->tbCache.metaLock);
  if (taosHashRemove(dbCache->tbCache.metaCache, msg->stbName, strlen(msg->stbName))) {  
    CTG_UNLOCK(CTG_READ, &dbCache->tbCache.metaLock);
    CTG_UNLOCK(CTG_WRITE, &dbCache->tbCache.stbLock);
    ctgError("stb not exist in cache, dbFName:%s, stb:%s, suid:%"PRIx64, msg->dbFName, msg->stbName, msg->suid);
D
dapan1121 已提交
1919
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
1920 1921 1922 1923 1924 1925 1926
  }  
  CTG_UNLOCK(CTG_READ, &dbCache->tbCache.metaLock);
  
  CTG_UNLOCK(CTG_WRITE, &dbCache->tbCache.stbLock);
  
  ctgInfo("stb removed from cache, dbFName:%s, stbName:%s, suid:%"PRIx64, msg->dbFName, msg->stbName, msg->suid);

D
dapan1121 已提交
1927
  CTG_ERR_JRET(ctgMetaRentRemove(&msg->pCtg->stbRent, msg->suid, ctgStbVersionCompare));
D
dapan1121 已提交
1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938
  
  ctgDebug("stb removed from rent, dbFName:%s, stbName:%s, suid:%"PRIx64, msg->dbFName, msg->stbName, msg->suid);
  
_return:

  tfree(msg);
  
  CTG_RET(code);
}

int32_t ctgActRemoveTbl(SCtgMetaAction *action) {
D
dapan1121 已提交
1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964
  int32_t code = 0;
  SCtgRemoveTblMsg *msg = action->data;
  SCatalog* pCtg = msg->pCtg;

  SCtgDBCache *dbCache = NULL;
  ctgGetDBCache(pCtg, msg->dbFName, &dbCache);
  if (NULL == dbCache) {
    return TSDB_CODE_SUCCESS;
  }

  if (dbCache->dbId != msg->dbId) {
    ctgDebug("dbId already modified, dbFName:%s, current:%"PRIx64", dbId:%"PRIx64", tbName:%s", msg->dbFName, dbCache->dbId, msg->dbId, msg->tbName);
    return TSDB_CODE_SUCCESS;
  }
  
  CTG_LOCK(CTG_READ, &dbCache->tbCache.metaLock);
  if (taosHashRemove(dbCache->tbCache.metaCache, msg->tbName, strlen(msg->tbName))) {  
    CTG_UNLOCK(CTG_READ, &dbCache->tbCache.metaLock);
    ctgError("stb not exist in cache, dbFName:%s, tbName:%s", msg->dbFName, msg->tbName);
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }  
  CTG_UNLOCK(CTG_READ, &dbCache->tbCache.metaLock);
  
  ctgInfo("table removed from cache, dbFName:%s, tbName:%s", msg->dbFName, msg->tbName);
  
_return:
D
dapan1121 已提交
1965

D
dapan1121 已提交
1966 1967 1968
  tfree(msg);
  
  CTG_RET(code);
D
dapan1121 已提交
1969 1970 1971 1972 1973 1974 1975 1976
}


void* ctgUpdateThreadFunc(void* param) {
  setThreadName("catalog");

  qInfo("catalog update thread started");

D
dapan 已提交
1977
  CTG_LOCK(CTG_READ, &gCtgMgmt.lock);
D
dapan1121 已提交
1978 1979
  
  while (true) {
D
dapan 已提交
1980
    tsem_wait(&gCtgMgmt.sem);
D
dapan1121 已提交
1981
    
D
dapan 已提交
1982
    if (atomic_load_8(&gCtgMgmt.exit)) {
D
dapan1121 已提交
1983 1984 1985 1986 1987
      break;
    }

    SCtgMetaAction *action = NULL;
    ctgPopAction(&action);
D
dapan1121 已提交
1988
    SCatalog *pCtg = ((SCtgUpdateMsgHeader *)action->data)->pCtg;
D
dapan1121 已提交
1989

D
dapan1121 已提交
1990
    ctgDebug("process [%s] action", gCtgAction[action->act].name);
D
dapan 已提交
1991 1992 1993 1994
    
    (*gCtgAction[action->act].func)(action);

    CTG_STAT_ADD(gCtgMgmt.stat.runtime.qDoneNum); 
D
dapan1121 已提交
1995 1996

    ctgDbgShowClusterCache(pCtg);
D
dapan1121 已提交
1997 1998
  }

D
dapan 已提交
1999
  CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock);
D
dapan1121 已提交
2000 2001 2002 2003 2004 2005 2006 2007 2008 2009 2010 2011

  qInfo("catalog update thread stopped");
  
  return NULL;
}


int32_t ctgStartUpdateThread() {
  pthread_attr_t thAttr;
  pthread_attr_init(&thAttr);
  pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);

D
dapan 已提交
2012
  if (pthread_create(&gCtgMgmt.updateThread, &thAttr, ctgUpdateThreadFunc, NULL) != 0) {
D
dapan1121 已提交
2013 2014
    terrno = TAOS_SYSTEM_ERROR(errno);
    CTG_ERR_RET(terrno);
D
dapan1121 已提交
2015 2016
  }
  
D
dapan1121 已提交
2017
  pthread_attr_destroy(&thAttr);
D
dapan1121 已提交
2018 2019 2020 2021
  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
2022
int32_t catalogInit(SCatalogCfg *cfg) {
D
dapan 已提交
2023
  if (gCtgMgmt.pCluster) {
D
dapan 已提交
2024
    qError("catalog already initialized");
D
dapan1121 已提交
2025
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2026 2027
  }

D
dapan 已提交
2028
  atomic_store_8(&gCtgMgmt.exit, false);
D
dapan1121 已提交
2029

D
dapan1121 已提交
2030
  if (cfg) {
D
dapan 已提交
2031
    memcpy(&gCtgMgmt.cfg, cfg, sizeof(*cfg));
H
Haojun Liao 已提交
2032

D
dapan 已提交
2033 2034
    if (gCtgMgmt.cfg.maxDBCacheNum == 0) {
      gCtgMgmt.cfg.maxDBCacheNum = CTG_DEFAULT_CACHE_DB_NUMBER;
D
dapan1121 已提交
2035 2036
    }

D
dapan 已提交
2037 2038
    if (gCtgMgmt.cfg.maxTblCacheNum == 0) {
      gCtgMgmt.cfg.maxTblCacheNum = CTG_DEFAULT_CACHE_TBLMETA_NUMBER;
D
dapan1121 已提交
2039
    }
D
dapan1121 已提交
2040

D
dapan 已提交
2041 2042
    if (gCtgMgmt.cfg.dbRentSec == 0) {
      gCtgMgmt.cfg.dbRentSec = CTG_DEFAULT_RENT_SECOND;
D
dapan1121 已提交
2043 2044
    }

D
dapan 已提交
2045 2046
    if (gCtgMgmt.cfg.stbRentSec == 0) {
      gCtgMgmt.cfg.stbRentSec = CTG_DEFAULT_RENT_SECOND;
D
dapan1121 已提交
2047
    }
D
dapan1121 已提交
2048
  } else {
D
dapan 已提交
2049 2050 2051 2052
    gCtgMgmt.cfg.maxDBCacheNum = CTG_DEFAULT_CACHE_DB_NUMBER;
    gCtgMgmt.cfg.maxTblCacheNum = CTG_DEFAULT_CACHE_TBLMETA_NUMBER;
    gCtgMgmt.cfg.dbRentSec = CTG_DEFAULT_RENT_SECOND;
    gCtgMgmt.cfg.stbRentSec = CTG_DEFAULT_RENT_SECOND;
D
dapan 已提交
2053 2054
  }

D
dapan 已提交
2055 2056
  gCtgMgmt.pCluster = taosHashInit(CTG_DEFAULT_CACHE_CLUSTER_NUMBER, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
  if (NULL == gCtgMgmt.pCluster) {
D
dapan1121 已提交
2057 2058
    qError("taosHashInit %d cluster cache failed", CTG_DEFAULT_CACHE_CLUSTER_NUMBER);
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
2059 2060
  }

D
dapan1121 已提交
2061 2062
  CTG_ERR_RET(ctgStartUpdateThread());

D
dapan 已提交
2063
  tsem_init(&gCtgMgmt.sem, 0, 0);
D
dapan1121 已提交
2064

D
dapan 已提交
2065 2066
  gCtgMgmt.head = calloc(1, sizeof(SCtgQNode));
  if (NULL == gCtgMgmt.head) {
D
dapan1121 已提交
2067 2068 2069
    qError("calloc %d failed", (int32_t)sizeof(SCtgQNode));
    CTG_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }
D
dapan 已提交
2070
  gCtgMgmt.tail = gCtgMgmt.head;
D
dapan1121 已提交
2071

D
dapan 已提交
2072
  qDebug("catalog initialized, maxDb:%u, maxTbl:%u, dbRentSec:%u, stbRentSec:%u", gCtgMgmt.cfg.maxDBCacheNum, gCtgMgmt.cfg.maxTblCacheNum, gCtgMgmt.cfg.dbRentSec, gCtgMgmt.cfg.stbRentSec);
D
dapan1121 已提交
2073

D
dapan 已提交
2074
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
2075 2076
}

D
dapan1121 已提交
2077
int32_t catalogGetHandle(uint64_t clusterId, SCatalog** catalogHandle) {
2078
  if (NULL == catalogHandle) {
D
dapan1121 已提交
2079
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
2080 2081
  }

D
dapan 已提交
2082
  if (NULL == gCtgMgmt.pCluster) {
D
dapan 已提交
2083
    qError("catalog cluster cache are not ready, clusterId:%"PRIx64, clusterId);
D
dapan1121 已提交
2084
    CTG_ERR_RET(TSDB_CODE_CTG_NOT_READY);
D
dapan 已提交
2085 2086
  }

D
dapan1121 已提交
2087 2088
  int32_t code = 0;
  SCatalog *clusterCtg = NULL;
D
dapan 已提交
2089

D
dapan1121 已提交
2090
  while (true) {
D
dapan 已提交
2091
    SCatalog **ctg = (SCatalog **)taosHashGet(gCtgMgmt.pCluster, (char*)&clusterId, sizeof(clusterId));
D
dapan 已提交
2092

D
dapan1121 已提交
2093 2094 2095 2096 2097
    if (ctg && (*ctg)) {
      *catalogHandle = *ctg;
      qDebug("got catalog handle from cache, clusterId:%"PRIx64", CTG:%p", clusterId, *ctg);
      return TSDB_CODE_SUCCESS;
    }
D
dapan 已提交
2098

D
dapan1121 已提交
2099 2100 2101 2102 2103 2104
    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 已提交
2105 2106
    clusterCtg->clusterId = clusterId;

D
dapan 已提交
2107 2108
    CTG_ERR_JRET(ctgMetaRentInit(&clusterCtg->dbRent, gCtgMgmt.cfg.dbRentSec, CTG_RENT_DB));
    CTG_ERR_JRET(ctgMetaRentInit(&clusterCtg->stbRent, gCtgMgmt.cfg.stbRentSec, CTG_RENT_STABLE));
D
dapan1121 已提交
2109

D
dapan 已提交
2110
    clusterCtg->dbCache = taosHashInit(gCtgMgmt.cfg.maxDBCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
D
dapan1121 已提交
2111 2112 2113 2114 2115
    if (NULL == clusterCtg->dbCache) {
      qError("taosHashInit %d dbCache failed", CTG_DEFAULT_CACHE_DB_NUMBER);
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
    }

D
dapan 已提交
2116
    SHashObj *metaCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
D
dapan1121 已提交
2117
    if (NULL == metaCache) {
D
dapan 已提交
2118
      qError("taosHashInit failed, num:%d", gCtgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
2119 2120 2121
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }
    
D
dapan 已提交
2122
    code = taosHashPut(gCtgMgmt.pCluster, &clusterId, sizeof(clusterId), &clusterCtg, POINTER_BYTES);
D
dapan1121 已提交
2123 2124 2125 2126 2127 2128 2129 2130 2131 2132 2133 2134 2135
    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 已提交
2136
  }
D
dapan1121 已提交
2137 2138

  *catalogHandle = clusterCtg;
D
dapan 已提交
2139
  
D
dapan1121 已提交
2140
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
2141 2142 2143 2144 2145 2146 2147 2148

_return:

  ctgFreeHandle(clusterCtg);
  
  CTG_RET(code);
}

D
dapan1121 已提交
2149 2150
void catalogFreeHandle(SCatalog* pCtg) {
  if (NULL == pCtg) {
D
dapan1121 已提交
2151 2152
    return;
  }
D
dapan1121 已提交
2153

D
dapan 已提交
2154
  if (taosHashRemove(gCtgMgmt.pCluster, &pCtg->clusterId, sizeof(pCtg->clusterId))) {
D
dapan1121 已提交
2155
    ctgWarn("taosHashRemove from cluster failed, may already be freed, clusterId:%"PRIx64, pCtg->clusterId);
D
dapan1121 已提交
2156 2157 2158
    return;
  }

D
dapan1121 已提交
2159
  uint64_t clusterId = pCtg->clusterId;
D
dapan1121 已提交
2160
  
D
dapan1121 已提交
2161
  ctgFreeHandle(pCtg);
D
dapan1121 已提交
2162 2163

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

D
dapan1121 已提交
2166
int32_t catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* version, int64_t* dbId) {
D
dapan1121 已提交
2167 2168
  CTG_API_ENTER();

D
dapan1121 已提交
2169
  if (NULL == pCtg || NULL == dbFName || NULL == version || NULL == dbId) {
D
dapan1121 已提交
2170 2171 2172 2173
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2174
    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
2175
    ctgInfo("empty db cache, dbFName:%s", dbFName);
D
dapan1121 已提交
2176
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2177 2178
  }

D
dapan1121 已提交
2179 2180 2181
  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
  if (NULL == dbCache) {
D
dapan1121 已提交
2182
    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
2183
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2184 2185
  }

D
dapan 已提交
2186 2187 2188
  bool inCache = false;
  ctgAcquireVgInfo(pCtg, dbCache, &inCache);
  if (!inCache) {
D
dapan1121 已提交
2189
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
2190 2191

    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
2192
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2193 2194
  }

D
dapan1121 已提交
2195
  *version = dbCache->vgInfo->vgVersion;
D
dapan1121 已提交
2196
  *dbId = dbCache->dbId;
D
dapan1121 已提交
2197 2198 2199

  ctgReleaseVgInfo(dbCache);
  ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
2200

D
dapan1121 已提交
2201
  ctgDebug("Got db vgVersion from cache, dbFName:%s, vgVersion:%d", dbFName, *version);
D
dapan1121 已提交
2202

D
dapan1121 已提交
2203
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2204 2205
}

D
dapan1121 已提交
2206
int32_t catalogGetDBVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* dbFName, bool forceUpdate, SArray** vgroupList) {
D
dapan1121 已提交
2207 2208
  CTG_API_ENTER();

D
dapan1121 已提交
2209 2210 2211
  if (NULL == pCtg || NULL == dbFName || NULL == pRpc || NULL == pMgmtEps || NULL == vgroupList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
2212

D
dapan1121 已提交
2213
  SCtgDBCache* dbCache = NULL;
2214
  int32_t code = 0;
D
dapan1121 已提交
2215
  SArray *vgList = NULL;
D
dapan1121 已提交
2216 2217 2218 2219 2220 2221 2222
  SHashObj *vgHash = NULL;
  SDBVgInfo *vgInfo = NULL;
  CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pRpc, pMgmtEps, dbFName, forceUpdate, &dbCache, &vgInfo));
  if (dbCache) {
    vgHash = dbCache->vgInfo->vgHash;
  } else {
    vgHash = vgInfo->vgHash;
D
dapan1121 已提交
2223 2224
  }

D
dapan1121 已提交
2225
  CTG_ERR_JRET(ctgGenerateVgList(pCtg, vgHash, &vgList));
D
dapan1121 已提交
2226 2227 2228 2229 2230

  *vgroupList = vgList;
  vgList = NULL;

_return:
D
dapan1121 已提交
2231 2232

  if (dbCache) {
D
dapan1121 已提交
2233 2234
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
2235 2236
  }

D
dapan1121 已提交
2237 2238 2239
  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
    tfree(vgInfo);
D
dapan1121 已提交
2240 2241
  }

D
dapan1121 已提交
2242
  CTG_API_LEAVE(code);  
D
dapan1121 已提交
2243 2244 2245
}


D
dapan1121 已提交
2246
int32_t catalogUpdateDBVgInfo(SCatalog* pCtg, const char* dbFName, uint64_t dbId, SDBVgInfo* dbInfo) {
D
dapan1121 已提交
2247
  CTG_API_ENTER();
D
dapan1121 已提交
2248 2249

  int32_t code = 0;
D
dapan1121 已提交
2250
  
D
dapan1121 已提交
2251
  if (NULL == pCtg || NULL == dbFName || NULL == dbInfo) {
D
dapan1121 已提交
2252 2253 2254
    CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2255 2256 2257 2258
  SCtgMetaAction action= {.act = CTG_ACT_UPDATE_VG};
  SCtgUpdateVgMsg *msg = malloc(sizeof(SCtgUpdateVgMsg));
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateVgMsg));
D
dapan1121 已提交
2259
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
2260 2261
  }

D
dapan1121 已提交
2262 2263 2264 2265
  msg->pCtg = pCtg;
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  msg->dbId = dbId;
  msg->dbInfo = dbInfo;
D
dapan1121 已提交
2266

D
dapan1121 已提交
2267
  action.data = msg;
D
dapan1121 已提交
2268

D
dapan1121 已提交
2269
  CTG_ERR_JRET(ctgPushAction(&action));
D
dapan1121 已提交
2270

D
dapan1121 已提交
2271 2272
  dbInfo = NULL;

D
dapan 已提交
2273 2274
  ctgDebug("action [%s] added into queue", gCtgAction[action.act].name);

D
dapan1121 已提交
2275 2276
  CTG_API_LEAVE(code);
  
D
dapan1121 已提交
2277 2278
_return:

D
dapan1121 已提交
2279
  ctgFreeVgInfo(dbInfo);
D
dapan1121 已提交
2280 2281

  tfree(msg);
D
dapan1121 已提交
2282
  
D
dapan1121 已提交
2283
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2284 2285 2286
}


D
dapan1121 已提交
2287 2288 2289
int32_t catalogRemoveDB(SCatalog* pCtg, const char* dbFName, uint64_t dbId) {
  CTG_API_ENTER();

D
dapan1121 已提交
2290 2291
  int32_t code = 0;
  
D
dapan1121 已提交
2292 2293
  if (NULL == pCtg || NULL == dbFName) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2294 2295
  }

D
dapan1121 已提交
2296
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2297
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2298
  }
D
dapan1121 已提交
2299

D
dapan1121 已提交
2300
  CTG_ERR_JRET(ctgPushRmDBMsgInQueue(pCtg, dbFName, dbId));
D
dapan 已提交
2301

D
dapan1121 已提交
2302
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2303
  
D
dapan1121 已提交
2304 2305
_return:

D
dapan1121 已提交
2306
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2307 2308
}

D
dapan1121 已提交
2309 2310 2311
int32_t catalogRemoveStbMeta(SCatalog* pCtg, const char* dbFName, uint64_t dbId, const char* stbName, uint64_t suid) {
  CTG_API_ENTER();

D
dapan 已提交
2312 2313
  int32_t code = 0;
  
D
dapan1121 已提交
2314 2315
  if (NULL == pCtg || NULL == dbFName || NULL == stbName) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
2316 2317
  }

D
dapan1121 已提交
2318
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2319
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2320
  }
D
dapan1121 已提交
2321

D
dapan1121 已提交
2322
  CTG_ERR_JRET(ctgPushRmStbMsgInQueue(pCtg, dbFName, dbId, stbName, suid));
D
dapan 已提交
2323

D
dapan1121 已提交
2324
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2325
  
D
dapan1121 已提交
2326 2327
_return:

D
dapan1121 已提交
2328
  CTG_API_LEAVE(code);
D
dapan 已提交
2329 2330
}

D
dapan1121 已提交
2331

D
dapan1121 已提交
2332
int32_t catalogGetTableMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta) {
D
dapan1121 已提交
2333 2334
  CTG_API_ENTER();

D
dapan1121 已提交
2335
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_UNKNOWN_STB));
D
dapan1121 已提交
2336
}
D
dapan1121 已提交
2337

D
dapan1121 已提交
2338
int32_t catalogGetSTableMeta(SCatalog* pCtg, void * pTrans, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta) {
D
dapan1121 已提交
2339 2340
  CTG_API_ENTER();

D
dapan1121 已提交
2341
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_STB));
D
dapan1121 已提交
2342 2343
}

D
dapan1121 已提交
2344 2345 2346 2347 2348 2349 2350 2351 2352 2353 2354 2355 2356
int32_t catalogUpdateSTableMeta(SCatalog* pCtg, STableMetaRsp *rspMsg) {
  CTG_API_ENTER();

  if (NULL == pCtg || NULL == rspMsg) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

  STableMetaOutput *output = calloc(1, sizeof(STableMetaOutput));
  if (NULL == output) {
    ctgError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
    CTG_API_LEAVE(TSDB_CODE_CTG_MEM_ERROR);
  }
  
D
dapan1121 已提交
2357 2358
  int32_t code = 0;

D
dapan1121 已提交
2359 2360
  strcpy(output->dbFName, rspMsg->dbFName);
  strcpy(output->tbName, rspMsg->tbName);
D
dapan1121 已提交
2361

D
dapan1121 已提交
2362
  output->dbId = rspMsg->dbId;
D
dapan1121 已提交
2363
  
D
dapan1121 已提交
2364
  SET_META_TYPE_TABLE(output->metaType);
D
dapan1121 已提交
2365
  
D
dapan1121 已提交
2366
  CTG_ERR_JRET(queryCreateTableMetaFromMsg(rspMsg, true, &output->tbMeta));
D
dapan1121 已提交
2367

D
dapan1121 已提交
2368 2369 2370 2371 2372 2373
  SCtgMetaAction action= {.act = CTG_ACT_UPDATE_TBL};
  SCtgUpdateTblMsg *msg = malloc(sizeof(SCtgUpdateTblMsg));
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateTblMsg));
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
  }
D
dapan1121 已提交
2374

D
dapan1121 已提交
2375 2376 2377 2378 2379 2380 2381
  msg->pCtg = pCtg;
  msg->output = output;

  action.data = msg;

  CTG_ERR_JRET(ctgPushAction(&action));

D
dapan 已提交
2382 2383
  ctgDebug("action [%s] added into queue", gCtgAction[action.act].name);

D
dapan1121 已提交
2384 2385
  CTG_API_LEAVE(code);
  
D
dapan1121 已提交
2386 2387
_return:

D
dapan1121 已提交
2388 2389 2390
  tfree(output->tbMeta);
  tfree(output);
  tfree(msg);
D
dapan1121 已提交
2391
  
D
dapan1121 已提交
2392
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2393 2394 2395
}


D
dapan1121 已提交
2396
int32_t catalogRefreshTableMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, int32_t isSTable) {
D
dapan1121 已提交
2397 2398
  CTG_API_ENTER();

D
dapan1121 已提交
2399
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pTableName) {
D
dapan1121 已提交
2400 2401 2402
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2403
  CTG_API_LEAVE(ctgRefreshTblMeta(pCtg, pTrans, pMgmtEps, pTableName, CTG_FLAG_FORCE_UPDATE | CTG_FLAG_MAKE_STB(isSTable), NULL));
2404
}
2405

D
dapan1121 已提交
2406
int32_t catalogRefreshGetTableMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t isSTable) {
D
dapan1121 已提交
2407 2408
  CTG_API_ENTER();

D
dapan1121 已提交
2409
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_FORCE_UPDATE | CTG_FLAG_MAKE_STB(isSTable)));
D
dapan1121 已提交
2410 2411
}

D
dapan1121 已提交
2412
int32_t catalogGetTableDistVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgList) {
D
dapan1121 已提交
2413
  CTG_API_ENTER();
D
dapan1121 已提交
2414 2415 2416 2417

  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == pVgList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
2418 2419 2420 2421 2422

  if (CTG_IS_INF_DBNAME(pTableName->dbname)) {
    ctgError("no valid vgInfo for db, dbname:%s", pTableName->dbname);
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
2423 2424 2425 2426
  
  STableMeta *tbMeta = NULL;
  int32_t code = 0;
  SVgroupInfo vgroupInfo = {0};
D
dapan1121 已提交
2427
  SCtgDBCache* dbCache = NULL;
D
dapan1121 已提交
2428
  SArray *vgList = NULL;
D
dapan 已提交
2429
  SDBVgInfo *vgInfo = NULL;
D
dapan1121 已提交
2430

D
dapan1121 已提交
2431
  *pVgList = NULL;
D
dapan1121 已提交
2432
  
D
dapan1121 已提交
2433
  CTG_ERR_JRET(ctgGetTableMeta(pCtg, pRpc, pMgmtEps, pTableName, &tbMeta, CTG_FLAG_UNKNOWN_STB));
D
dapan1121 已提交
2434

H
Haojun Liao 已提交
2435 2436
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);
D
dapan 已提交
2437

D
dapan1121 已提交
2438 2439 2440 2441 2442 2443 2444 2445 2446 2447
  SHashObj *vgHash = NULL;  
  CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pRpc, pMgmtEps, db, false, &dbCache, &vgInfo));

  if (dbCache) {
    vgHash = dbCache->vgInfo->vgHash;
  } else {
    vgHash = vgInfo->vgHash;
  }

  /* TODO REMOEV THIS ....
D
dapan 已提交
2448 2449 2450
  if (0 == tbMeta->vgId) {
    SVgroupInfo vgroup = {0};
    
D
dapan1121 已提交
2451
    catalogGetTableHashVgroup(pCtg, pRpc, pMgmtEps, pTableName, &vgroup);
D
dapan 已提交
2452 2453 2454

    tbMeta->vgId = vgroup.vgId;
  }
D
dapan1121 已提交
2455
  // TODO REMOVE THIS ....*/
D
dapan 已提交
2456

2457
  if (tbMeta->tableType == TSDB_SUPER_TABLE) {
D
dapan1121 已提交
2458
    CTG_ERR_JRET(ctgGenerateVgList(pCtg, vgHash, pVgList));
D
dapan1121 已提交
2459
  } else {
2460
    int32_t vgId = tbMeta->vgId;
D
dapan1121 已提交
2461
    if (NULL == taosHashGetClone(vgHash, &vgId, sizeof(vgId), &vgroupInfo)) {
2462
      ctgError("table's vgId not found in vgroup list, vgId:%d, tbName:%s", vgId, tNameGetTableName(pTableName));
D
dapan 已提交
2463
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);    
2464
    }
D
dapan1121 已提交
2465

D
dapan1121 已提交
2466 2467
    vgList = taosArrayInit(1, sizeof(SVgroupInfo));
    if (NULL == vgList) {
D
dapan1121 已提交
2468
      ctgError("taosArrayInit %d failed", (int32_t)sizeof(SVgroupInfo));
D
dapan 已提交
2469 2470 2471
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);    
    }

D
dapan1121 已提交
2472
    if (NULL == taosArrayPush(vgList, &vgroupInfo)) {
2473
      ctgError("taosArrayPush vgroupInfo to array failed, vgId:%d, tbName:%s", vgId, tNameGetTableName(pTableName));
D
dapan1121 已提交
2474 2475
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
    }
D
dapan 已提交
2476

D
dapan1121 已提交
2477
    *pVgList = vgList;
D
dapan1121 已提交
2478 2479
    vgList = NULL;
  }
D
dapan 已提交
2480

D
dapan1121 已提交
2481
_return:
D
dapan 已提交
2482

D
dapan1121 已提交
2483
  if (dbCache) {
D
dapan1121 已提交
2484 2485 2486 2487 2488 2489 2490 2491 2492
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
  }

  tfree(tbMeta);

  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
    tfree(vgInfo);
D
dapan1121 已提交
2493 2494 2495 2496 2497 2498
  }

  if (vgList) {
    taosArrayDestroy(vgList);
    vgList = NULL;
  }
D
dapan1121 已提交
2499
  
D
dapan1121 已提交
2500
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2501 2502 2503
}


D
dapan1121 已提交
2504
int32_t catalogGetTableHashVgroup(SCatalog *pCtg, void *pTrans, const SEpSet *pMgmtEps, const SName *pTableName, SVgroupInfo *pVgroup) {
D
dapan1121 已提交
2505 2506
  CTG_API_ENTER();

D
dapan1121 已提交
2507 2508 2509 2510 2511
  if (CTG_IS_INF_DBNAME(pTableName->dbname)) {
    ctgError("no valid vgInfo for db, dbname:%s", pTableName->dbname);
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2512 2513
  SCtgDBCache* dbCache = NULL;
  int32_t code = 0;
H
Haojun Liao 已提交
2514 2515
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);
D
dapan1121 已提交
2516

D
dapan1121 已提交
2517
  SDBVgInfo *vgInfo = NULL;
D
dapan1121 已提交
2518
  CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pTrans, pMgmtEps, db, false, &dbCache, &vgInfo));
D
dapan1121 已提交
2519

D
dapan1121 已提交
2520
  CTG_ERR_JRET(ctgGetVgInfoFromHashValue(pCtg, vgInfo ? vgInfo : dbCache->vgInfo, pTableName, pVgroup));
D
dapan1121 已提交
2521

D
dapan1121 已提交
2522
_return:
D
dapan1121 已提交
2523

D
dapan1121 已提交
2524
  if (dbCache) {
D
dapan1121 已提交
2525 2526 2527 2528 2529 2530 2531
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
  }

  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
    tfree(vgInfo);
D
dapan1121 已提交
2532
  }
D
dapan1121 已提交
2533

D
dapan1121 已提交
2534
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2535 2536 2537
}


D
dapan1121 已提交
2538
int32_t catalogGetAllMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SCatalogReq* pReq, SMetaData* pRsp) {
D
dapan1121 已提交
2539 2540
  CTG_API_ENTER();

D
dapan1121 已提交
2541
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pReq || NULL == pRsp) {
D
dapan1121 已提交
2542 2543 2544
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2545
  int32_t code = 0;
D
dapan1121 已提交
2546
  pRsp->pTableMeta = NULL;
D
dapan1121 已提交
2547 2548 2549

  if (pReq->pTableName) {
    int32_t tbNum = (int32_t)taosArrayGetSize(pReq->pTableName);
D
dapan1121 已提交
2550
    if (tbNum <= 0) {
D
dapan1121 已提交
2551
      ctgError("empty table name list, tbNum:%d", tbNum);
D
dapan1121 已提交
2552
      CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2553
    }
H
Haojun Liao 已提交
2554

D
dapan1121 已提交
2555 2556
    pRsp->pTableMeta = taosArrayInit(tbNum, POINTER_BYTES);
    if (NULL == pRsp->pTableMeta) {
D
dapan1121 已提交
2557
      ctgError("taosArrayInit %d failed", tbNum);
D
dapan1121 已提交
2558
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
2559 2560 2561 2562 2563 2564
    }
    
    for (int32_t i = 0; i < tbNum; ++i) {
      SName *name = taosArrayGet(pReq->pTableName, i);
      STableMeta *pTableMeta = NULL;
      
D
dapan1121 已提交
2565
      CTG_ERR_JRET(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, name, &pTableMeta, CTG_FLAG_UNKNOWN_STB));
D
dapan1121 已提交
2566 2567 2568 2569 2570 2571 2572 2573 2574

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

D
dapan1121 已提交
2575
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2576 2577

_return:  
D
dapan1121 已提交
2578

D
dapan1121 已提交
2579 2580 2581 2582 2583 2584 2585 2586
  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 已提交
2587
    pRsp->pTableMeta = NULL;
D
dapan1121 已提交
2588
  }
D
dapan 已提交
2589
  
D
dapan1121 已提交
2590
  CTG_API_LEAVE(code);
2591
}
D
dapan 已提交
2592

D
dapan1121 已提交
2593
int32_t catalogGetQnodeList(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SArray* pQnodeList) {
D
dapan1121 已提交
2594 2595
  CTG_API_ENTER();

D
dapan1121 已提交
2596 2597 2598 2599
  if (NULL == pCtg || NULL == pRpc  || NULL == pMgmtEps || NULL == pQnodeList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2600
  //TODO
D
dapan 已提交
2601

D
dapan1121 已提交
2602
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2603 2604
}

D
dapan1121 已提交
2605
int32_t catalogGetExpiredSTables(SCatalog* pCtg, SSTableMetaVersion **stables, uint32_t *num) {
D
dapan1121 已提交
2606 2607
  CTG_API_ENTER();

D
dapan1121 已提交
2608 2609
  if (NULL == pCtg || NULL == stables || NULL == num) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2610 2611
  }

D
dapan1121 已提交
2612 2613 2614 2615
  CTG_API_LEAVE(ctgMetaRentGet(&pCtg->stbRent, (void **)stables, num, sizeof(SSTableMetaVersion)));
}

int32_t catalogGetExpiredDBs(SCatalog* pCtg, SDbVgVersion **dbs, uint32_t *num) {
D
dapan1121 已提交
2616
  CTG_API_ENTER();
D
dapan1121 已提交
2617 2618 2619 2620
  
  if (NULL == pCtg || NULL == dbs || NULL == num) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
2621

D
dapan1121 已提交
2622
  CTG_API_LEAVE(ctgMetaRentGet(&pCtg->dbRent, (void **)dbs, num, sizeof(SDbVgVersion)));
D
dapan1121 已提交
2623 2624
}

D
dapan 已提交
2625

D
dapan 已提交
2626
void catalogDestroy(void) {
D
dapan1121 已提交
2627 2628
  qInfo("start to destroy catalog");
  
D
dapan 已提交
2629
  if (NULL == gCtgMgmt.pCluster || atomic_load_8(&gCtgMgmt.exit)) {
D
dapan1121 已提交
2630 2631 2632
    return;
  }

D
dapan 已提交
2633 2634 2635
  atomic_store_8(&gCtgMgmt.exit, true);

  tsem_post(&gCtgMgmt.sem);
D
dapan1121 已提交
2636

D
dapan1121 已提交
2637 2638 2639 2640
  while (CTG_IS_LOCKED(&gCtgMgmt.lock)) {
    usleep(1);
  }
  
D
dapan 已提交
2641
  CTG_LOCK(CTG_WRITE, &gCtgMgmt.lock);
D
dapan1121 已提交
2642

D
dapan1121 已提交
2643
  SCatalog *pCtg = NULL;
D
dapan 已提交
2644
  void *pIter = taosHashIterate(gCtgMgmt.pCluster, NULL);
D
dapan1121 已提交
2645
  while (pIter) {
D
dapan1121 已提交
2646
    pCtg = *(SCatalog **)pIter;
D
dapan1121 已提交
2647

D
dapan1121 已提交
2648 2649
    if (pCtg) {
      catalogFreeHandle(pCtg);
D
dapan1121 已提交
2650 2651
    }
    
D
dapan 已提交
2652
    pIter = taosHashIterate(gCtgMgmt.pCluster, pIter);
D
dapan 已提交
2653
  }
D
dapan1121 已提交
2654
  
D
dapan 已提交
2655 2656
  taosHashCleanup(gCtgMgmt.pCluster);
  gCtgMgmt.pCluster = NULL;
D
dapan1121 已提交
2657

D
dapan 已提交
2658
  CTG_UNLOCK(CTG_WRITE, &gCtgMgmt.lock);
D
dapan1121 已提交
2659

D
dapan1121 已提交
2660
  qInfo("catalog destroyed");
D
dapan 已提交
2661 2662 2663 2664
}