catalog.c 75.9 KB
Newer Older
H
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
H
Haojun Liao 已提交
14 15
 */

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

D
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 ctgStbVersionSearchCompare(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
    return 1;
  } else {
    return 0;
  }
}

D
dapan1121 已提交
978
int32_t ctgDbVgVersionSearchCompare(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 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008
int32_t ctgStbVersionSortCompare(const void* key1, const void* key2) {
  if (((SSTableMetaVersion*)key1)->suid < ((SSTableMetaVersion*)key2)->suid) {
    return -1;
  } else if (((SSTableMetaVersion*)key1)->suid > ((SSTableMetaVersion*)key2)->suid) {
    return 1;
  } else {
    return 0;
  }
}

int32_t ctgDbVgVersionSortCompare(const void* key1, const void* key2) {
  if (((SDbVgVersion*)key1)->dbId < ((SDbVgVersion*)key2)->dbId) {
    return -1;
  } else if (((SDbVgVersion*)key1)->dbId > ((SDbVgVersion*)key2)->dbId) {
    return 1;
  } else {
    return 0;
  }
}


D
dapan1121 已提交
1009
int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type) {
D
dapan1121 已提交
1010 1011 1012 1013
  mgmt->slotRIdx = 0;
  mgmt->slotNum = rentSec / CTG_RENT_SLOT_SECOND;
  mgmt->type = type;

D
dapan1121 已提交
1014
  size_t msgSize = sizeof(SCtgRentSlot) * mgmt->slotNum;
D
dapan1121 已提交
1015
  
D
dapan1121 已提交
1016 1017 1018
  mgmt->slots = calloc(1, msgSize);
  if (NULL == mgmt->slots) {
    qError("calloc %d failed", (int32_t)msgSize);
D
dapan 已提交
1019
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1020
  }
D
dapan1121 已提交
1021

D
dapan1121 已提交
1022 1023 1024 1025
  qDebug("meta rent initialized, type:%d, slotNum:%d", type, mgmt->slotNum);
  
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
1026

D
dapan1121 已提交
1027

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

D
dapan1121 已提交
1031
  SCtgRentSlot *slot = &mgmt->slots[widx];
D
dapan1121 已提交
1032 1033 1034 1035 1036 1037 1038 1039
  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 已提交
1040
    }
D
dapan1121 已提交
1041
  }
D
dapan1121 已提交
1042

D
dapan1121 已提交
1043 1044 1045
  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 已提交
1046 1047
  }

D
dapan1121 已提交
1048
  slot->needSort = true;
D
dapan1121 已提交
1049

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

D
dapan1121 已提交
1052 1053 1054 1055 1056 1057
_return:

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

D
dapan1121 已提交
1058
int32_t ctgMetaRentUpdate(SCtgRentMgmt *mgmt, void *meta, int64_t id, int32_t size, __compar_fn_t sortCompare, __compar_fn_t searchCompare) {
D
dapan1121 已提交
1059
  int16_t widx = abs(id % mgmt->slotNum);
D
dapan1121 已提交
1060

D
dapan1121 已提交
1061
  SCtgRentSlot *slot = &mgmt->slots[widx];
D
dapan1121 已提交
1062 1063 1064 1065
  int32_t code = 0;
  
  CTG_LOCK(CTG_WRITE, &slot->lock);
  if (NULL == slot->meta) {
D
dapan1121 已提交
1066 1067
    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 已提交
1068 1069 1070
  }

  if (slot->needSort) {
D
dapan1121 已提交
1071
    qDebug("meta slot before sorte, slot idx:%d, type:%d, size:%d", widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta));
D
dapan1121 已提交
1072
    taosArraySort(slot->meta, sortCompare);
D
dapan1121 已提交
1073
    slot->needSort = false;
D
dapan1121 已提交
1074
    qDebug("meta slot sorted, slot idx:%d, type:%d, size:%d", widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta));
D
dapan1121 已提交
1075 1076
  }

D
dapan1121 已提交
1077
  void *orig = taosArraySearch(slot->meta, &id, searchCompare, TD_EQ);
D
dapan1121 已提交
1078
  if (NULL == orig) {
D
dapan1121 已提交
1079
    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 已提交
1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098
    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 已提交
1099
int32_t ctgMetaRentRemove(SCtgRentMgmt *mgmt, int64_t id, __compar_fn_t sortCompare, __compar_fn_t searchCompare) {
D
dapan1121 已提交
1100 1101
  int16_t widx = abs(id % mgmt->slotNum);

D
dapan1121 已提交
1102
  SCtgRentSlot *slot = &mgmt->slots[widx];
D
dapan1121 已提交
1103 1104 1105 1106 1107 1108 1109 1110 1111
  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) {
D
dapan1121 已提交
1112
    taosArraySort(slot->meta, sortCompare);
D
dapan1121 已提交
1113 1114 1115 1116
    slot->needSort = false;
    qDebug("meta slot sorted, slot idx:%d, type:%d", widx, mgmt->type);
  }

D
dapan1121 已提交
1117
  int32_t idx = taosArraySearchIdx(slot->meta, &id, searchCompare, TD_EQ);
D
dapan1121 已提交
1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134
  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 已提交
1135
int32_t ctgMetaRentGetImpl(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size) {
D
dapan1121 已提交
1136 1137 1138 1139
  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 已提交
1140
  }
D
dapan1121 已提交
1141

D
dapan1121 已提交
1142
  SCtgRentSlot *slot = &mgmt->slots[ridx];
D
dapan1121 已提交
1143
  int32_t code = 0;
D
dapan1121 已提交
1144
  
D
dapan1121 已提交
1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180
  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 已提交
1181
int32_t ctgMetaRentGet(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size) {
D
dapan1121 已提交
1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200
  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 已提交
1201 1202 1203
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
1204
int32_t ctgAddNewDBCache(SCatalog *pCtg, const char *dbFName, uint64_t dbId) {
D
dapan1121 已提交
1205
  int32_t code = 0;
D
dapan1121 已提交
1206

D
dapan1121 已提交
1207 1208 1209
  SCtgDBCache newDBCache = {0};
  newDBCache.dbId = dbId;

D
dapan 已提交
1210
  newDBCache.tbCache.metaCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
D
dapan1121 已提交
1211
  if (NULL == newDBCache.tbCache.metaCache) {
D
dapan 已提交
1212
    ctgError("taosHashInit %d metaCache failed", gCtgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
1213 1214 1215
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

D
dapan 已提交
1216
  newDBCache.tbCache.stbCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), true, HASH_ENTRY_LOCK);
D
dapan1121 已提交
1217
  if (NULL == newDBCache.tbCache.stbCache) {
D
dapan 已提交
1218
    ctgError("taosHashInit %d stbCache failed", gCtgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
1219 1220 1221 1222
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
  }

  code = taosHashPut(pCtg->dbCache, dbFName, strlen(dbFName), &newDBCache, sizeof(SCtgDBCache));
D
dapan1121 已提交
1223 1224 1225
  if (code) {
    if (HASH_NODE_EXIST(code)) {
      ctgDebug("db already in cache, dbFName:%s", dbFName);
D
dapan1121 已提交
1226
      goto _return;
D
dapan1121 已提交
1227 1228 1229
    }
    
    ctgError("taosHashPut db to cache failed, dbFName:%s", dbFName);
D
dapan1121 已提交
1230 1231 1232
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
  }
  
D
dapan1121 已提交
1233
  SDbVgVersion vgVersion = {.dbId = newDBCache.dbId, .vgVersion = -1};
D
dapan1121 已提交
1234 1235
  strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));

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

D
dapan1121 已提交
1238 1239 1240 1241
  if (CTG_IS_INF_DBNAME(dbFName)) {
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
1242 1243 1244
  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 已提交
1245

D
dapan1121 已提交
1246
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1247

D
dapan1121 已提交
1248
_return:
D
dapan1121 已提交
1249

D
dapan1121 已提交
1250
  ctgFreeDbCache(&newDBCache);
D
dapan1121 已提交
1251

D
dapan1121 已提交
1252 1253
  CTG_RET(code);
}
D
dapan1121 已提交
1254

D
dapan1121 已提交
1255

D
dapan1121 已提交
1256
void ctgRemoveStbRent(SCatalog* pCtg, SCtgTbMetaCache *cache) {
D
dapan1121 已提交
1257 1258 1259 1260 1261 1262 1263
  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 已提交
1264
      if (TSDB_CODE_SUCCESS == ctgMetaRentRemove(&pCtg->stbRent, *suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare)) {
D
dapan1121 已提交
1265 1266 1267 1268 1269 1270 1271 1272 1273 1274
        ctgDebug("stb removed from rent, suid:%"PRIx64, *suid);
      }
          
      pIter = taosHashIterate(cache->stbCache, pIter);
    }
  }
  CTG_UNLOCK(CTG_WRITE, &cache->stbLock);
}


