catalog.c 75.1 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
dapan 已提交
1334
  SDbVgVersion vgVersion = {.dbId = dbId, .vgVersion = dbInfo->vgVersion, .numOfTable = dbInfo->numOfTable};
D
dapan1121 已提交
1335 1336 1337 1338 1339 1340 1341 1342 1343 1344

  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
  if (dbCache->vgInfo) {
D
dapan 已提交
1347 1348 1349 1350 1351 1352 1353 1354 1355
    if (dbInfo->vgVersion < dbCache->vgInfo->vgVersion) {
      ctgDebug("db vgVersion is old, dbFName:%s, vgVersion:%d, currentVersion:%d", dbFName, dbInfo->vgVersion, dbCache->vgInfo->vgVersion);
      ctgWReleaseVgInfo(dbCache);
      
      return TSDB_CODE_SUCCESS;
    }

    if (dbInfo->vgVersion == dbCache->vgInfo->vgVersion && dbInfo->numOfTable == dbCache->vgInfo->numOfTable) {
      ctgDebug("no new db vgVersion or numOfTable, dbFName:%s, vgVersion:%d, numOfTable:%d", dbFName, dbInfo->vgVersion, dbInfo->numOfTable);
D
dapan1121 已提交
1356
      ctgWReleaseVgInfo(dbCache);
D
dapan1121 已提交
1357
      
D
dapan1121 已提交
1358
      return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1359
    }
D
dapan1121 已提交
1360 1361

    ctgFreeVgInfo(dbCache->vgInfo);
D
dapan1121 已提交
1362 1363
  }

D
dapan1121 已提交
1364
  dbCache->vgInfo = dbInfo;
D
dapan1121 已提交
1365

D
dapan1121 已提交
1366
  *pDbInfo = NULL;
D
dapan1121 已提交
1367

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

D
dapan1121 已提交
1370
  ctgWReleaseVgInfo(dbCache);
D
dapan1121 已提交
1371

D
dapan1121 已提交
1372
  dbCache = NULL;
D
dapan1121 已提交
1373

D
dapan1121 已提交
1374 1375 1376
  strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));
  CTG_ERR_RET(ctgMetaRentUpdate(&pCtg->dbRent, &vgVersion, vgVersion.dbId, sizeof(SDbVgVersion), ctgDbVgVersionCompare));
  
D
dapan1121 已提交
1377 1378 1379 1380
  CTG_RET(code);
}


D
dapan1121 已提交
1381 1382
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 已提交
1383

D
dapan1121 已提交
1384 1385 1386 1387 1388
  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 已提交
1389 1390
  }

D
dapan1121 已提交
1391 1392 1393 1394 1395 1396 1397
  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 已提交
1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408
    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 已提交
1409
      }
D
dapan1121 已提交
1410

D
dapan1121 已提交
1411
      origSuid = orig->suid;
D
dapan1121 已提交
1412
    }
D
dapan1121 已提交
1413
  }
D
dapan1121 已提交
1414

D
dapan1121 已提交
1415 1416
  if (isStb) {
    CTG_LOCK(CTG_WRITE, &tbCache->stbLock);
D
dapan1121 已提交
1417
  }
D
dapan1121 已提交
1418
  
D
dapan1121 已提交
1419 1420 1421 1422 1423 1424 1425 1426 1427
  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 已提交
1428

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

D
dapan1121 已提交
1432 1433 1434
  if (!isStb) {
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);  
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1435
  }
D
dapan1121 已提交
1436

D
dapan1121 已提交
1437
  if (origType == TSDB_SUPER_TABLE && origSuid == meta->suid) {
D
dapan1121 已提交
1438 1439 1440 1441
    CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);  
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
1442

D
dapan1121 已提交
1443 1444 1445 1446 1447 1448
  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 已提交
1449
  }
D
dapan1121 已提交
1450 1451
  
  CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);
D
dapan1121 已提交
1452

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

D
dapan 已提交
1455
  ctgDebug("stb updated to stbCache, dbFName:%s, tbName:%s, tbType:%d", dbFName, tbName, meta->tableType);
D
dapan1121 已提交
1456 1457 1458 1459 1460 1461 1462

  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 已提交
1463 1464
}

D
dapan 已提交
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 1497 1498 1499 1500 1501 1502 1503
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 已提交
1504
int32_t ctgGetDBVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* dbFName, bool forceUpdate, SCtgDBCache** dbCache, SDBVgInfo **pInfo) {
D
dapan1121 已提交
1505
  bool inCache = false;
D
dapan1121 已提交
1506
  int32_t code = 0;
D
dapan1121 已提交
1507 1508 1509 1510 1511

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

  if (inCache && !forceUpdate) {
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1512 1513 1514 1515 1516
  }

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

D
dapan1121 已提交
1517
  tstrncpy(input.db, dbFName, tListLen(input.db));
D
dapan1121 已提交
1518 1519 1520
  if (inCache) {
    input.dbId = (*dbCache)->dbId;
    input.vgVersion = (*dbCache)->vgInfo->vgVersion;
D
dapan 已提交
1521
    input.numOfTable = (*dbCache)->vgInfo->numOfTable;
D
dapan1121 已提交
1522 1523 1524
  } else {
    input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
  }
H
Haojun Liao 已提交
1525

D
dapan1121 已提交
1526 1527 1528 1529 1530 1531 1532 1533 1534
  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 已提交
1535

D
dapan 已提交
1536 1537 1538 1539 1540 1541 1542 1543
  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 已提交
1544 1545
  }