D
dapan1121 已提交
1275
int32_t ctgRemoveDB(SCatalog* pCtg, SCtgDBCache *dbCache, const char* dbFName) {
D
dapan 已提交
1276 1277 1278 1279
  uint64_t dbId = dbCache->dbId;
  
  ctgInfo("start to remove db from cache, dbFName:%s, dbId:%"PRIx64, dbFName, dbCache->dbId);

D
dapan1121 已提交
1280 1281 1282 1283 1284 1285 1286 1287
  atomic_store_8(&dbCache->deleted, 1);

  ctgRemoveStbRent(pCtg, &dbCache->tbCache);

  ctgFreeDbCache(dbCache);

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

D
dapan1121 已提交
1288
  CTG_ERR_RET(ctgMetaRentRemove(&pCtg->dbRent, dbCache->dbId, ctgDbVgVersionSortCompare, ctgDbVgVersionSearchCompare));
D
dapan1121 已提交
1289 1290 1291 1292
  
  ctgDebug("db removed from rent, dbFName:%s, dbId:%"PRIx64, dbFName, dbCache->dbId);

  if (taosHashRemove(pCtg->dbCache, dbFName, strlen(dbFName))) {
D
dapan1121 已提交
1293 1294
    ctgInfo("taosHashRemove from dbCache failed, may be removed, dbFName:%s", dbFName);
    CTG_ERR_RET(TSDB_CODE_CTG_DB_DROPPED);
D
dapan1121 已提交
1295
  }
D
dapan 已提交
1296 1297

  ctgInfo("db removed from cache, dbFName:%s, dbId:%"PRIx64, dbFName, dbId);
D
dapan1121 已提交
1298 1299 1300
  
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
1301 1302


D
dapan1121 已提交
1303
int32_t ctgGetAddDBCache(SCatalog* pCtg, const char *dbFName, uint64_t dbId, SCtgDBCache **pCache) {
D
dapan1121 已提交
1304 1305
  int32_t code = 0;
  SCtgDBCache *dbCache = NULL;
D
dapan1121 已提交
1306
  ctgGetDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
1307
  
D
dapan1121 已提交
1308 1309
  if (dbCache) {
  // TODO OPEN IT
D
dapan1121 已提交
1310
#if 0    
D
dapan1121 已提交
1311 1312 1313 1314
    if (dbCache->dbId == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
D
dapan1121 已提交
1315
#else
D
dapan1121 已提交
1316 1317 1318
    if (0 == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1319 1320
    }

D
dapan1121 已提交
1321 1322 1323 1324 1325
    if (dbId && (dbCache->dbId == 0)) {
      dbCache->dbId = dbId;
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
D
dapan1121 已提交
1326
    
D
dapan1121 已提交
1327 1328 1329 1330 1331 1332
    if (dbCache->dbId == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
#endif
    CTG_ERR_RET(ctgRemoveDB(pCtg, dbCache, dbFName));
D
dapan1121 已提交
1333
  }
D
dapan1121 已提交
1334 1335
  
  CTG_ERR_RET(ctgAddNewDBCache(pCtg, dbFName, dbId));
D
dapan1121 已提交
1336

D
dapan1121 已提交
1337
  ctgGetDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
1338

D
dapan1121 已提交
1339
  *pCache = dbCache;
D
dapan1121 已提交
1340

D
dapan1121 已提交
1341
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1342 1343 1344
}


D
dapan1121 已提交
1345 1346 1347 1348 1349 1350
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 已提交
1351
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1352 1353
  }

D
dapan1121 已提交
1354
  bool newAdded = false;
D
dapan 已提交
1355
  SDbVgVersion vgVersion = {.dbId = dbId, .vgVersion = dbInfo->vgVersion, .numOfTable = dbInfo->numOfTable};
D
dapan1121 已提交
1356 1357 1358 1359 1360 1361 1362 1363 1364 1365

  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 已提交
1366
  
D
dapan1121 已提交
1367
  if (dbCache->vgInfo) {
D
dapan 已提交
1368 1369 1370 1371 1372 1373 1374 1375 1376
    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 已提交
1377
      ctgWReleaseVgInfo(dbCache);
D
dapan1121 已提交
1378
      
D
dapan1121 已提交
1379
      return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1380
    }
D
dapan1121 已提交
1381 1382

    ctgFreeVgInfo(dbCache->vgInfo);
D
dapan1121 已提交
1383 1384
  }

D
dapan1121 已提交
1385
  dbCache->vgInfo = dbInfo;
D
dapan1121 已提交
1386

D
dapan1121 已提交
1387
  *pDbInfo = NULL;
D
dapan1121 已提交
1388

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

D
dapan1121 已提交
1391
  ctgWReleaseVgInfo(dbCache);
D
dapan1121 已提交
1392

D
dapan1121 已提交
1393
  dbCache = NULL;
D
dapan1121 已提交
1394

D
dapan1121 已提交
1395
  strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));