D
dapan 已提交
1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556
  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 已提交
1557
  return TSDB_CODE_SUCCESS;
D
dapan 已提交
1558 1559 1560 1561 1562 1563 1564 1565 1566

_return:

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

D
dapan 已提交
1569

D
dapan1121 已提交
1570 1571 1572 1573 1574
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 已提交
1575 1576
  }

D
dapan1121 已提交
1577 1578 1579 1580 1581 1582
  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 已提交
1583
      qError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
D
dapan1121 已提交
1584 1585 1586 1587 1588
      tfree(*pOutput);
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }

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

D
dapan1121 已提交
1591 1592 1593
  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
1594 1595


D
dapan1121 已提交
1596 1597
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 已提交
1598 1599 1600 1601 1602 1603
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }

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

D
dapan1121 已提交
1604
  if (!CTG_FLAG_IS_INF_DB(flag)) {
D
dapan1121 已提交
1605 1606
    CTG_ERR_RET(catalogGetTableHashVgroup(pCtg, pTrans, pMgmtEps, pTableName, &vgroupInfo));
  }
D
dapan1121 已提交
1607

D
dapan 已提交
1608
  SCtgUpdateTblMsg *msg = NULL;
D
dapan1121 已提交
1609
  STableMetaOutput  moutput = {0};
S
Shengliang Guan 已提交
1610
  STableMetaOutput *output = calloc(1, sizeof(STableMetaOutput));
D
dapan1121 已提交
1611 1612 1613 1614
  if (NULL == output) {
    ctgError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }
D
dapan1121 已提交
1615

D
dapan1121 已提交
1616
  if (CTG_FLAG_IS_INF_DB(flag)) {
D
dapan1121 已提交
1617 1618 1619
    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 已提交
1620
  } else if (CTG_FLAG_IS_STB(flag)) {
D
dapan 已提交
1621
    ctgDebug("will refresh tbmeta, supposed to be stb, tbName:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
1622 1623

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

D
dapan1121 已提交
1626
    if (CTG_IS_META_NULL(output->metaType)) {
D
dapan1121 已提交
1627
      CTG_ERR_JRET(ctgGetTableMetaFromVnode(pCtg, pTrans, pMgmtEps, pTableName, &vgroupInfo, output));
D
dapan1121 已提交
1628 1629
    }
  } else {
D
dapan1121 已提交
1630
    ctgDebug("will refresh tbmeta, not supposed to be stb, tbName:%s, flag:%d", tNameGetTableName(pTableName), flag);
D
dapan1121 已提交
1631 1632

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

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

D
dapan1121 已提交
1638
      tfree(output->tbMeta);
D
dapan1121 已提交
1639
      
D
dapan1121 已提交
1640
      CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, output->dbFName, output->tbName, output));
D
dapan1121 已提交
1641
    } else if (CTG_IS_META_BOTH(output->metaType)) {
D
dapan1121 已提交
1642
      int32_t exist = 0;
D
dapan1121 已提交
1643 1644 1645 1646
      if (!CTG_FLAG_IS_FORCE_UPDATE(flag)) {
        CTG_ERR_JRET(ctgIsTableMetaExistInCache(pCtg, output->dbFName, output->tbName, &exist));
      }
      
D
dapan1121 已提交
1647
      if (0 == exist) {
D
dapan1121 已提交
1648
        CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, output->dbFName, output->tbName, &moutput));
D
dapan1121 已提交
1649

D
dapan1121 已提交
1650
        if (CTG_IS_META_NULL(moutput.metaType)) {
D
dapan1121 已提交
1651
          SET_META_TYPE_NULL(output->metaType);
D
dapan1121 已提交
1652 1653
        }
        
D
dapan1121 已提交
1654 1655
        tfree(output->tbMeta);
        output->tbMeta = moutput.tbMeta;
D
dapan1121 已提交
1656 1657
        moutput.tbMeta = NULL;
      } else {
D
dapan1121 已提交
1658
        tfree(output->tbMeta);
D
dapan1121 已提交
1659
        
D
dapan1121 已提交
1660
        SET_META_TYPE_CTABLE(output->metaType); 
D
dapan1121 已提交
1661
      }
D
dapan1121 已提交
1662 1663 1664
    }
  }

D
dapan1121 已提交
1665
  if (CTG_IS_META_NULL(output->metaType)) {
D
dapan 已提交
1666
    ctgError("no tbmeta got, tbNmae:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
1667 1668 1669
    CTG_ERR_JRET(CTG_ERR_CODE_TABLE_NOT_EXIST);
  }

D
dapan 已提交
1670 1671 1672 1673 1674 1675
  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 已提交
1676 1677 1678 1679 1680
  if (pOutput) {
    CTG_ERR_JRET(ctgCloneMetaOutput(output, pOutput));
  }

  SCtgMetaAction action= {.act = CTG_ACT_UPDATE_TBL};
D
dapan 已提交
1681
  msg = malloc(sizeof(SCtgUpdateTblMsg));
D
dapan1121 已提交
1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693
  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 已提交
1694 1695
  ctgDebug("action [%s] added into queue", gCtgAction[action.act].name);

D
dapan1121 已提交
1696
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1697 1698 1699

_return:

D
dapan1121 已提交
1700
  tfree(output->tbMeta);
D
dapan 已提交
1701
  tfree(output);
D
dapan1121 已提交
1702
  tfree(msg);
D
dapan1121 已提交
1703 1704 1705 1706
  
  CTG_RET(code);
}