D
dapan1121 已提交
1396
  CTG_ERR_RET(ctgMetaRentUpdate(&pCtg->dbRent, &vgVersion, vgVersion.dbId, sizeof(SDbVgVersion), ctgDbVgVersionSortCompare, ctgDbVgVersionSearchCompare));
D
dapan1121 已提交
1397
  
D
dapan1121 已提交
1398 1399 1400 1401
  CTG_RET(code);
}


D
dapan1121 已提交
1402 1403
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 已提交
1404

D
dapan1121 已提交
1405 1406 1407 1408 1409
  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 已提交
1410 1411
  }

D
dapan1121 已提交
1412 1413 1414 1415 1416 1417 1418
  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 已提交
1419 1420 1421 1422 1423 1424 1425 1426 1427 1428
    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);
        
D
dapan1121 已提交
1429
        ctgMetaRentRemove(&pCtg->stbRent, orig->suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare);
D
dapan1121 已提交
1430
      }
D
dapan1121 已提交
1431

D
dapan1121 已提交
1432
      origSuid = orig->suid;
D
dapan1121 已提交
1433
    }
D
dapan1121 已提交
1434
  }
D
dapan1121 已提交
1435

D
dapan1121 已提交
1436 1437
  if (isStb) {
    CTG_LOCK(CTG_WRITE, &tbCache->stbLock);
D
dapan1121 已提交
1438
  }
D
dapan1121 已提交
1439
  
D
dapan1121 已提交
1440 1441 1442 1443 1444 1445 1446 1447 1448
  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 已提交
1449

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

D
dapan1121 已提交
1453 1454 1455
  if (!isStb) {
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);  
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1456
  }
D
dapan1121 已提交
1457

D
dapan1121 已提交
1458
  if (origType == TSDB_SUPER_TABLE && origSuid == meta->suid) {
D
dapan1121 已提交
1459 1460 1461 1462
    CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);  
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
1463

D
dapan1121 已提交
1464 1465 1466 1467 1468 1469
  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 已提交
1470
  }
D
dapan1121 已提交
1471 1472
  
  CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);
D
dapan1121 已提交
1473

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

D
dapan 已提交
1476
  ctgDebug("stb updated to stbCache, dbFName:%s, tbName:%s, tbType:%d", dbFName, tbName, meta->tableType);
D
dapan1121 已提交
1477 1478 1479 1480 1481 1482 1483

  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 已提交
1484 1485
}

D
dapan 已提交
1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524
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 已提交
1525
int32_t ctgGetDBVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* dbFName, bool forceUpdate, SCtgDBCache** dbCache, SDBVgInfo **pInfo) {
D
dapan1121 已提交
1526
  bool inCache = false;
D
dapan1121 已提交
1527
  int32_t code = 0;
D
dapan1121 已提交
1528 1529 1530 1531 1532

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

  if (inCache && !forceUpdate) {
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1533 1534 1535 1536 1537
  }

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

D
dapan1121 已提交
1538
  tstrncpy(input.db, dbFName, tListLen(input.db));
D
dapan1121 已提交
1539 1540 1541
  if (inCache) {
    input.dbId = (*dbCache)->dbId;
    input.vgVersion = (*dbCache)->vgInfo->vgVersion;
D
dapan 已提交
1542
    input.numOfTable = (*dbCache)->vgInfo->numOfTable;
D
dapan1121 已提交
1543 1544 1545
  } else {
    input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
  }
H
Haojun Liao 已提交
1546

D
dapan1121 已提交
1547 1548 1549 1550 1551 1552 1553 1554 1555
  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 已提交
1556

D
dapan 已提交
1557 1558 1559 1560 1561 1562 1563 1564
  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 已提交
1565 1566
  }

D
dapan 已提交
1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577
  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 已提交
1578
  return TSDB_CODE_SUCCESS;
D
dapan 已提交
1579 1580 1581 1582 1583 1584 1585 1586 1587

_return:

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

D
dapan 已提交
1590

D
dapan1121 已提交
1591 1592 1593 1594 1595
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 已提交
1596 1597
  }

D
dapan1121 已提交
1598 1599 1600 1601 1602 1603
  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 已提交
1604
      qError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
D
dapan1121 已提交
1605 1606 1607 1608 1609
      tfree(*pOutput);
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }

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

D
dapan1121 已提交
1612 1613 1614
  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
1615 1616


D
dapan1121 已提交
1617 1618
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 已提交
1619 1620 1621 1622 1623 1624
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }

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

D
dapan1121 已提交
1625
  if (!CTG_FLAG_IS_INF_DB(flag)) {
D
dapan1121 已提交
1626 1627
    CTG_ERR_RET(catalogGetTableHashVgroup(pCtg, pTrans, pMgmtEps, pTableName, &vgroupInfo));
  }
D
dapan1121 已提交
1628

D
dapan 已提交
1629
  SCtgUpdateTblMsg *msg = NULL;
D
dapan1121 已提交
1630
  STableMetaOutput  moutput = {0};
S
Shengliang Guan 已提交
1631
  STableMetaOutput *output = calloc(1, sizeof(STableMetaOutput));
D
dapan1121 已提交
1632 1633 1634 1635
  if (NULL == output) {
    ctgError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }
D
dapan1121 已提交
1636

D
dapan1121 已提交
1637
  if (CTG_FLAG_IS_INF_DB(flag)) {
D
dapan1121 已提交
1638 1639 1640
    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 已提交
1641
  } else if (CTG_FLAG_IS_STB(flag)) {
D
dapan 已提交
1642
    ctgDebug("will refresh tbmeta, supposed to be stb, tbName:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
1643 1644

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

D
dapan1121 已提交
1647
    if (CTG_IS_META_NULL(output->metaType)) {
D
dapan1121 已提交
1648
      CTG_ERR_JRET(ctgGetTableMetaFromVnode(pCtg, pTrans, pMgmtEps, pTableName, &vgroupInfo, output));
D
dapan1121 已提交
1649 1650
    }
  } else {
D
dapan1121 已提交
1651
    ctgDebug("will refresh tbmeta, not supposed to be stb, tbName:%s, flag:%d", tNameGetTableName(pTableName), flag);
D
dapan1121 已提交
1652 1653

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

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

D
dapan1121 已提交
1659
      tfree(output->tbMeta);
D
dapan1121 已提交
1660
      
D
dapan1121 已提交
1661
      CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, output->dbFName, output->tbName, output));
D
dapan1121 已提交
1662
    } else if (CTG_IS_META_BOTH(output->metaType)) {
D
dapan1121 已提交
1663
      int32_t exist = 0;
D
dapan1121 已提交
1664 1665 1666 1667
      if (!CTG_FLAG_IS_FORCE_UPDATE(flag)) {
        CTG_ERR_JRET(ctgIsTableMetaExistInCache(pCtg, output->dbFName, output->tbName, &exist));
      }
      
D
dapan1121 已提交
1668
      if (0 == exist) {
D
dapan1121 已提交
1669
        CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, output->dbFName, output->tbName, &moutput));
D
dapan1121 已提交
1670

D
dapan1121 已提交
1671
        if (CTG_IS_META_NULL(moutput.metaType)) {
D
dapan1121 已提交
1672
          SET_META_TYPE_NULL(output->metaType);
D
dapan1121 已提交
1673 1674
        }
        
D
dapan1121 已提交
1675 1676
        tfree(output->tbMeta);
        output->tbMeta = moutput.tbMeta;
D
dapan1121 已提交
1677 1678
        moutput.tbMeta = NULL;
      } else {
D
dapan1121 已提交
1679
        tfree(output->tbMeta);
D
dapan1121 已提交
1680
        
D
dapan1121 已提交
1681
        SET_META_TYPE_CTABLE(output->metaType); 
D
dapan1121 已提交
1682
      }
D
dapan1121 已提交
1683 1684 1685
    }
  }

D
dapan1121 已提交
1686
  if (CTG_IS_META_NULL(output->metaType)) {
D
dapan 已提交
1687
    ctgError("no tbmeta got, tbNmae:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
1688 1689 1690
    CTG_ERR_JRET(CTG_ERR_CODE_TABLE_NOT_EXIST);
  }

D
dapan 已提交
1691 1692 1693 1694 1695 1696
  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 已提交
1697 1698 1699 1700 1701
  if (pOutput) {
    CTG_ERR_JRET(ctgCloneMetaOutput(output, pOutput));
  }

  SCtgMetaAction action= {.act = CTG_ACT_UPDATE_TBL};
D
dapan 已提交
1702
  msg = malloc(sizeof(SCtgUpdateTblMsg));
D
dapan1121 已提交
1703 1704 1705 1706 1707 1708 1709 1710 1711 1712 1713 1714
  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 已提交
1715 1716
  ctgDebug("action [%s] added into queue", gCtgAction[action.act].name);

D
dapan1121 已提交
1717
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1718 1719 1720

_return:

D
dapan1121 已提交
1721
  tfree(output->tbMeta);
D
dapan 已提交
1722
  tfree(output);
D
dapan1121 已提交
1723
  tfree(msg);
D
dapan1121 已提交
1724 1725 1726 1727
  
  CTG_RET(code);
}

D
dapan1121 已提交
1728
int32_t ctgGetTableMeta(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t flag) {
D
dapan1121 已提交
1729
  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == pTableMeta) {
D
dapan1121 已提交
1730 1731 1732 1733
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }
  
  int32_t exist = 0;
D
dapan1121 已提交
1734
  int32_t code = 0;
D
dapan1121 已提交
1735 1736 1737
  uint64_t dbId = 0;
  uint64_t suid = 0;
  STableMetaOutput *output = NULL;
D
dapan1121 已提交
1738

D
dapan1121 已提交
1739
  if (CTG_IS_INF_DBNAME(pTableName->dbname)) {
D
dapan1121 已提交
1740
    CTG_FLAG_SET_INF_DB(flag);
D
dapan1121 已提交
1741 1742
  }

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

D
dapan1121 已提交
1745 1746 1747 1748 1749
  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 已提交
1750
    }
D
dapan1121 已提交
1751

D
dapan1121 已提交
1752 1753
    tbType = (*pTableMeta)->tableType;
    suid = (*pTableMeta)->suid;
D
dapan1121 已提交
1754

D
dapan1121 已提交
1755 1756 1757 1758 1759
    tfree(*pTableMeta);    
  }
    
  if (CTG_FLAG_IS_UNKNOWN_STB(flag)) {
    CTG_FLAG_SET_STB(flag, tbType);
D
dapan1121 已提交
1760 1761
  }

D
dapan1121 已提交
1762

D
dapan 已提交
1763
  while (true) {
D
dapan1121 已提交
1764
    CTG_ERR_JRET(ctgRefreshTblMeta(pCtg, pRpc, pMgmtEps, pTableName, flag, &output));
D
dapan1121 已提交
1765

D
dapan 已提交
1766 1767 1768 1769
    if (CTG_IS_META_TABLE(output->metaType)) {
      *pTableMeta = output->tbMeta;
      goto _return;
    }
D
dapan1121 已提交
1770

D
dapan 已提交
1771 1772 1773 1774 1775 1776
    if (CTG_IS_META_BOTH(output->metaType)) {
      memcpy(output->tbMeta, &output->ctbMeta, sizeof(output->ctbMeta));
      
      *pTableMeta = output->tbMeta;
      goto _return;
    }
D
dapan1121 已提交
1777

D
dapan 已提交
1778 1779 1780 1781 1782
    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 已提交
1783

D
dapan 已提交
1784
    // HANDLE ONLY CHILD TABLE META
D
dapan1121 已提交
1785

D
dapan 已提交
1786 1787 1788
    SName stbName = *pTableName;
    strcpy(stbName.tname, output->tbName);
    
D
dapan1121 已提交
1789
    CTG_ERR_JRET(ctgGetTableMetaFromCache(pCtg, &stbName, pTableMeta, &exist, flag, NULL));
D
dapan 已提交
1790 1791 1792 1793
    if (0 == exist) {
      ctgDebug("stb no longer exist, dbFName:%s, tbName:%s", output->dbFName, pTableName->tname);
      continue;
    }
D
dapan1121 已提交
1794

D
dapan 已提交
1795 1796 1797 1798
    memcpy(*pTableMeta, &output->ctbMeta, sizeof(output->ctbMeta));

    break;
  }
D
dapan1121 已提交
1799 1800 1801

_return:

D
dapan1121 已提交
1802 1803 1804 1805 1806 1807 1808 1809 1810 1811 1812 1813 1814 1815 1816
  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 已提交
1817 1818
  tfree(output);

D
dapan 已提交
1819 1820
  if (*pTableMeta) {
    ctgDebug("tbmeta returned, tbName:%s, tbType:%d", pTableName->tname, (*pTableMeta)->tableType);
D
dapan1121 已提交
1821
    ctgDbgShowTableMeta(pCtg, pTableName->tname, *pTableMeta);
D
dapan 已提交
1822 1823
  }

D
dapan1121 已提交
1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836
  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 已提交
1837
  ctgFreeVgInfo(msg->dbInfo);
D
dapan1121 已提交
1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848
  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 已提交
1849
  ctgGetDBCache(msg->pCtg, msg->dbFName, &dbCache);
D
dapan1121 已提交
1850 1851 1852 1853 1854 1855 1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877
  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 已提交