D
dapan1121 已提交
1707
int32_t ctgGetTableMeta(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t flag) {
D
dapan1121 已提交
1708
  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == pTableMeta) {
D
dapan1121 已提交
1709 1710 1711 1712
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }
  
  int32_t exist = 0;
D
dapan1121 已提交
1713
  int32_t code = 0;
D
dapan1121 已提交
1714 1715 1716
  uint64_t dbId = 0;
  uint64_t suid = 0;
  STableMetaOutput *output = NULL;
D
dapan1121 已提交
1717

D
dapan1121 已提交
1718
  if (CTG_IS_INF_DBNAME(pTableName->dbname)) {
D
dapan1121 已提交
1719
    CTG_FLAG_SET_INF_DB(flag);
D
dapan1121 已提交
1720 1721
  }

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

D
dapan1121 已提交
1724 1725 1726 1727 1728
  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 已提交
1729
    }
D
dapan1121 已提交
1730

D
dapan1121 已提交
1731 1732
    tbType = (*pTableMeta)->tableType;
    suid = (*pTableMeta)->suid;
D
dapan1121 已提交
1733

D
dapan1121 已提交
1734 1735 1736 1737 1738
    tfree(*pTableMeta);    
  }
    
  if (CTG_FLAG_IS_UNKNOWN_STB(flag)) {
    CTG_FLAG_SET_STB(flag, tbType);
D
dapan1121 已提交
1739 1740
  }

D
dapan1121 已提交
1741

D
dapan 已提交
1742
  while (true) {
D
dapan1121 已提交
1743
    CTG_ERR_JRET(ctgRefreshTblMeta(pCtg, pRpc, pMgmtEps, pTableName, flag, &output));
D
dapan1121 已提交
1744

D
dapan 已提交
1745 1746 1747 1748
    if (CTG_IS_META_TABLE(output->metaType)) {
      *pTableMeta = output->tbMeta;
      goto _return;
    }
D
dapan1121 已提交
1749

D
dapan 已提交
1750 1751 1752 1753 1754 1755
    if (CTG_IS_META_BOTH(output->metaType)) {
      memcpy(output->tbMeta, &output->ctbMeta, sizeof(output->ctbMeta));
      
      *pTableMeta = output->tbMeta;
      goto _return;
    }
D
dapan1121 已提交
1756

D
dapan 已提交
1757 1758 1759 1760 1761
    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 已提交
1762

D
dapan 已提交
1763
    // HANDLE ONLY CHILD TABLE META
D
dapan1121 已提交
1764

D
dapan 已提交
1765 1766 1767
    SName stbName = *pTableName;
    strcpy(stbName.tname, output->tbName);
    
D
dapan1121 已提交
1768
    CTG_ERR_JRET(ctgGetTableMetaFromCache(pCtg, &stbName, pTableMeta, &exist, flag, NULL));
D
dapan 已提交
1769 1770 1771 1772
    if (0 == exist) {
      ctgDebug("stb no longer exist, dbFName:%s, tbName:%s", output->dbFName, pTableName->tname);
      continue;
    }
D
dapan1121 已提交
1773

D
dapan 已提交
1774 1775 1776 1777
    memcpy(*pTableMeta, &output->ctbMeta, sizeof(output->ctbMeta));

    break;
  }
D
dapan1121 已提交
1778 1779 1780

_return:

D
dapan1121 已提交
1781 1782 1783 1784 1785 1786 1787 1788 1789 1790 1791 1792 1793 1794 1795
  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 已提交
1796 1797
  tfree(output);

D
dapan 已提交
1798 1799
  if (*pTableMeta) {
    ctgDebug("tbmeta returned, tbName:%s, tbType:%d", pTableName->tname, (*pTableMeta)->tableType);
D
dapan1121 已提交
1800
    ctgDbgShowTableMeta(pCtg, pTableName->tname, *pTableMeta);
D
dapan 已提交
1801 1802
  }

D
dapan1121 已提交
1803 1804 1805 1806 1807 1808 1809 1810 1811 1812 1813 1814 1815
  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 已提交
1816
  ctgFreeVgInfo(msg->dbInfo);
D
dapan1121 已提交
1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827
  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 已提交
1828
  ctgGetDBCache(msg->pCtg, msg->dbFName, &dbCache);
D
dapan1121 已提交
1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856
  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 已提交
1857
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
1858 1859 1860 1861 1862 1863 1864
  }

  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 已提交
1865 1866 1867 1868 1869
  char *p = strchr(output->dbFName, '.');
  if (p && CTG_IS_INF_DBNAME(p + 1)) {
    memmove(output->dbFName, p + 1, strlen(p + 1));
  }
  
D
dapan1121 已提交
1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885 1886 1887
  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 已提交
1888 1889 1890
  if (output) {
    tfree(output->tbMeta);
    tfree(output);
D
dapan1121 已提交
1891
  }
D
dapan 已提交
1892
  
D
dapan1121 已提交
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 1919 1920 1921 1922 1923 1924 1925 1926
  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 已提交
1927
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
1928 1929 1930 1931 1932 1933 1934
  }  
  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 已提交
1935
  CTG_ERR_JRET(ctgMetaRentRemove(&msg->pCtg->stbRent, msg->suid, ctgStbVersionCompare));
D
dapan1121 已提交
1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946
  
  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 已提交
1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964 1965 1966 1967 1968 1969 1970 1971 1972
  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 已提交
1973

D
dapan1121 已提交
1974 1975 1976
  tfree(msg);
  
  CTG_RET(code);
D
dapan1121 已提交
1977 1978 1979 1980 1981 1982 1983 1984
}


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

  qInfo("catalog update thread started");

D
dapan 已提交
1985
  CTG_LOCK(CTG_READ, &gCtgMgmt.lock);
D
dapan1121 已提交
1986 1987
  
  while (true) {
D
dapan 已提交
1988
    tsem_wait(&gCtgMgmt.sem);
D
dapan1121 已提交
1989
    
D
dapan 已提交
1990
    if (atomic_load_8(&gCtgMgmt.exit)) {
D
dapan1121 已提交
1991 1992 1993 1994 1995
      break;
    }

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

D
dapan1121 已提交
1998
    ctgDebug("process [%s] action", gCtgAction[action->act].name);
D
dapan 已提交
1999 2000 2001 2002
    
    (*gCtgAction[action->act].func)(action);

    CTG_STAT_ADD(gCtgMgmt.stat.runtime.qDoneNum); 
D
dapan1121 已提交
2003 2004

    ctgDbgShowClusterCache(pCtg);
D
dapan1121 已提交
2005 2006
  }

D
dapan 已提交
2007
  CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock);
D
dapan1121 已提交
2008 2009 2010 2011 2012 2013 2014 2015 2016 2017 2018 2019

  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 已提交
2020
  if (pthread_create(&gCtgMgmt.updateThread, &thAttr, ctgUpdateThreadFunc, NULL) != 0) {
D
dapan1121 已提交
2021 2022
    terrno = TAOS_SYSTEM_ERROR(errno);
    CTG_ERR_RET(terrno);
D
dapan1121 已提交
2023 2024
  }
  
D
dapan1121 已提交
2025
  pthread_attr_destroy(&thAttr);
D
dapan1121 已提交
2026 2027 2028 2029
  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