1878
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
1879 1880 1881 1882 1883 1884 1885
  }

  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 已提交
1886 1887 1888 1889 1890
  char *p = strchr(output->dbFName, '.');
  if (p && CTG_IS_INF_DBNAME(p + 1)) {
    memmove(output->dbFName, p + 1, strlen(p + 1));
  }
  
D
dapan1121 已提交
1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908
  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 已提交
1909 1910 1911
  if (output) {
    tfree(output->tbMeta);
    tfree(output);
D
dapan1121 已提交
1912
  }
D
dapan 已提交
1913
  
D
dapan1121 已提交
1914 1915 1916 1917 1918 1919 1920 1921 1922 1923 1924 1925 1926 1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947
  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 已提交
1948
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
1949 1950 1951 1952 1953 1954 1955
  }  
  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 已提交
1956
  CTG_ERR_JRET(ctgMetaRentRemove(&msg->pCtg->stbRent, msg->suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare));
D
dapan1121 已提交
1957 1958 1959 1960 1961 1962 1963 1964 1965 1966 1967
  
  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 已提交
1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979 1980 1981 1982 1983 1984 1985 1986 1987 1988 1989 1990 1991 1992 1993
  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 已提交
1994

D
dapan1121 已提交
1995 1996 1997
  tfree(msg);
  
  CTG_RET(code);
D
dapan1121 已提交
1998 1999 2000 2001 2002 2003 2004 2005
}


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

  qInfo("catalog update thread started");

D
dapan 已提交
2006
  CTG_LOCK(CTG_READ, &gCtgMgmt.lock);
D
dapan1121 已提交
2007 2008
  
  while (true) {
D
dapan 已提交
2009
    tsem_wait(&gCtgMgmt.sem);
D
dapan1121 已提交
2010
    
D
dapan 已提交
2011
    if (atomic_load_8(&gCtgMgmt.exit)) {
D
dapan1121 已提交
2012 2013 2014 2015 2016
      break;
    }

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

D
dapan1121 已提交
2019
    ctgDebug("process [%s] action", gCtgAction[action->act].name);
D
dapan 已提交
2020 2021 2022 2023
    
    (*gCtgAction[action->act].func)(action);

    CTG_STAT_ADD(gCtgMgmt.stat.runtime.qDoneNum); 
D
dapan1121 已提交
2024 2025

    ctgDbgShowClusterCache(pCtg);
D
dapan1121 已提交
2026 2027
  }

D
dapan 已提交
2028
  CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock);
D
dapan1121 已提交
2029 2030 2031 2032 2033 2034 2035 2036 2037 2038 2039 2040

  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 已提交
2041
  if (pthread_create(&gCtgMgmt.updateThread, &thAttr, ctgUpdateThreadFunc, NULL) != 0) {
D
dapan1121 已提交
2042 2043
    terrno = TAOS_SYSTEM_ERROR(errno);
    CTG_ERR_RET(terrno);
D
dapan1121 已提交
2044 2045
  }
  
D
dapan1121 已提交
2046
  pthread_attr_destroy(&thAttr);
D
dapan1121 已提交
2047 2048 2049 2050
  return TSDB_CODE_SUCCESS;
}