2030
int32_t catalogInit(SCatalogCfg *cfg) {
D
dapan 已提交
2031
  if (gCtgMgmt.pCluster) {
D
dapan 已提交
2032
    qError("catalog already initialized");
D
dapan1121 已提交
2033
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2034 2035
  }

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

D
dapan1121 已提交
2038
  if (cfg) {
D
dapan 已提交
2039
    memcpy(&gCtgMgmt.cfg, cfg, sizeof(*cfg));
H
Haojun Liao 已提交
2040

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

D
dapan 已提交
2045 2046
    if (gCtgMgmt.cfg.maxTblCacheNum == 0) {
      gCtgMgmt.cfg.maxTblCacheNum = CTG_DEFAULT_CACHE_TBLMETA_NUMBER;
D
dapan1121 已提交
2047
    }
D
dapan1121 已提交
2048

D
dapan 已提交
2049 2050
    if (gCtgMgmt.cfg.dbRentSec == 0) {
      gCtgMgmt.cfg.dbRentSec = CTG_DEFAULT_RENT_SECOND;
D
dapan1121 已提交
2051 2052
    }

D
dapan 已提交
2053 2054
    if (gCtgMgmt.cfg.stbRentSec == 0) {
      gCtgMgmt.cfg.stbRentSec = CTG_DEFAULT_RENT_SECOND;
D
dapan1121 已提交
2055
    }
D
dapan1121 已提交
2056
  } else {
D
dapan 已提交
2057 2058 2059 2060
    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 已提交
2061 2062
  }

D
dapan 已提交
2063 2064
  gCtgMgmt.pCluster = taosHashInit(CTG_DEFAULT_CACHE_CLUSTER_NUMBER, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
  if (NULL == gCtgMgmt.pCluster) {
D
dapan1121 已提交
2065 2066
    qError("taosHashInit %d cluster cache failed", CTG_DEFAULT_CACHE_CLUSTER_NUMBER);
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
2067 2068
  }

D
dapan1121 已提交
2069 2070
  CTG_ERR_RET(ctgStartUpdateThread());

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

D
dapan 已提交
2073 2074
  gCtgMgmt.head = calloc(1, sizeof(SCtgQNode));
  if (NULL == gCtgMgmt.head) {
D
dapan1121 已提交
2075 2076 2077
    qError("calloc %d failed", (int32_t)sizeof(SCtgQNode));
    CTG_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }
D
dapan 已提交
2078
  gCtgMgmt.tail = gCtgMgmt.head;
D
dapan1121 已提交
2079

D
dapan 已提交
2080
  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 已提交
2081

D
dapan 已提交
2082
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
2083 2084
}

D
dapan1121 已提交
2085
int32_t catalogGetHandle(uint64_t clusterId, SCatalog** catalogHandle) {
2086
  if (NULL == catalogHandle) {
D
dapan1121 已提交
2087
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
2088 2089
  }

D
dapan 已提交
2090
  if (NULL == gCtgMgmt.pCluster) {
D
dapan 已提交
2091
    qError("catalog cluster cache are not ready, clusterId:%"PRIx64, clusterId);
D
dapan1121 已提交
2092
    CTG_ERR_RET(TSDB_CODE_CTG_NOT_READY);
D
dapan 已提交
2093 2094
  }

D
dapan1121 已提交
2095 2096
  int32_t code = 0;
  SCatalog *clusterCtg = NULL;
D
dapan 已提交
2097

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

D
dapan1121 已提交
2101 2102 2103 2104 2105
    if (ctg && (*ctg)) {
      *catalogHandle = *ctg;
      qDebug("got catalog handle from cache, clusterId:%"PRIx64", CTG:%p", clusterId, *ctg);
      return TSDB_CODE_SUCCESS;
    }
D
dapan 已提交
2106

D
dapan1121 已提交
2107 2108 2109 2110 2111 2112
    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 已提交
2113 2114
    clusterCtg->clusterId = clusterId;

D
dapan 已提交
2115 2116
    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 已提交
2117

D
dapan 已提交
2118
    clusterCtg->dbCache = taosHashInit(gCtgMgmt.cfg.maxDBCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
D
dapan1121 已提交
2119 2120 2121 2122 2123
    if (NULL == clusterCtg->dbCache) {
      qError("taosHashInit %d dbCache failed", CTG_DEFAULT_CACHE_DB_NUMBER);
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
    }

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

  *catalogHandle = clusterCtg;
D
dapan 已提交
2147
  
D
dapan1121 已提交
2148
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
2149 2150 2151 2152 2153 2154 2155 2156

_return:

  ctgFreeHandle(clusterCtg);
  
  CTG_RET(code);
}

D
dapan1121 已提交
2157 2158
void catalogFreeHandle(SCatalog* pCtg) {
  if (NULL == pCtg) {
D
dapan1121 已提交
2159 2160
    return;
  }
D
dapan1121 已提交
2161

D
dapan 已提交
2162
  if (taosHashRemove(gCtgMgmt.pCluster, &pCtg->clusterId, sizeof(pCtg->clusterId))) {
D
dapan1121 已提交
2163
    ctgWarn("taosHashRemove from cluster failed, may already be freed, clusterId:%"PRIx64, pCtg->clusterId);
D
dapan1121 已提交
2164 2165 2166
    return;
  }

D
dapan1121 已提交
2167
  uint64_t clusterId = pCtg->clusterId;
D
dapan1121 已提交
2168
  
D
dapan1121 已提交
2169
  ctgFreeHandle(pCtg);
D
dapan1121 已提交
2170 2171

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

D
dapan1121 已提交
2174
int32_t catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* version, int64_t* dbId) {
D
dapan1121 已提交
2175 2176
  CTG_API_ENTER();

D
dapan1121 已提交
2177
  if (NULL == pCtg || NULL == dbFName || NULL == version || NULL == dbId) {
D
dapan1121 已提交
2178 2179 2180 2181
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2182
    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
2183
    ctgInfo("empty db cache, dbFName:%s", dbFName);
D
dapan1121 已提交
2184
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2185 2186
  }

D
dapan1121 已提交
2187 2188 2189
  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
  if (NULL == dbCache) {
D
dapan1121 已提交
2190
    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
2191
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2192 2193
  }

D
dapan 已提交
2194 2195 2196
  bool inCache = false;
  ctgAcquireVgInfo(pCtg, dbCache, &inCache);
  if (!inCache) {
D
dapan1121 已提交
2197
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
2198 2199

    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
2200
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2201 2202
  }

D
dapan1121 已提交
2203
  *version = dbCache->vgInfo->vgVersion;
D
dapan1121 已提交
2204
  *dbId = dbCache->dbId;
D
dapan1121 已提交
2205 2206 2207

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

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

D
dapan1121 已提交
2211
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2212 2213
}

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

D
dapan1121 已提交
2217 2218 2219
  if (NULL == pCtg || NULL == dbFName || NULL == pRpc || NULL == pMgmtEps || NULL == vgroupList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
2220

D
dapan1121 已提交
2221
  SCtgDBCache* dbCache = NULL;
2222
  int32_t code = 0;
D
dapan1121 已提交
2223
  SArray *vgList = NULL;
D
dapan1121 已提交
2224 2225 2226 2227 2228 2229 2230
  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 已提交
2231 2232
  }

D
dapan1121 已提交
2233
  CTG_ERR_JRET(ctgGenerateVgList(pCtg, vgHash, &vgList));
D
dapan1121 已提交
2234 2235 2236 2237 2238

  *vgroupList = vgList;
  vgList = NULL;

_return:
D
dapan1121 已提交
2239 2240

  if (dbCache) {
D
dapan1121 已提交
2241 2242
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
2243 2244
  }

D
dapan1121 已提交
2245 2246 2247
  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
    tfree(vgInfo);
D
dapan1121 已提交
2248 2249
  }

D
dapan1121 已提交
2250
  CTG_API_LEAVE(code);  
D
dapan1121 已提交
2251 2252 2253
}