D
dapan1121 已提交
2051
int32_t catalogInit(SCatalogCfg *cfg) {
D
dapan 已提交
2052
  if (gCtgMgmt.pCluster) {
D
dapan 已提交
2053
    qError("catalog already initialized");
D
dapan1121 已提交
2054
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2055 2056
  }

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

D
dapan1121 已提交
2059
  if (cfg) {
D
dapan 已提交
2060
    memcpy(&gCtgMgmt.cfg, cfg, sizeof(*cfg));
H
Haojun Liao 已提交
2061

D
dapan 已提交
2062 2063
    if (gCtgMgmt.cfg.maxDBCacheNum == 0) {
      gCtgMgmt.cfg.maxDBCacheNum = CTG_DEFAULT_CACHE_DB_NUMBER;
D
dapan1121 已提交
2064 2065
    }

D
dapan 已提交
2066 2067
    if (gCtgMgmt.cfg.maxTblCacheNum == 0) {
      gCtgMgmt.cfg.maxTblCacheNum = CTG_DEFAULT_CACHE_TBLMETA_NUMBER;
D
dapan1121 已提交
2068
    }
D
dapan1121 已提交
2069

D
dapan 已提交
2070 2071
    if (gCtgMgmt.cfg.dbRentSec == 0) {
      gCtgMgmt.cfg.dbRentSec = CTG_DEFAULT_RENT_SECOND;
D
dapan1121 已提交
2072 2073
    }

D
dapan 已提交
2074 2075
    if (gCtgMgmt.cfg.stbRentSec == 0) {
      gCtgMgmt.cfg.stbRentSec = CTG_DEFAULT_RENT_SECOND;
D
dapan1121 已提交
2076
    }
D
dapan1121 已提交
2077
  } else {
D
dapan 已提交
2078 2079 2080 2081
    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 已提交
2082 2083
  }

D
dapan 已提交
2084 2085
  gCtgMgmt.pCluster = taosHashInit(CTG_DEFAULT_CACHE_CLUSTER_NUMBER, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
  if (NULL == gCtgMgmt.pCluster) {
D
dapan1121 已提交
2086 2087
    qError("taosHashInit %d cluster cache failed", CTG_DEFAULT_CACHE_CLUSTER_NUMBER);
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
2088 2089
  }

D
dapan1121 已提交
2090 2091
  CTG_ERR_RET(ctgStartUpdateThread());

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

D
dapan 已提交
2094 2095
  gCtgMgmt.head = calloc(1, sizeof(SCtgQNode));
  if (NULL == gCtgMgmt.head) {
D
dapan1121 已提交
2096 2097 2098
    qError("calloc %d failed", (int32_t)sizeof(SCtgQNode));
    CTG_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }
D
dapan 已提交
2099
  gCtgMgmt.tail = gCtgMgmt.head;
D
dapan1121 已提交
2100

D
dapan 已提交
2101
  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 已提交
2102

D
dapan 已提交
2103
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
2104 2105
}

D
dapan1121 已提交
2106
int32_t catalogGetHandle(uint64_t clusterId, SCatalog** catalogHandle) {
2107
  if (NULL == catalogHandle) {
D
dapan1121 已提交
2108
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
2109 2110
  }

D
dapan 已提交
2111
  if (NULL == gCtgMgmt.pCluster) {
D
dapan 已提交
2112
    qError("catalog cluster cache are not ready, clusterId:%"PRIx64, clusterId);
D
dapan1121 已提交
2113
    CTG_ERR_RET(TSDB_CODE_CTG_NOT_READY);
D
dapan 已提交
2114 2115
  }

D
dapan1121 已提交
2116 2117
  int32_t code = 0;
  SCatalog *clusterCtg = NULL;
D
dapan 已提交
2118

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

D
dapan1121 已提交
2122 2123 2124 2125 2126
    if (ctg && (*ctg)) {
      *catalogHandle = *ctg;
      qDebug("got catalog handle from cache, clusterId:%"PRIx64", CTG:%p", clusterId, *ctg);
      return TSDB_CODE_SUCCESS;
    }
D
dapan 已提交
2127

D
dapan1121 已提交
2128 2129 2130 2131 2132 2133
    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 已提交
2134 2135
    clusterCtg->clusterId = clusterId;

D
dapan 已提交
2136 2137
    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 已提交
2138

D
dapan 已提交
2139
    clusterCtg->dbCache = taosHashInit(gCtgMgmt.cfg.maxDBCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
D
dapan1121 已提交
2140 2141 2142 2143 2144
    if (NULL == clusterCtg->dbCache) {
      qError("taosHashInit %d dbCache failed", CTG_DEFAULT_CACHE_DB_NUMBER);
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
    }

D
dapan 已提交
2145
    SHashObj *metaCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
D
dapan1121 已提交
2146
    if (NULL == metaCache) {
D
dapan 已提交
2147
      qError("taosHashInit failed, num:%d", gCtgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
2148 2149 2150
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }
    
D
dapan 已提交
2151
    code = taosHashPut(gCtgMgmt.pCluster, &clusterId, sizeof(clusterId), &clusterCtg, POINTER_BYTES);
D
dapan1121 已提交
2152 2153 2154 2155 2156 2157 2158 2159 2160 2161 2162 2163 2164
    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 已提交
2165
  }
D
dapan1121 已提交
2166 2167

  *catalogHandle = clusterCtg;
D
dapan 已提交
2168
  
D
dapan1121 已提交
2169
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
2170 2171 2172 2173 2174 2175 2176 2177

_return:

  ctgFreeHandle(clusterCtg);
  
  CTG_RET(code);
}

D
dapan1121 已提交
2178 2179
void catalogFreeHandle(SCatalog* pCtg) {
  if (NULL == pCtg) {
D
dapan1121 已提交
2180 2181
    return;
  }
D
dapan1121 已提交
2182

D
dapan 已提交
2183
  if (taosHashRemove(gCtgMgmt.pCluster, &pCtg->clusterId, sizeof(pCtg->clusterId))) {
D
dapan1121 已提交
2184
    ctgWarn("taosHashRemove from cluster failed, may already be freed, clusterId:%"PRIx64, pCtg->clusterId);
D
dapan1121 已提交
2185 2186 2187
    return;
  }

D
dapan1121 已提交
2188
  uint64_t clusterId = pCtg->clusterId;
D
dapan1121 已提交
2189
  
D
dapan1121 已提交
2190
  ctgFreeHandle(pCtg);
D
dapan1121 已提交
2191 2192

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

D
dapan1121 已提交
2195
int32_t catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* version, int64_t* dbId, int32_t *tableNum) {
D
dapan1121 已提交
2196 2197
  CTG_API_ENTER();

D
dapan1121 已提交
2198
  if (NULL == pCtg || NULL == dbFName || NULL == version || NULL == dbId) {
D
dapan1121 已提交
2199 2200 2201 2202
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2203
    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
2204
    ctgInfo("empty db cache, dbFName:%s", dbFName);
D
dapan1121 已提交
2205
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2206 2207
  }

D
dapan1121 已提交
2208 2209 2210
  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
  if (NULL == dbCache) {
D
dapan1121 已提交
2211
    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
2212
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2213 2214
  }

D
dapan 已提交
2215 2216 2217
  bool inCache = false;
  ctgAcquireVgInfo(pCtg, dbCache, &inCache);
  if (!inCache) {
D
dapan1121 已提交
2218
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
2219 2220

    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
2221
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2222 2223
  }

D
dapan1121 已提交
2224
  *version = dbCache->vgInfo->vgVersion;
D
dapan1121 已提交
2225
  *dbId = dbCache->dbId;
D
dapan1121 已提交
2226
  *tableNum = dbCache->vgInfo->numOfTable;
D
dapan1121 已提交
2227 2228 2229

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

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

D
dapan1121 已提交
2233
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2234 2235
}

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

D
dapan1121 已提交
2239 2240 2241
  if (NULL == pCtg || NULL == dbFName || NULL == pRpc || NULL == pMgmtEps || NULL == vgroupList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
2242

D
dapan1121 已提交
2243
  SCtgDBCache* dbCache = NULL;
2244
  int32_t code = 0;
D
dapan1121 已提交
2245
  SArray *vgList = NULL;
D
dapan1121 已提交
2246 2247 2248 2249 2250 2251 2252
  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 已提交
2253 2254
  }

D
dapan1121 已提交
2255
  CTG_ERR_JRET(ctgGenerateVgList(pCtg, vgHash, &vgList));
D
dapan1121 已提交
2256 2257 2258 2259 2260

  *vgroupList = vgList;
  vgList = NULL;

_return:
D
dapan1121 已提交
2261 2262

  if (dbCache) {
D
dapan1121 已提交
2263 2264
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
2265 2266
  }

D
dapan1121 已提交
2267 2268 2269
  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
    tfree(vgInfo);
D
dapan1121 已提交
2270 2271
  }

D
dapan1121 已提交
2272
  CTG_API_LEAVE(code);  
D
dapan1121 已提交
2273 2274 2275
}


D
dapan1121 已提交
2276
int32_t catalogUpdateDBVgInfo(SCatalog* pCtg, const char* dbFName, uint64_t dbId, SDBVgInfo* dbInfo) {
D
dapan1121 已提交
2277
  CTG_API_ENTER();
D
dapan1121 已提交
2278 2279

  int32_t code = 0;
D
dapan1121 已提交
2280
  
D
dapan1121 已提交
2281
  if (NULL == pCtg || NULL == dbFName || NULL == dbInfo) {
D
dapan1121 已提交
2282 2283 2284
    CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2285 2286 2287 2288
  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 已提交
2289
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
2290 2291
  }

D
dapan1121 已提交
2292 2293 2294 2295
  msg->pCtg = pCtg;
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  msg->dbId = dbId;
  msg->dbInfo = dbInfo;
D
dapan1121 已提交
2296

D
dapan1121 已提交
2297
  action.data = msg;
D
dapan1121 已提交
2298

D
dapan1121 已提交
2299
  CTG_ERR_JRET(ctgPushAction(&action));
D
dapan1121 已提交
2300

D
dapan1121 已提交
2301 2302
  dbInfo = NULL;

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

D
dapan1121 已提交
2305 2306
  CTG_API_LEAVE(code);
  
D
dapan1121 已提交
2307 2308
_return:

D
dapan1121 已提交
2309
  ctgFreeVgInfo(dbInfo);
D
dapan1121 已提交
2310 2311

  tfree(msg);
D
dapan1121 已提交
2312
  
D
dapan1121 已提交
2313
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2314 2315 2316
}


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

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

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

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

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

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