D
dapan1121 已提交
2254
int32_t catalogUpdateDBVgInfo(SCatalog* pCtg, const char* dbFName, uint64_t dbId, SDBVgInfo* dbInfo) {
D
dapan1121 已提交
2255
  CTG_API_ENTER();
D
dapan1121 已提交
2256 2257

  int32_t code = 0;
D
dapan1121 已提交
2258
  
D
dapan1121 已提交
2259
  if (NULL == pCtg || NULL == dbFName || NULL == dbInfo) {
D
dapan1121 已提交
2260 2261 2262
    CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2263 2264 2265 2266
  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 已提交
2267
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
2268 2269
  }

D
dapan1121 已提交
2270 2271 2272 2273
  msg->pCtg = pCtg;
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  msg->dbId = dbId;
  msg->dbInfo = dbInfo;
D
dapan1121 已提交
2274

D
dapan1121 已提交
2275
  action.data = msg;
D
dapan1121 已提交
2276

D
dapan1121 已提交
2277
  CTG_ERR_JRET(ctgPushAction(&action));
D
dapan1121 已提交
2278

D
dapan1121 已提交
2279 2280
  dbInfo = NULL;

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

D
dapan1121 已提交
2283 2284
  CTG_API_LEAVE(code);
  
D
dapan1121 已提交
2285 2286
_return:

D
dapan1121 已提交
2287
  ctgFreeVgInfo(dbInfo);
D
dapan1121 已提交
2288 2289

  tfree(msg);
D
dapan1121 已提交
2290
  
D
dapan1121 已提交
2291
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2292 2293 2294
}


D
dapan1121 已提交
2295 2296 2297
int32_t catalogRemoveDB(SCatalog* pCtg, const char* dbFName, uint64_t dbId) {
  CTG_API_ENTER();

D
dapan1121 已提交
2298 2299
  int32_t code = 0;
  
D
dapan1121 已提交
2300 2301
  if (NULL == pCtg || NULL == dbFName) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2302 2303
  }

D
dapan1121 已提交
2304
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2305
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2306
  }
D
dapan1121 已提交
2307

D
dapan1121 已提交
2308
  CTG_ERR_JRET(ctgPushRmDBMsgInQueue(pCtg, dbFName, dbId));
D
dapan 已提交
2309

D
dapan1121 已提交
2310
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2311
  
D
dapan1121 已提交
2312 2313
_return:

D
dapan1121 已提交
2314
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2315 2316
}

D
dapan1121 已提交
2317 2318 2319
int32_t catalogRemoveStbMeta(SCatalog* pCtg, const char* dbFName, uint64_t dbId, const char* stbName, uint64_t suid) {
  CTG_API_ENTER();

D
dapan 已提交
2320 2321
  int32_t code = 0;
  
D
dapan1121 已提交
2322 2323
  if (NULL == pCtg || NULL == dbFName || NULL == stbName) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
2324 2325
  }

D
dapan1121 已提交
2326
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2327
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2328
  }
D
dapan1121 已提交
2329

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

D
dapan1121 已提交
2332
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2333
  
D
dapan1121 已提交
2334 2335
_return:

D
dapan1121 已提交
2336
  CTG_API_LEAVE(code);
D
dapan 已提交
2337 2338
}

D
dapan1121 已提交
2339

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

D
dapan1121 已提交
2343
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_UNKNOWN_STB));
D
dapan1121 已提交
2344
}
D
dapan1121 已提交
2345

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

D
dapan1121 已提交
2349
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_STB));
D
dapan1121 已提交
2350 2351
}

D
dapan1121 已提交
2352 2353 2354 2355 2356 2357 2358 2359 2360 2361 2362 2363 2364
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 已提交
2365 2366
  int32_t code = 0;

D
dapan1121 已提交
2367 2368
  strcpy(output->dbFName, rspMsg->dbFName);
  strcpy(output->tbName, rspMsg->tbName);
D
dapan1121 已提交
2369

D
dapan1121 已提交
2370
  output->dbId = rspMsg->dbId;
D
dapan1121 已提交
2371
  
D
dapan1121 已提交
2372
  SET_META_TYPE_TABLE(output->metaType);
D
dapan1121 已提交
2373
  
D
dapan1121 已提交
2374
  CTG_ERR_JRET(queryCreateTableMetaFromMsg(rspMsg, true, &output->tbMeta));
D
dapan1121 已提交
2375

D
dapan1121 已提交
2376 2377 2378 2379 2380 2381
  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 已提交
2382

D
dapan1121 已提交
2383 2384 2385 2386 2387 2388 2389
  msg->pCtg = pCtg;
  msg->output = output;

  action.data = msg;

  CTG_ERR_JRET(ctgPushAction(&action));

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

D
dapan1121 已提交
2392 2393
  CTG_API_LEAVE(code);
  
D
dapan1121 已提交
2394 2395
_return:

D
dapan1121 已提交
2396 2397 2398
  tfree(output->tbMeta);
  tfree(output);
  tfree(msg);
D
dapan1121 已提交
2399
  
D
dapan1121 已提交
2400
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2401 2402 2403
}


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

D
dapan1121 已提交
2407
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pTableName) {
D
dapan1121 已提交
2408 2409 2410
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2411
  CTG_API_LEAVE(ctgRefreshTblMeta(pCtg, pTrans, pMgmtEps, pTableName, CTG_FLAG_FORCE_UPDATE | CTG_FLAG_MAKE_STB(isSTable), NULL));
2412
}
2413

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

D
dapan1121 已提交
2417
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_FORCE_UPDATE | CTG_FLAG_MAKE_STB(isSTable)));
D
dapan1121 已提交
2418 2419
}

D
dapan1121 已提交
2420
int32_t catalogGetTableDistVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgList) {
D
dapan1121 已提交
2421
  CTG_API_ENTER();
D
dapan1121 已提交
2422 2423 2424 2425

  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == pVgList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
2426 2427 2428 2429 2430

  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 已提交
2431 2432 2433 2434
  
  STableMeta *tbMeta = NULL;
  int32_t code = 0;
  SVgroupInfo vgroupInfo = {0};
D
dapan1121 已提交
2435
  SCtgDBCache* dbCache = NULL;
D
dapan1121 已提交
2436
  SArray *vgList = NULL;
D
dapan 已提交
2437
  SDBVgInfo *vgInfo = NULL;
D
dapan1121 已提交
2438

D
dapan1121 已提交
2439
  *pVgList = NULL;
D
dapan1121 已提交
2440
  
D
dapan1121 已提交
2441
  CTG_ERR_JRET(ctgGetTableMeta(pCtg, pRpc, pMgmtEps, pTableName, &tbMeta, CTG_FLAG_UNKNOWN_STB));
D
dapan1121 已提交
2442

H
Haojun Liao 已提交
2443 2444
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);
D
dapan 已提交
2445

D
dapan1121 已提交
2446 2447 2448 2449 2450 2451 2452 2453 2454 2455
  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 已提交
2456 2457 2458
  if (0 == tbMeta->vgId) {
    SVgroupInfo vgroup = {0};
    
D
dapan1121 已提交
2459
    catalogGetTableHashVgroup(pCtg, pRpc, pMgmtEps, pTableName, &vgroup);
D
dapan 已提交
2460 2461 2462

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

2465
  if (tbMeta->tableType == TSDB_SUPER_TABLE) {
D
dapan1121 已提交
2466
    CTG_ERR_JRET(ctgGenerateVgList(pCtg, vgHash, pVgList));
D
dapan1121 已提交
2467
  } else {
2468
    int32_t vgId = tbMeta->vgId;
D
dapan1121 已提交
2469
    if (NULL == taosHashGetClone(vgHash, &vgId, sizeof(vgId), &vgroupInfo)) {
2470
      ctgError("table's vgId not found in vgroup list, vgId:%d, tbName:%s", vgId, tNameGetTableName(pTableName));
D
dapan 已提交
2471
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);    
2472
    }
D
dapan1121 已提交
2473

D
dapan1121 已提交
2474 2475
    vgList = taosArrayInit(1, sizeof(SVgroupInfo));
    if (NULL == vgList) {
D
dapan1121 已提交
2476
      ctgError("taosArrayInit %d failed", (int32_t)sizeof(SVgroupInfo));
D
dapan 已提交
2477 2478 2479
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);    
    }

D
dapan1121 已提交
2480
    if (NULL == taosArrayPush(vgList, &vgroupInfo)) {
2481
      ctgError("taosArrayPush vgroupInfo to array failed, vgId:%d, tbName:%s", vgId, tNameGetTableName(pTableName));
D
dapan1121 已提交
2482 2483
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
    }
D
dapan 已提交
2484

D
dapan1121 已提交
2485
    *pVgList = vgList;
D
dapan1121 已提交
2486 2487
    vgList = NULL;
  }
D
dapan 已提交
2488

D
dapan1121 已提交
2489
_return:
D
dapan 已提交
2490

D
dapan1121 已提交
2491
  if (dbCache) {
D
dapan1121 已提交
2492 2493 2494 2495 2496 2497 2498 2499 2500
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
  }

  tfree(tbMeta);

  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
    tfree(vgInfo);
D
dapan1121 已提交
2501 2502 2503 2504 2505 2506
  }

  if (vgList) {
    taosArrayDestroy(vgList);
    vgList = NULL;
  }
D
dapan1121 已提交
2507
  
D
dapan1121 已提交
2508
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2509 2510 2511
}


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

D
dapan1121 已提交
2515 2516 2517 2518 2519
  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 已提交
2520 2521
  SCtgDBCache* dbCache = NULL;
  int32_t code = 0;
H
Haojun Liao 已提交
2522 2523
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);
D
dapan1121 已提交
2524

D
dapan1121 已提交
2525
  SDBVgInfo *vgInfo = NULL;
D
dapan1121 已提交
2526
  CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pTrans, pMgmtEps, db, false, &dbCache, &vgInfo));
D
dapan1121 已提交
2527

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

D
dapan1121 已提交
2530
_return:
D
dapan1121 已提交
2531

D
dapan1121 已提交
2532
  if (dbCache) {
D
dapan1121 已提交
2533 2534 2535 2536 2537 2538 2539
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
  }

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

D
dapan1121 已提交
2542
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2543 2544 2545
}


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