D
dapan1121 已提交
2339 2340 2341
int32_t catalogRemoveStbMeta(SCatalog* pCtg, const char* dbFName, uint64_t dbId, const char* stbName, uint64_t suid) {
  CTG_API_ENTER();

D
dapan 已提交
2342 2343
  int32_t code = 0;
  
D
dapan1121 已提交
2344 2345
  if (NULL == pCtg || NULL == dbFName || NULL == stbName) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
2346 2347
  }

D
dapan1121 已提交
2348
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2349
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2350
  }
D
dapan1121 已提交
2351

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

D
dapan1121 已提交
2354
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2355
  
D
dapan1121 已提交
2356 2357
_return:

D
dapan1121 已提交
2358
  CTG_API_LEAVE(code);
D
dapan 已提交
2359 2360
}

D
dapan1121 已提交
2361

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

D
dapan1121 已提交
2365
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_UNKNOWN_STB));
D
dapan1121 已提交
2366
}
D
dapan1121 已提交
2367

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

D
dapan1121 已提交
2371
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_STB));
D
dapan1121 已提交
2372 2373
}

D
dapan1121 已提交
2374 2375 2376 2377 2378 2379 2380 2381 2382 2383 2384 2385 2386
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 已提交
2387 2388
  int32_t code = 0;

D
dapan1121 已提交
2389 2390
  strcpy(output->dbFName, rspMsg->dbFName);
  strcpy(output->tbName, rspMsg->tbName);
D
dapan1121 已提交
2391

D
dapan1121 已提交
2392
  output->dbId = rspMsg->dbId;
D
dapan1121 已提交
2393
  
D
dapan1121 已提交
2394
  SET_META_TYPE_TABLE(output->metaType);
D
dapan1121 已提交
2395
  
D
dapan1121 已提交
2396
  CTG_ERR_JRET(queryCreateTableMetaFromMsg(rspMsg, true, &output->tbMeta));
D
dapan1121 已提交
2397

D
dapan1121 已提交
2398 2399 2400 2401 2402 2403
  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 已提交
2404

D
dapan1121 已提交
2405 2406 2407 2408 2409 2410 2411
  msg->pCtg = pCtg;
  msg->output = output;

  action.data = msg;

  CTG_ERR_JRET(ctgPushAction(&action));

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

D
dapan1121 已提交
2414 2415
  CTG_API_LEAVE(code);
  
D
dapan1121 已提交
2416 2417
_return:

D
dapan1121 已提交
2418 2419 2420
  tfree(output->tbMeta);
  tfree(output);
  tfree(msg);
D
dapan1121 已提交
2421
  
D
dapan1121 已提交
2422
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2423 2424 2425
}


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

D
dapan1121 已提交
2429
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pTableName) {
D
dapan1121 已提交
2430 2431 2432
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2433
  CTG_API_LEAVE(ctgRefreshTblMeta(pCtg, pTrans, pMgmtEps, pTableName, CTG_FLAG_FORCE_UPDATE | CTG_FLAG_MAKE_STB(isSTable), NULL));
2434
}
2435

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

D
dapan1121 已提交
2439
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_FORCE_UPDATE | CTG_FLAG_MAKE_STB(isSTable)));
D
dapan1121 已提交
2440 2441
}

D
dapan1121 已提交
2442
int32_t catalogGetTableDistVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgList) {
D
dapan1121 已提交
2443
  CTG_API_ENTER();
D
dapan1121 已提交
2444 2445 2446 2447

  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == pVgList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
2448 2449 2450 2451 2452

  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 已提交
2453 2454 2455 2456
  
  STableMeta *tbMeta = NULL;
  int32_t code = 0;
  SVgroupInfo vgroupInfo = {0};
D
dapan1121 已提交
2457
  SCtgDBCache* dbCache = NULL;
D
dapan1121 已提交
2458
  SArray *vgList = NULL;
D
dapan 已提交
2459
  SDBVgInfo *vgInfo = NULL;
D
dapan1121 已提交
2460

D
dapan1121 已提交
2461
  *pVgList = NULL;
D
dapan1121 已提交
2462
  
D
dapan1121 已提交
2463
  CTG_ERR_JRET(ctgGetTableMeta(pCtg, pRpc, pMgmtEps, pTableName, &tbMeta, CTG_FLAG_UNKNOWN_STB));
D
dapan1121 已提交
2464

H
Haojun Liao 已提交
2465 2466
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);
D
dapan 已提交
2467

D
dapan1121 已提交
2468 2469 2470 2471 2472 2473 2474 2475 2476 2477
  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 已提交
2478 2479 2480
  if (0 == tbMeta->vgId) {
    SVgroupInfo vgroup = {0};
    
D
dapan1121 已提交
2481
    catalogGetTableHashVgroup(pCtg, pRpc, pMgmtEps, pTableName, &vgroup);
D
dapan 已提交
2482 2483 2484

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

2487
  if (tbMeta->tableType == TSDB_SUPER_TABLE) {
D
dapan1121 已提交
2488
    CTG_ERR_JRET(ctgGenerateVgList(pCtg, vgHash, pVgList));
D
dapan1121 已提交
2489
  } else {
2490
    int32_t vgId = tbMeta->vgId;
D
dapan1121 已提交
2491
    if (NULL == taosHashGetClone(vgHash, &vgId, sizeof(vgId), &vgroupInfo)) {
2492
      ctgError("table's vgId not found in vgroup list, vgId:%d, tbName:%s", vgId, tNameGetTableName(pTableName));
D
dapan 已提交
2493
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);    
2494
    }
D
dapan1121 已提交
2495

D
dapan1121 已提交
2496 2497
    vgList = taosArrayInit(1, sizeof(SVgroupInfo));
    if (NULL == vgList) {
D
dapan1121 已提交
2498
      ctgError("taosArrayInit %d failed", (int32_t)sizeof(SVgroupInfo));
D
dapan 已提交
2499 2500 2501
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);    
    }

D
dapan1121 已提交
2502
    if (NULL == taosArrayPush(vgList, &vgroupInfo)) {
2503
      ctgError("taosArrayPush vgroupInfo to array failed, vgId:%d, tbName:%s", vgId, tNameGetTableName(pTableName));
D
dapan1121 已提交
2504 2505
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
    }
D
dapan 已提交
2506

D
dapan1121 已提交
2507
    *pVgList = vgList;
D
dapan1121 已提交
2508 2509
    vgList = NULL;
  }
D
dapan 已提交
2510

D
dapan1121 已提交
2511
_return:
D
dapan 已提交
2512