D
dapan1121 已提交
2549
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pReq || NULL == pRsp) {
D
dapan1121 已提交
2550 2551 2552
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2553
  int32_t code = 0;
D
dapan1121 已提交
2554
  pRsp->pTableMeta = NULL;
D
dapan1121 已提交
2555 2556 2557

  if (pReq->pTableName) {
    int32_t tbNum = (int32_t)taosArrayGetSize(pReq->pTableName);
D
dapan1121 已提交
2558
    if (tbNum <= 0) {
D
dapan1121 已提交
2559
      ctgError("empty table name list, tbNum:%d", tbNum);
D
dapan1121 已提交
2560
      CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2561
    }
H
Haojun Liao 已提交
2562

D
dapan1121 已提交
2563 2564
    pRsp->pTableMeta = taosArrayInit(tbNum, POINTER_BYTES);
    if (NULL == pRsp->pTableMeta) {
D
dapan1121 已提交
2565
      ctgError("taosArrayInit %d failed", tbNum);
D
dapan1121 已提交
2566
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
2567 2568 2569 2570 2571 2572
    }
    
    for (int32_t i = 0; i < tbNum; ++i) {
      SName *name = taosArrayGet(pReq->pTableName, i);
      STableMeta *pTableMeta = NULL;
      
D
dapan1121 已提交
2573
      CTG_ERR_JRET(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, name, &pTableMeta, CTG_FLAG_UNKNOWN_STB));
D
dapan1121 已提交
2574 2575 2576 2577 2578 2579 2580 2581 2582

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

D
dapan1121 已提交
2583
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2584 2585

_return:  
D
dapan1121 已提交
2586

D
dapan1121 已提交
2587 2588 2589 2590 2591 2592 2593 2594
  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 已提交
2595
    pRsp->pTableMeta = NULL;
D
dapan1121 已提交
2596
  }
D
dapan 已提交
2597
  
D
dapan1121 已提交
2598
  CTG_API_LEAVE(code);
2599
}
D
dapan 已提交
2600

D
dapan1121 已提交
2601
int32_t catalogGetQnodeList(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SArray* pQnodeList) {
D
dapan1121 已提交
2602 2603
  CTG_API_ENTER();

D
dapan1121 已提交
2604 2605 2606 2607
  if (NULL == pCtg || NULL == pRpc  || NULL == pMgmtEps || NULL == pQnodeList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2608
  //TODO
D
dapan 已提交
2609

D
dapan1121 已提交
2610
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2611 2612
}

D
dapan1121 已提交
2613
int32_t catalogGetExpiredSTables(SCatalog* pCtg, SSTableMetaVersion **stables, uint32_t *num) {
D
dapan1121 已提交
2614 2615
  CTG_API_ENTER();

D
dapan1121 已提交
2616 2617
  if (NULL == pCtg || NULL == stables || NULL == num) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2618 2619
  }

D
dapan1121 已提交
2620 2621 2622 2623
  CTG_API_LEAVE(ctgMetaRentGet(&pCtg->stbRent, (void **)stables, num, sizeof(SSTableMetaVersion)));
}

int32_t catalogGetExpiredDBs(SCatalog* pCtg, SDbVgVersion **dbs, uint32_t *num) {
D
dapan1121 已提交
2624
  CTG_API_ENTER();
D
dapan1121 已提交
2625 2626 2627 2628
  
  if (NULL == pCtg || NULL == dbs || NULL == num) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
2629

D
dapan1121 已提交
2630
  CTG_API_LEAVE(ctgMetaRentGet(&pCtg->dbRent, (void **)dbs, num, sizeof(SDbVgVersion)));
D
dapan1121 已提交
2631 2632
}

D
dapan 已提交
2633

D
dapan 已提交
2634
void catalogDestroy(void) {
D
dapan1121 已提交
2635 2636
  qInfo("start to destroy catalog");
  
D
dapan 已提交
2637
  if (NULL == gCtgMgmt.pCluster || atomic_load_8(&gCtgMgmt.exit)) {
D
dapan1121 已提交
2638 2639 2640
    return;
  }

D
dapan 已提交
2641 2642 2643
  atomic_store_8(&gCtgMgmt.exit, true);

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

D
dapan1121 已提交
2645 2646 2647 2648
  while (CTG_IS_LOCKED(&gCtgMgmt.lock)) {
    usleep(1);
  }
  
D
dapan 已提交
2649
  CTG_LOCK(CTG_WRITE, &gCtgMgmt.lock);
D
dapan1121 已提交
2650

D
dapan1121 已提交
2651
  SCatalog *pCtg = NULL;
D
dapan 已提交
2652
  void *pIter = taosHashIterate(gCtgMgmt.pCluster, NULL);
D
dapan1121 已提交
2653
  while (pIter) {
D
dapan1121 已提交
2654
    pCtg = *(SCatalog **)pIter;
D
dapan1121 已提交
2655

D
dapan1121 已提交
2656 2657
    if (pCtg) {
      catalogFreeHandle(pCtg);
D
dapan1121 已提交
2658 2659
    }
    
D
dapan 已提交
2660
    pIter = taosHashIterate(gCtgMgmt.pCluster, pIter);
D
dapan 已提交
2661
  }
D
dapan1121 已提交
2662
  
D
dapan 已提交
2663 2664
  taosHashCleanup(gCtgMgmt.pCluster);
  gCtgMgmt.pCluster = NULL;
D
dapan1121 已提交
2665

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

D
dapan1121 已提交
2668
  qInfo("catalog destroyed");
D
dapan 已提交
2669 2670 2671 2672
}