D
dapan1121 已提交
2513
  if (dbCache) {
D
dapan1121 已提交
2514 2515 2516 2517 2518 2519 2520 2521 2522
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
  }

  tfree(tbMeta);

  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
    tfree(vgInfo);
D
dapan1121 已提交
2523 2524 2525 2526 2527 2528
  }

  if (vgList) {
    taosArrayDestroy(vgList);
    vgList = NULL;
  }
D
dapan1121 已提交
2529
  
D
dapan1121 已提交
2530
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2531 2532 2533
}


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

D
dapan1121 已提交
2537 2538 2539 2540 2541
  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 已提交
2542 2543
  SCtgDBCache* dbCache = NULL;
  int32_t code = 0;
H
Haojun Liao 已提交
2544 2545
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);
D
dapan1121 已提交
2546

D
dapan1121 已提交
2547
  SDBVgInfo *vgInfo = NULL;
D
dapan1121 已提交
2548
  CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pTrans, pMgmtEps, db, false, &dbCache, &vgInfo));
D
dapan1121 已提交
2549

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

D
dapan1121 已提交
2552
_return:
D
dapan1121 已提交
2553

D
dapan1121 已提交
2554
  if (dbCache) {
D
dapan1121 已提交
2555 2556 2557 2558 2559 2560 2561
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
  }

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

D
dapan1121 已提交
2564
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2565 2566 2567
}


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

D
dapan1121 已提交
2571
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pReq || NULL == pRsp) {
D
dapan1121 已提交
2572 2573 2574
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2575
  int32_t code = 0;
D
dapan1121 已提交
2576
  pRsp->pTableMeta = NULL;
D
dapan1121 已提交
2577 2578 2579

  if (pReq->pTableName) {
    int32_t tbNum = (int32_t)taosArrayGetSize(pReq->pTableName);
D
dapan1121 已提交
2580
    if (tbNum <= 0) {
D
dapan1121 已提交
2581
      ctgError("empty table name list, tbNum:%d", tbNum);
D
dapan1121 已提交
2582
      CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2583
    }
H
Haojun Liao 已提交
2584

D
dapan1121 已提交
2585 2586
    pRsp->pTableMeta = taosArrayInit(tbNum, POINTER_BYTES);
    if (NULL == pRsp->pTableMeta) {
D
dapan1121 已提交
2587
      ctgError("taosArrayInit %d failed", tbNum);
D
dapan1121 已提交
2588
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
2589 2590 2591 2592 2593 2594
    }
    
    for (int32_t i = 0; i < tbNum; ++i) {
      SName *name = taosArrayGet(pReq->pTableName, i);
      STableMeta *pTableMeta = NULL;
      
D
dapan1121 已提交
2595
      CTG_ERR_JRET(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, name, &pTableMeta, CTG_FLAG_UNKNOWN_STB));
D
dapan1121 已提交
2596 2597 2598 2599 2600 2601 2602 2603 2604

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

D
dapan1121 已提交
2605
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2606 2607

_return:  
D
dapan1121 已提交
2608

D
dapan1121 已提交
2609 2610 2611 2612 2613 2614 2615 2616
  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 已提交
2617
    pRsp->pTableMeta = NULL;
D
dapan1121 已提交
2618
  }
D
dapan 已提交
2619
  
D
dapan1121 已提交
2620
  CTG_API_LEAVE(code);
2621
}
D
dapan 已提交
2622

D
dapan1121 已提交
2623
int32_t catalogGetQnodeList(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SArray* pQnodeList) {
D
dapan1121 已提交
2624 2625
  CTG_API_ENTER();

D
dapan1121 已提交
2626 2627 2628 2629
  if (NULL == pCtg || NULL == pRpc  || NULL == pMgmtEps || NULL == pQnodeList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2630
  //TODO
D
dapan 已提交
2631

D
dapan1121 已提交
2632
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2633 2634
}

D
dapan1121 已提交
2635
int32_t catalogGetExpiredSTables(SCatalog* pCtg, SSTableMetaVersion **stables, uint32_t *num) {
D
dapan1121 已提交
2636 2637
  CTG_API_ENTER();

D
dapan1121 已提交
2638 2639
  if (NULL == pCtg || NULL == stables || NULL == num) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2640 2641
  }

D
dapan1121 已提交
2642 2643 2644 2645
  CTG_API_LEAVE(ctgMetaRentGet(&pCtg->stbRent, (void **)stables, num, sizeof(SSTableMetaVersion)));
}

int32_t catalogGetExpiredDBs(SCatalog* pCtg, SDbVgVersion **dbs, uint32_t *num) {
D
dapan1121 已提交
2646
  CTG_API_ENTER();
D
dapan1121 已提交
2647 2648 2649 2650
  
  if (NULL == pCtg || NULL == dbs || NULL == num) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
2651

D
dapan1121 已提交
2652
  CTG_API_LEAVE(ctgMetaRentGet(&pCtg->dbRent, (void **)dbs, num, sizeof(SDbVgVersion)));
D
dapan1121 已提交
2653 2654
}

D
dapan 已提交
2655

D
dapan 已提交
2656
void catalogDestroy(void) {
D
dapan1121 已提交
2657 2658
  qInfo("start to destroy catalog");
  
D
dapan 已提交
2659
  if (NULL == gCtgMgmt.pCluster || atomic_load_8(&gCtgMgmt.exit)) {
D
dapan1121 已提交
2660 2661 2662
    return;
  }

D
dapan 已提交
2663 2664 2665
  atomic_store_8(&gCtgMgmt.exit, true);

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

D
dapan1121 已提交
2667 2668 2669 2670
  while (CTG_IS_LOCKED(&gCtgMgmt.lock)) {
    usleep(1);
  }
  
D
dapan 已提交
2671
  CTG_LOCK(CTG_WRITE, &gCtgMgmt.lock);
D
dapan1121 已提交
2672

D
dapan1121 已提交
2673
  SCatalog *pCtg = NULL;
D
dapan 已提交
2674
  void *pIter = taosHashIterate(gCtgMgmt.pCluster, NULL);
D
dapan1121 已提交
2675
  while (pIter) {
D
dapan1121 已提交
2676
    pCtg = *(SCatalog **)pIter;
D
dapan1121 已提交
2677

D
dapan1121 已提交
2678 2679
    if (pCtg) {
      catalogFreeHandle(pCtg);
D
dapan1121 已提交
2680 2681
    }
    
D
dapan 已提交
2682
    pIter = taosHashIterate(gCtgMgmt.pCluster, pIter);
D
dapan 已提交
2683
  }
D
dapan1121 已提交
2684
  
D
dapan 已提交
2685 2686
  taosHashCleanup(gCtgMgmt.pCluster);
  gCtgMgmt.pCluster = NULL;
D
dapan1121 已提交
2687

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

D
dapan1121 已提交
2690
  qInfo("catalog destroyed");
D
dapan 已提交
2691 2692 2693 2694
}