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

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

D
dapan 已提交
21 22 23 24 25 26
int32_t ctgActUpdateVg(SCtgMetaAction *action);
int32_t ctgActUpdateTbl(SCtgMetaAction *action);
int32_t ctgActRemoveDB(SCtgMetaAction *action);
int32_t ctgActRemoveStb(SCtgMetaAction *action);
int32_t ctgActRemoveTbl(SCtgMetaAction *action);

D
dapan 已提交
27
SCatalogMgmt gCtgMgmt = {0};
D
dapan1121 已提交
28
SCtgDebug gCTGDebug = {0};
D
dapan 已提交
29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57
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")) {
D
dapan1121 已提交
58
    gCTGDebug.lockEnable = true;
D
dapan 已提交
59 60 61 62 63
    qDebug("lock debug enabled");
    return TSDB_CODE_SUCCESS;
  }

  if (0 == strcasecmp(option, "cache")) {
D
dapan1121 已提交
64
    gCTGDebug.cacheEnable = true;
D
dapan 已提交
65 66 67 68 69
    qDebug("cache debug enabled");
    return TSDB_CODE_SUCCESS;
  }

  if (0 == strcasecmp(option, "api")) {
D
dapan1121 已提交
70
    gCTGDebug.apiEnable = true;
D
dapan 已提交
71 72 73 74
    qDebug("api debug enabled");
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
75
  if (0 == strcasecmp(option, "meta")) {
D
dapan1121 已提交
76
    gCTGDebug.metaEnable = true;
D
dapan1121 已提交
77 78 79 80
    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
void ctgDbgShowTableMeta(SCatalog* pCtg, const char *tbName, STableMeta* p) {
D
dapan1121 已提交
158
  if (!gCTGDebug.metaEnable) {
D
dapan1121 已提交
159 160 161 162 163 164 165 166 167 168 169 170
    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);
  }
H
Haojun Liao 已提交
171

D
dapan1121 已提交
172 173 174 175 176 177
  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
void ctgDbgShowDBCache(SCatalog* pCtg, SHashObj *dbHash) {
D
dapan1121 已提交
180
  if (NULL == dbHash || !gCTGDebug.cacheEnable) {
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
    dbFName = taosHashGetKey(pIter, &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
      }
H
Haojun Liao 已提交
207
    }
D
dapan1121 已提交
208
    
H
Haojun Liao 已提交
209
    ctgDebug("[%d] db [%.*s][%"PRIx64"] %s: metaNum:%d, stbNum:%d, vgVersion:%d, hashMethod:%d, vgNum:%d",
D
dapan1121 已提交
210 211
      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.cacheEnable || NULL == pCtg) {
D
dapan1121 已提交
221 222 223
    return;
  }

D
dapan1121 已提交
224
  ctgDebug("## cluster %"PRIx64" %p cache Info ##", pCtg->clusterId, pCtg);
H
Haojun Liao 已提交
225
  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

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

wafwerar's avatar
wafwerar 已提交
245
  taosMemoryFreeClear(mgmt->slots);
D
dapan1121 已提交
246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274
}


void ctgFreeTableMetaCache(SCtgTbMetaCache *cache) {
  CTG_LOCK(CTG_WRITE, &cache->stbLock);
  if (cache->stbCache) {
    taosHashCleanup(cache->stbCache);
    cache->stbCache = NULL;
  }
  CTG_UNLOCK(CTG_WRITE, &cache->stbLock);

  CTG_LOCK(CTG_WRITE, &cache->metaLock);
  if (cache->metaCache) {
    taosHashCleanup(cache->metaCache);
    cache->metaCache = NULL;
  }
  CTG_UNLOCK(CTG_WRITE, &cache->metaLock);
}

void ctgFreeVgInfo(SDBVgInfo *vgInfo) {
  if (NULL == vgInfo) {
    return;
  }

  if (vgInfo->vgHash) {
    taosHashCleanup(vgInfo->vgHash);
    vgInfo->vgHash = NULL;
  }
  
wafwerar's avatar
wafwerar 已提交
275
  taosMemoryFreeClear(vgInfo);
D
dapan1121 已提交
276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309
}

void ctgFreeDbCache(SCtgDBCache *dbCache) {
  if (NULL == dbCache) {
    return;
  }

  CTG_LOCK(CTG_WRITE, &dbCache->vgLock);
  ctgFreeVgInfo (dbCache->vgInfo);
  CTG_UNLOCK(CTG_WRITE, &dbCache->vgLock);

  ctgFreeTableMetaCache(&dbCache->tbCache);
}


void ctgFreeHandle(SCatalog* pCtg) {
  ctgFreeMetaRent(&pCtg->dbRent);
  ctgFreeMetaRent(&pCtg->stbRent);
  
  if (pCtg->dbCache) {
    void *pIter = taosHashIterate(pCtg->dbCache, NULL);
    while (pIter) {
      SCtgDBCache *dbCache = pIter;

      atomic_store_8(&dbCache->deleted, 1);

      ctgFreeDbCache(dbCache);
            
      pIter = taosHashIterate(pCtg->dbCache, pIter);
    }  

    taosHashCleanup(pCtg->dbCache);
  }
  
wafwerar's avatar
wafwerar 已提交
310
  taosMemoryFree(pCtg);
D
dapan1121 已提交
311 312 313 314 315 316 317 318
}



void ctgWaitAction(SCtgMetaAction *action) {
  while (true) {
    tsem_wait(&gCtgMgmt.queue.rspSem);
    
X
Xiaoyu Wang 已提交
319
    if (atomic_load_8((int8_t*)&gCtgMgmt.exit)) {
D
dapan1121 已提交
320 321 322 323 324 325 326 327 328 329 330 331
      tsem_post(&gCtgMgmt.queue.rspSem);
      break;
    }

    if (gCtgMgmt.queue.seqDone >= action->seqId) {
      break;
    }

    tsem_post(&gCtgMgmt.queue.rspSem);
    sched_yield();
  }
}
D
dapan 已提交
332 333

void ctgPopAction(SCtgMetaAction **action) {
D
dapan1121 已提交
334
  SCtgQNode *orig = gCtgMgmt.queue.head;
D
dapan 已提交
335
  
D
dapan1121 已提交
336 337
  SCtgQNode *node = gCtgMgmt.queue.head->next;
  gCtgMgmt.queue.head = gCtgMgmt.queue.head->next;
D
dapan 已提交
338 339 340

  CTG_QUEUE_SUB();
  
wafwerar's avatar
wafwerar 已提交
341
  taosMemoryFreeClear(orig);
D
dapan 已提交
342 343 344 345 346

  *action = &node->action;
}


D
dapan1121 已提交
347
int32_t ctgPushAction(SCatalog* pCtg, SCtgMetaAction *action) {
wafwerar's avatar
wafwerar 已提交
348
  SCtgQNode *node = taosMemoryCalloc(1, sizeof(SCtgQNode));
D
dapan 已提交
349 350 351 352
  if (NULL == node) {
    qError("calloc %d failed", (int32_t)sizeof(SCtgQNode));
    CTG_RET(TSDB_CODE_CTG_MEM_ERROR);
  }
D
dapan1121 已提交
353 354

  action->seqId = atomic_add_fetch_64(&gCtgMgmt.queue.seqId, 1);
D
dapan 已提交
355 356 357
  
  node->action = *action;

D
dapan1121 已提交
358 359 360 361
  CTG_LOCK(CTG_WRITE, &gCtgMgmt.queue.qlock);
  gCtgMgmt.queue.tail->next = node;
  gCtgMgmt.queue.tail = node;
  CTG_UNLOCK(CTG_WRITE, &gCtgMgmt.queue.qlock);
D
dapan 已提交
362 363

  CTG_QUEUE_ADD();
D
dapan1121 已提交
364
  CTG_STAT_ADD(gCtgMgmt.stat.runtime.qNum);
D
dapan 已提交
365

D
dapan1121 已提交
366
  tsem_post(&gCtgMgmt.queue.reqSem);
D
dapan 已提交
367

D
dapan1121 已提交
368
  ctgDebug("action [%s] added into queue", gCtgAction[action->act].name);
D
dapan 已提交
369

D
dapan1121 已提交
370 371
  if (action->syncReq) {
    ctgWaitAction(action);
D
dapan1121 已提交
372 373
  }

D
dapan1121 已提交
374
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
375 376 377
}


D
dapan1121 已提交
378 379 380
int32_t ctgPushRmDBMsgInQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId) {
  int32_t code = 0;
  SCtgMetaAction action= {.act = CTG_ACT_REMOVE_DB};
wafwerar's avatar
wafwerar 已提交
381
  SCtgRemoveDBMsg *msg = taosMemoryMalloc(sizeof(SCtgRemoveDBMsg));
D
dapan1121 已提交
382 383 384 385 386
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgRemoveDBMsg));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

D
dapan1121 已提交
387 388 389 390 391
  char *p = strchr(dbFName, '.');
  if (p && CTG_IS_INF_DBNAME(p + 1)) {
    dbFName = p + 1;
  }

D
dapan1121 已提交
392 393 394 395 396 397
  msg->pCtg = pCtg;
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  msg->dbId = dbId;

  action.data = msg;

D
dapan1121 已提交
398
  CTG_ERR_JRET(ctgPushAction(pCtg, &action));
D
dapan1121 已提交
399 400 401 402

  return TSDB_CODE_SUCCESS;

_return:
H
Haojun Liao 已提交
403

wafwerar's avatar
wafwerar 已提交
404
  taosMemoryFreeClear(action.data);
H
Haojun Liao 已提交
405
  CTG_RET(code);
D
dapan1121 已提交
406 407 408
}


D
dapan1121 已提交
409 410 411
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};
wafwerar's avatar
wafwerar 已提交
412
  SCtgRemoveStbMsg *msg = taosMemoryMalloc(sizeof(SCtgRemoveStbMsg));
D
dapan1121 已提交
413 414 415 416 417 418 419 420 421 422 423 424 425
  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;

D
dapan1121 已提交
426
  CTG_ERR_JRET(ctgPushAction(pCtg, &action));
D
dapan1121 已提交
427 428 429 430

  return TSDB_CODE_SUCCESS;

_return:
H
Haojun Liao 已提交
431

wafwerar's avatar
wafwerar 已提交
432
  taosMemoryFreeClear(action.data);
H
Haojun Liao 已提交
433
  CTG_RET(code);
D
dapan1121 已提交
434 435 436 437 438 439 440
}



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};
wafwerar's avatar
wafwerar 已提交
441
  SCtgRemoveTblMsg *msg = taosMemoryMalloc(sizeof(SCtgRemoveTblMsg));
D
dapan1121 已提交
442 443 444 445 446 447 448 449 450 451 452 453
  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;

D
dapan1121 已提交
454
  CTG_ERR_JRET(ctgPushAction(pCtg, &action));
D
dapan1121 已提交
455 456 457 458

  return TSDB_CODE_SUCCESS;

_return:
H
Haojun Liao 已提交
459

wafwerar's avatar
wafwerar 已提交
460
  taosMemoryFreeClear(action.data);
H
Haojun Liao 已提交
461
  CTG_RET(code);
D
dapan1121 已提交
462 463
}

D
dapan1121 已提交
464 465 466
int32_t ctgPushUpdateVgMsgInQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, SDBVgInfo* dbInfo, bool syncReq) {
  int32_t code = 0;
  SCtgMetaAction action= {.act = CTG_ACT_UPDATE_VG, .syncReq = syncReq};
wafwerar's avatar
wafwerar 已提交
467
  SCtgUpdateVgMsg *msg = taosMemoryMalloc(sizeof(SCtgUpdateVgMsg));
D
dapan1121 已提交
468 469 470 471
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateVgMsg));
    ctgFreeVgInfo(dbInfo);
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
472 473
  }

D
dapan1121 已提交
474 475 476 477 478
  char *p = strchr(dbFName, '.');
  if (p && CTG_IS_INF_DBNAME(p + 1)) {
    dbFName = p + 1;
  }

D
dapan1121 已提交
479 480 481 482
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  msg->pCtg = pCtg;
  msg->dbId = dbId;
  msg->dbInfo = dbInfo;
D
dapan1121 已提交
483

D
dapan1121 已提交
484
  action.data = msg;
D
dapan1121 已提交
485

D
dapan1121 已提交
486
  CTG_ERR_JRET(ctgPushAction(pCtg, &action));
D
dapan1121 已提交
487

D
dapan1121 已提交
488
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
489

D
dapan1121 已提交
490
_return:
D
dapan1121 已提交
491

D
dapan1121 已提交
492
  ctgFreeVgInfo(dbInfo);
wafwerar's avatar
wafwerar 已提交
493
  taosMemoryFreeClear(action.data);
D
dapan1121 已提交
494
  CTG_RET(code);
D
dapan1121 已提交
495 496
}

D
dapan1121 已提交
497 498 499
int32_t ctgPushUpdateTblMsgInQueue(SCatalog* pCtg, STableMetaOutput *output, bool syncReq) {
  int32_t code = 0;
  SCtgMetaAction action= {.act = CTG_ACT_UPDATE_TBL};
wafwerar's avatar
wafwerar 已提交
500
  SCtgUpdateTblMsg *msg = taosMemoryMalloc(sizeof(SCtgUpdateTblMsg));
D
dapan1121 已提交
501 502 503 504 505
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateTblMsg));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

D
dapan1121 已提交
506 507 508 509 510
  char *p = strchr(output->dbFName, '.');
  if (p && CTG_IS_INF_DBNAME(p + 1)) {
    memmove(output->dbFName, p + 1, strlen(p + 1));
  }

D
dapan1121 已提交
511 512 513 514 515 516 517 518 519 520 521
  msg->pCtg = pCtg;
  msg->output = output;

  action.data = msg;

  CTG_ERR_JRET(ctgPushAction(pCtg, &action));

  return TSDB_CODE_SUCCESS;
  
_return:

wafwerar's avatar
wafwerar 已提交
522
  taosMemoryFreeClear(msg);
D
dapan1121 已提交
523 524 525 526
  
  CTG_RET(code);
}

D
dapan1121 已提交
527

D
dapan 已提交
528
int32_t ctgAcquireVgInfo(SCatalog *pCtg, SCtgDBCache *dbCache, bool *inCache) {
D
dapan1121 已提交
529 530
  CTG_LOCK(CTG_READ, &dbCache->vgLock);
  
D
dapan 已提交
531 532 533 534 535 536 537 538 539 540
  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 已提交
541 542 543
  if (NULL == dbCache->vgInfo) {
    CTG_UNLOCK(CTG_READ, &dbCache->vgLock);

D
dapan 已提交
544
    *inCache = false;
D
dapan1121 已提交
545
    ctgDebug("db vgInfo is empty, dbId:%"PRIx64, dbCache->dbId);
D
dapan1121 已提交
546 547 548
    return TSDB_CODE_SUCCESS;
  }

D
dapan 已提交
549 550
  *inCache = true;
  
D
dapan1121 已提交
551 552 553 554 555 556 557 558 559 560 561 562 563 564
  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 已提交
565

D
dapan1121 已提交
566 567 568
void ctgReleaseDBCache(SCatalog *pCtg, SCtgDBCache *dbCache) {
  taosHashRelease(pCtg->dbCache, dbCache);
}
D
dapan1121 已提交
569

D
dapan1121 已提交
570 571 572
void ctgReleaseVgInfo(SCtgDBCache *dbCache) {
  CTG_UNLOCK(CTG_READ, &dbCache->vgLock);
}
D
dapan1121 已提交
573

D
dapan1121 已提交
574 575 576
void ctgWReleaseVgInfo(SCtgDBCache *dbCache) {
  CTG_UNLOCK(CTG_WRITE, &dbCache->vgLock);
}
D
dapan1121 已提交
577

D
dapan1121 已提交
578

D
dapan 已提交
579
int32_t ctgAcquireDBCacheImpl(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache, bool acquire) {
D
dapan1121 已提交
580 581 582 583 584
  char *p = strchr(dbFName, '.');
  if (p && CTG_IS_INF_DBNAME(p + 1)) {
    dbFName = p + 1;
  }

D
dapan 已提交
585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621
  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 已提交
622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637
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 已提交
638 639
  ctgAcquireVgInfo(pCtg, dbCache, inCache);
  if (!(*inCache)) {
D
dapan1121 已提交
640 641 642 643
    ctgReleaseDBCache(pCtg, dbCache);
  
    *pCache = NULL;
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
644
  }
D
dapan1121 已提交
645

D
dapan1121 已提交
646
  *pCache = dbCache;
D
dapan1121 已提交
647
  *inCache = true;
D
dapan1121 已提交
648

D
dapan1121 已提交
649
  ctgDebug("Got db vgInfo from cache, dbFName:%s", dbFName);
D
dapan1121 已提交
650 651 652 653
  
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675
int32_t ctgGetQnodeListFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SArray **out) {
  char *msg = NULL;
  int32_t msgLen = 0;

  ctgDebug("try to get qnode list from mnode, mgmtEpInUse:%d", pMgmtEps->inUse);

  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_MND_QNODE_LIST)](NULL, &msg, 0, &msgLen);
  if (code) {
    ctgError("Build qnode list msg failed, error:%s", tstrerror(code));
    CTG_ERR_RET(code);
  }
  
  SRpcMsg rpcMsg = {
      .msgType = TDMT_MND_QNODE_LIST,
      .pCont   = msg,
      .contLen = msgLen,
  };

  SRpcMsg rpcRsp = {0};

  rpcSendRecv(pRpc, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
D
dapan1121 已提交
676
    ctgError("error rsp for qnode list, error:%s", tstrerror(rpcRsp.code));
D
dapan1121 已提交
677 678 679 680 681 682 683 684 685
    CTG_ERR_RET(rpcRsp.code);
  }

  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_QNODE_LIST)](out, rpcRsp.pCont, rpcRsp.contLen);
  if (code) {
    ctgError("Process qnode list rsp failed, error:%s", tstrerror(rpcRsp.code));
    CTG_ERR_RET(code);
  }

D
dapan1121 已提交
686
  ctgDebug("Got qnode list from mnode, listNum:%d", (int32_t)taosArrayGetSize(*out));
D
dapan1121 已提交
687 688 689

  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
690 691


D
dapan1121 已提交
692
int32_t ctgGetDBVgInfoFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SBuildUseDBInput *input, SUseDbOutput *out) {
D
dapan1121 已提交
693 694 695
  char *msg = NULL;
  int32_t msgLen = 0;

D
dapan 已提交
696
  ctgDebug("try to get db vgInfo from mnode, dbFName:%s", input->db);
D
dapan1121 已提交
697 698 699 700 701 702

  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 已提交
703
  
D
dapan1121 已提交
704
  SRpcMsg rpcMsg = {
H
Hongze Cheng 已提交
705
      .msgType = TDMT_MND_USE_DB,
D
catalog  
dapan1121 已提交
706
      .pCont   = msg,
D
dapan1121 已提交
707 708 709 710 711 712
      .contLen = msgLen,
  };

  SRpcMsg rpcRsp = {0};

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

D
dapan1121 已提交
718 719 720 721 722
  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 已提交
723

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

D
dapan1121 已提交
726 727
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
728

D
dapan1121 已提交
729 730
int32_t ctgIsTableMetaExistInCache(SCatalog* pCtg, char *dbFName, char* tbName, int32_t *exist) {
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
731
    *exist = 0;
D
dapan1121 已提交
732 733 734 735
    ctgWarn("empty db cache, dbFName:%s, tbName:%s", dbFName, tbName);
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
736 737
  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
738 739
  if (NULL == dbCache) {
    *exist = 0;
D
dapan1121 已提交
740 741 742
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
743
  size_t sz = 0;
D
dapan1121 已提交
744 745 746 747
  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 已提交
748
  if (NULL == tbMeta) {
D
dapan 已提交
749
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
750
    
D
dapan1121 已提交
751
    *exist = 0;
D
dapan1121 已提交
752
    ctgDebug("tbmeta not in cache, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
753 754 755 756
    return TSDB_CODE_SUCCESS;
  }

  *exist = 1;
D
dapan1121 已提交
757

D
dapan 已提交
758
  ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
759
  
D
dapan1121 已提交
760
  ctgDebug("tbmeta is in cache, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
761 762 763 764
  
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
765

D
dapan1121 已提交
766
int32_t ctgGetTableMetaFromCache(SCatalog* pCtg, const SName* pTableName, STableMeta** pTableMeta, int32_t *exist, int32_t flag, uint64_t *dbId) {
D
dapan1121 已提交
767
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
768
    *exist = 0;
D
dapan1121 已提交
769
    ctgWarn("empty tbmeta cache, tbName:%s", pTableName->tname);
D
dapan1121 已提交
770 771 772
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
773
  char dbFName[TSDB_DB_FNAME_LEN] = {0};
D
dapan1121 已提交
774
  if (CTG_FLAG_IS_INF_DB(flag)) {
D
dapan1121 已提交
775 776 777 778
    strcpy(dbFName, pTableName->dbname);
  } else {
    tNameGetFullDbName(pTableName, dbFName);
  }
D
dapan1121 已提交
779

D
dapan1121 已提交
780 781
  *pTableMeta = NULL;

D
dapan1121 已提交
782 783
  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
784 785 786 787 788
  if (NULL == dbCache) {
    *exist = 0;
    return TSDB_CODE_SUCCESS;
  }
  
H
Haojun Liao 已提交
789
  int32_t sz = 0;
D
dapan1121 已提交
790
  CTG_LOCK(CTG_READ, &dbCache->tbCache.metaLock);
H
Haojun Liao 已提交
791
  int32_t code = taosHashGetDup_m(dbCache->tbCache.metaCache, pTableName->tname, strlen(pTableName->tname), (void **)pTableMeta, &sz);
D
dapan1121 已提交
792 793
  CTG_UNLOCK(CTG_READ, &dbCache->tbCache.metaLock);

D
dapan1121 已提交
794
  if (NULL == *pTableMeta) {
D
dapan1121 已提交
795
    *exist = 0;
D
dapan1121 已提交
796 797
    ctgReleaseDBCache(pCtg, dbCache);
    ctgDebug("tbl not in cache, dbFName:%s, tbName:%s", dbFName, pTableName->tname);
D
dapan1121 已提交
798 799 800
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
801
  *exist = 1;
D
dapan1121 已提交
802 803 804
  if (dbId) {
    *dbId = dbCache->dbId;
  }
H
Haojun Liao 已提交
805

H
Haojun Liao 已提交
806
  STableMeta* tbMeta = *pTableMeta;
D
dapan1121 已提交
807

D
dapan1121 已提交
808
  if (tbMeta->tableType != TSDB_CHILD_TABLE) {
D
dapan1121 已提交
809
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
810
    ctgDebug("Got meta from cache, type:%d, dbFName:%s, tbName:%s", tbMeta->tableType, dbFName, pTableName->tname);
D
dapan1121 已提交
811 812 813
    return TSDB_CODE_SUCCESS;
  }
  
D
dapan1121 已提交
814
  CTG_LOCK(CTG_READ, &dbCache->tbCache.stbLock);
D
dapan1121 已提交
815
  
D
dapan1121 已提交
816
  STableMeta **stbMeta = taosHashGet(dbCache->tbCache.stbCache, &tbMeta->suid, sizeof(tbMeta->suid));
D
dapan1121 已提交
817
  if (NULL == stbMeta || NULL == *stbMeta) {
D
dapan1121 已提交
818
    CTG_UNLOCK(CTG_READ, &dbCache->tbCache.stbLock);
D
dapan1121 已提交
819 820
    ctgReleaseDBCache(pCtg, dbCache);
    ctgError("stb not in stbCache, suid:%"PRIx64, tbMeta->suid);
wafwerar's avatar
wafwerar 已提交
821
    taosMemoryFreeClear(*pTableMeta);
D
dapan1121 已提交
822 823 824
    *exist = 0;
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
825

D
dapan1121 已提交
826
  if ((*stbMeta)->suid != tbMeta->suid) {    
D
dapan1121 已提交
827
    CTG_UNLOCK(CTG_READ, &dbCache->tbCache.stbLock);
D
dapan1121 已提交
828
    ctgReleaseDBCache(pCtg, dbCache);
wafwerar's avatar
wafwerar 已提交
829
    taosMemoryFreeClear(*pTableMeta);
D
dapan1121 已提交
830
    ctgError("stable suid in stbCache mis-match, expected suid:%"PRIx64 ",actual suid:%"PRIx64, tbMeta->suid, (*stbMeta)->suid);
D
dapan1121 已提交
831 832
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }
D
dapan1121 已提交
833

D
dapan1121 已提交
834
  int32_t metaSize = CTG_META_SIZE(*stbMeta);
wafwerar's avatar
wafwerar 已提交
835
  *pTableMeta = taosMemoryRealloc(*pTableMeta, metaSize);
D
dapan1121 已提交
836
  if (NULL == *pTableMeta) {    
D
dapan1121 已提交
837
    CTG_UNLOCK(CTG_READ, &dbCache->tbCache.stbLock);
D
dapan1121 已提交
838
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
839
    ctgError("realloc size[%d] failed", metaSize);
D
dapan1121 已提交
840
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
841 842
  }

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

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

D
dapan1121 已提交
847
  ctgReleaseDBCache(pCtg, dbCache);
D
dapan 已提交
848

D
dapan1121 已提交
849
  ctgDebug("Got tbmeta from cache, dbFName:%s, tbName:%s", dbFName, pTableName->tname);
D
dapan1121 已提交
850 851 852 853
  
  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
854
int32_t ctgGetTableTypeFromCache(SCatalog* pCtg, const char* dbFName, const char *tableName, int32_t *tbType) {
D
dapan1121 已提交
855
  if (NULL == pCtg->dbCache) {
D
dapan 已提交
856
    ctgWarn("empty db cache, dbFName:%s, tbName:%s", dbFName, tableName);  
D
dapan1121 已提交
857 858
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
859
  
D
dapan 已提交
860 861
  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
862 863 864
  if (NULL == dbCache) {
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
865

D
dapan1121 已提交
866
  CTG_LOCK(CTG_READ, &dbCache->tbCache.metaLock);
D
dapan 已提交
867
  STableMeta *pTableMeta = (STableMeta *)taosHashAcquire(dbCache->tbCache.metaCache, tableName, strlen(tableName));
D
dapan1121 已提交
868

D
dapan1121 已提交
869
  if (NULL == pTableMeta) {
D
dapan1121 已提交
870
    CTG_UNLOCK(CTG_READ, &dbCache->tbCache.metaLock);
D
dapan 已提交
871
    ctgWarn("tbl not in cache, dbFName:%s, tbName:%s", dbFName, tableName);  
D
dapan 已提交
872
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
873
    
D
dapan1121 已提交
874 875 876
    return TSDB_CODE_SUCCESS;
  }

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

D
dapan1121 已提交
879 880 881 882
  taosHashRelease(dbCache->tbCache.metaCache, pTableMeta);

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

D
dapan 已提交
883
  ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
884

D
dapan 已提交
885
  ctgDebug("Got tbtype from cache, dbFName:%s, tbName:%s, type:%d", dbFName, tableName, *tbType);  
D
dapan1121 已提交
886 887 888 889
  
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
890
int32_t ctgGetTableMetaFromMnodeImpl(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, char *dbFName, char* tbName, STableMetaOutput* output) {
D
dapan1121 已提交
891
  SBuildTableMetaInput bInput = {.vgId = 0, .dbFName = dbFName, .tbName = tbName};
D
dapan1121 已提交
892 893 894 895
  char *msg = NULL;
  SEpSet *pVnodeEpSet = NULL;
  int32_t msgLen = 0;

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

D
dapan1121 已提交
898
  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_MND_TABLE_META)](&bInput, &msg, 0, &msgLen);
D
dapan1121 已提交
899 900 901 902
  if (code) {
    ctgError("Build mnode stablemeta msg failed, code:%x", code);
    CTG_ERR_RET(code);
  }
D
dapan1121 已提交
903 904

  SRpcMsg rpcMsg = {
D
dapan1121 已提交
905
      .msgType = TDMT_MND_TABLE_META,
D
dapan1121 已提交
906 907 908
      .pCont   = msg,
      .contLen = msgLen,
  };
D
dapan1121 已提交
909

D
dapan1121 已提交
910 911
  SRpcMsg rpcRsp = {0};

D
dapan1121 已提交
912
  rpcSendRecv(pTrans, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
D
dapan1121 已提交
913 914
  
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
D
dapan1121 已提交
915
    if (CTG_TABLE_NOT_EXIST(rpcRsp.code)) {
D
dapan1121 已提交
916
      SET_META_TYPE_NULL(output->metaType);
D
dapan1121 已提交
917
      ctgDebug("stablemeta not exist in mnode, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
918 919 920
      return TSDB_CODE_SUCCESS;
    }
    
D
dapan1121 已提交
921
    ctgError("error rsp for stablemeta from mnode, code:%s, dbFName:%s, tbName:%s", tstrerror(rpcRsp.code), dbFName, tbName);
D
dapan1121 已提交
922 923 924
    CTG_ERR_RET(rpcRsp.code);
  }

D
dapan1121 已提交
925
  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_TABLE_META)](output, rpcRsp.pCont, rpcRsp.contLen);
D
dapan1121 已提交
926
  if (code) {
D
dapan1121 已提交
927
    ctgError("Process mnode stablemeta rsp failed, code:%x, dbFName:%s, tbName:%s", code, dbFName, tbName);
D
dapan1121 已提交
928 929 930
    CTG_ERR_RET(code);
  }

D
dapan1121 已提交
931
  ctgDebug("Got table meta from mnode, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
932 933 934 935

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
936
int32_t ctgGetTableMetaFromMnode(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, STableMetaOutput* output) {
D
dapan1121 已提交
937 938
  char dbFName[TSDB_DB_FNAME_LEN];
  tNameGetFullDbName(pTableName, dbFName);
D
dapan1121 已提交
939

D
dapan1121 已提交
940
  return ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, dbFName, (char *)pTableName->tname, output);
D
dapan1121 已提交
941
}
D
dapan1121 已提交
942

D
dapan1121 已提交
943
int32_t ctgGetTableMetaFromVnodeImpl(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, SVgroupInfo *vgroupInfo, STableMetaOutput* output) {
D
dapan1121 已提交
944
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pTableName || NULL == vgroupInfo || NULL == output) {
D
dapan1121 已提交
945
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
946 947
  }

D
dapan1121 已提交
948 949
  char dbFName[TSDB_DB_FNAME_LEN];
  tNameGetFullDbName(pTableName, dbFName);
D
dapan1121 已提交
950

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

D
dapan1121 已提交
953
  SBuildTableMetaInput bInput = {.vgId = vgroupInfo->vgId, .dbFName = dbFName, .tbName = (char *)tNameGetTableName(pTableName)};
D
dapan1121 已提交
954 955 956
  char *msg = NULL;
  int32_t msgLen = 0;

D
dapan1121 已提交
957 958
  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_VND_TABLE_META)](&bInput, &msg, 0, &msgLen);
  if (code) {
D
dapan1121 已提交
959
    ctgError("Build vnode tablemeta msg failed, code:%x, dbFName:%s, tbName:%s", code, dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
960 961
    CTG_ERR_RET(code);
  }
D
dapan1121 已提交
962 963

  SRpcMsg rpcMsg = {
H
Hongze Cheng 已提交
964
      .msgType = TDMT_VND_TABLE_META,
D
dapan1121 已提交
965 966 967 968 969
      .pCont   = msg,
      .contLen = msgLen,
  };

  SRpcMsg rpcRsp = {0};
L
Liu Jicong 已提交
970
  rpcSendRecv(pTrans, &vgroupInfo->epSet, &rpcMsg, &rpcRsp);
D
dapan1121 已提交
971
  
D
dapan1121 已提交
972
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
D
dapan1121 已提交
973
    if (CTG_TABLE_NOT_EXIST(rpcRsp.code)) {
D
dapan1121 已提交
974
      SET_META_TYPE_NULL(output->metaType);
D
dapan1121 已提交
975
      ctgDebug("tablemeta not exist in vnode, dbFName:%s, tbName:%s", dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
976 977 978
      return TSDB_CODE_SUCCESS;
    }
  
D
dapan1121 已提交
979
    ctgError("error rsp for table meta from vnode, code:%s, dbFName:%s, tbName:%s", tstrerror(rpcRsp.code), dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
980
    CTG_ERR_RET(rpcRsp.code);
D
dapan1121 已提交
981 982
  }

D
dapan1121 已提交
983 984
  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_VND_TABLE_META)](output, rpcRsp.pCont, rpcRsp.contLen);
  if (code) {
D
dapan1121 已提交
985
    ctgError("Process vnode tablemeta rsp failed, code:%s, dbFName:%s, tbName:%s", tstrerror(code), dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
986 987 988
    CTG_ERR_RET(code);
  }

D
dapan1121 已提交
989
  ctgDebug("Got table meta from vnode, dbFName:%s, tbName:%s", dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
990 991 992
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018
int32_t ctgGetTableMetaFromVnode(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, SVgroupInfo *vgroupInfo, STableMetaOutput* output) {
  int32_t code = 0;
  int32_t retryNum = 0;
  
  while (retryNum < CTG_DEFAULT_MAX_RETRY_TIMES) {
    code = ctgGetTableMetaFromVnodeImpl(pCtg, pTrans, pMgmtEps, pTableName, vgroupInfo, output);
    if (code) {
      if (TSDB_CODE_VND_HASH_MISMATCH == code) {
        char dbFName[TSDB_DB_FNAME_LEN];
        tNameGetFullDbName(pTableName, dbFName);
        
        code = catalogRefreshDBVgInfo(pCtg, pTrans, pMgmtEps, dbFName);
        if (code != TSDB_CODE_SUCCESS) {
          break;
        }

        ++retryNum;
        continue;
      }
    }

    break;
  }

  CTG_RET(code);
}
D
dapan1121 已提交
1019

1020 1021
int32_t ctgGetHashFunction(int8_t hashMethod, tableNameHashFp *fp) {
  switch (hashMethod) {
D
dapan1121 已提交
1022 1023 1024 1025 1026 1027 1028 1029
    default:
      *fp = MurmurHash3_32;
      break;
  }

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
1030
int32_t ctgGenerateVgList(SCatalog *pCtg, SHashObj *vgHash, SArray** pList) {
D
dapan1121 已提交
1031
  SHashObj *vgroupHash = NULL;
1032
  SVgroupInfo *vgInfo = NULL;
D
dapan1121 已提交
1033 1034
  SArray *vgList = NULL;
  int32_t code = 0;
D
dapan1121 已提交
1035
  int32_t vgNum = taosHashGetSize(vgHash);
1036

D
dapan1121 已提交
1037
  vgList = taosArrayInit(vgNum, sizeof(SVgroupInfo));
D
dapan1121 已提交
1038
  if (NULL == vgList) {
D
dapan1121 已提交
1039
    ctgError("taosArrayInit failed, num:%d", vgNum);
D
dapan 已提交
1040 1041 1042
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);    
  }

D
dapan1121 已提交
1043
  void *pIter = taosHashIterate(vgHash, NULL);
1044 1045
  while (pIter) {
    vgInfo = pIter;
D
dapan1121 已提交
1046

D
dapan1121 已提交
1047
    if (NULL == taosArrayPush(vgList, vgInfo)) {
D
dapan1121 已提交
1048
      ctgError("taosArrayPush failed, vgId:%d", vgInfo->vgId);
D
dapan1121 已提交
1049
      taosHashCancelIterate(vgHash, pIter);      
D
dapan1121 已提交
1050
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
1051 1052
    }
    
D
dapan1121 已提交
1053
    pIter = taosHashIterate(vgHash, pIter);
1054
    vgInfo = NULL;
D
dapan1121 已提交
1055 1056
  }

D
dapan1121 已提交
1057
  *pList = vgList;
D
dapan1121 已提交
1058

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

D
dapan1121 已提交
1061
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1062 1063 1064 1065 1066 1067 1068 1069

_return:

  if (vgList) {
    taosArrayDestroy(vgList);
  }

  CTG_RET(code);
D
dapan1121 已提交
1070 1071
}

D
dapan1121 已提交
1072
int32_t ctgGetVgInfoFromHashValue(SCatalog *pCtg, SDBVgInfo *dbInfo, const SName *pTableName, SVgroupInfo *pVgroup) {
D
dapan1121 已提交
1073 1074
  int32_t code = 0;
  
D
dapan1121 已提交
1075
  int32_t vgNum = taosHashGetSize(dbInfo->vgHash);
H
Haojun Liao 已提交
1076 1077 1078
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);

1079
  if (vgNum <= 0) {
D
dapan1121 已提交
1080
    ctgError("db vgroup cache invalid, db:%s, vgroup number:%d", db, vgNum);
D
dapan1121 已提交
1081
    CTG_ERR_RET(TSDB_CODE_TSC_DB_NOT_SELECTED);
D
dapan1121 已提交
1082 1083
  }

1084 1085
  tableNameHashFp fp = NULL;
  SVgroupInfo *vgInfo = NULL;
D
dapan1121 已提交
1086

D
dapan1121 已提交
1087
  CTG_ERR_RET(ctgGetHashFunction(dbInfo->hashMethod, &fp));
1088 1089

  char tbFullName[TSDB_TABLE_FNAME_LEN];
H
Haojun Liao 已提交
1090
  tNameExtractFullName(pTableName, tbFullName);
1091 1092 1093

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

D
dapan1121 已提交
1094
  void *pIter = taosHashIterate(dbInfo->vgHash, NULL);
1095 1096 1097
  while (pIter) {
    vgInfo = pIter;
    if (hashValue >= vgInfo->hashBegin && hashValue <= vgInfo->hashEnd) {
D
dapan1121 已提交
1098
      taosHashCancelIterate(dbInfo->vgHash, pIter);
1099
      break;
D
dapan1121 已提交
1100
    }
1101
    
D
dapan1121 已提交
1102
    pIter = taosHashIterate(dbInfo->vgHash, pIter);
1103
    vgInfo = NULL;
D
dapan1121 已提交
1104 1105
  }

1106
  if (NULL == vgInfo) {
D
dapan1121 已提交
1107 1108
    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);
1109 1110 1111 1112
  }

  *pVgroup = *vgInfo;

1113
  CTG_RET(code);
D
dapan1121 已提交
1114 1115
}

D
dapan1121 已提交
1116
int32_t ctgStbVersionSearchCompare(const void* key1, const void* key2) {
D
dapan 已提交
1117
  if (*(uint64_t *)key1 < ((SSTableMetaVersion*)key2)->suid) {
D
dapan1121 已提交
1118
    return -1;
D
dapan 已提交
1119
  } else if (*(uint64_t *)key1 > ((SSTableMetaVersion*)key2)->suid) {
D
dapan1121 已提交
1120 1121 1122 1123 1124 1125
    return 1;
  } else {
    return 0;
  }
}

D
dapan1121 已提交
1126
int32_t ctgDbVgVersionSearchCompare(const void* key1, const void* key2) {
D
dapan1121 已提交
1127
  if (*(int64_t *)key1 < ((SDbVgVersion*)key2)->dbId) {
D
dapan1121 已提交
1128
    return -1;
D
dapan1121 已提交
1129
  } else if (*(int64_t *)key1 > ((SDbVgVersion*)key2)->dbId) {
D
dapan1121 已提交
1130 1131 1132
    return 1;
  } else {
    return 0;
D
dapan1121 已提交
1133
  }
D
dapan1121 已提交
1134 1135
}

D
dapan1121 已提交
1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156
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 已提交
1157
int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type) {
D
dapan1121 已提交
1158 1159 1160 1161
  mgmt->slotRIdx = 0;
  mgmt->slotNum = rentSec / CTG_RENT_SLOT_SECOND;
  mgmt->type = type;

D
dapan1121 已提交
1162
  size_t msgSize = sizeof(SCtgRentSlot) * mgmt->slotNum;
D
dapan1121 已提交
1163
  
wafwerar's avatar
wafwerar 已提交
1164
  mgmt->slots = taosMemoryCalloc(1, msgSize);
D
dapan1121 已提交
1165 1166
  if (NULL == mgmt->slots) {
    qError("calloc %d failed", (int32_t)msgSize);
D
dapan 已提交
1167
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1168
  }
D
dapan1121 已提交
1169

D
dapan1121 已提交
1170 1171 1172 1173
  qDebug("meta rent initialized, type:%d, slotNum:%d", type, mgmt->slotNum);
  
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
1174

D
dapan1121 已提交
1175

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

D
dapan1121 已提交
1179
  SCtgRentSlot *slot = &mgmt->slots[widx];
D
dapan1121 已提交
1180 1181 1182 1183 1184 1185 1186 1187
  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 已提交
1188
    }
D
dapan1121 已提交
1189
  }
D
dapan1121 已提交
1190

D
dapan1121 已提交
1191 1192 1193
  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 已提交
1194 1195
  }

D
dapan1121 已提交
1196
  slot->needSort = true;
D
dapan1121 已提交
1197

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

D
dapan1121 已提交
1200 1201 1202 1203 1204 1205
_return:

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

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

D
dapan1121 已提交
1209
  SCtgRentSlot *slot = &mgmt->slots[widx];
D
dapan1121 已提交
1210
  int32_t code = 0;
1211

D
dapan1121 已提交
1212 1213
  CTG_LOCK(CTG_WRITE, &slot->lock);
  if (NULL == slot->meta) {
D
dapan1121 已提交
1214 1215
    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 已提交
1216 1217 1218
  }

  if (slot->needSort) {
D
dapan1121 已提交
1219
    qDebug("meta slot before sorte, slot idx:%d, type:%d, size:%d", widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta));
D
dapan1121 已提交
1220
    taosArraySort(slot->meta, sortCompare);
D
dapan1121 已提交
1221
    slot->needSort = false;
D
dapan1121 已提交
1222
    qDebug("meta slot sorted, slot idx:%d, type:%d, size:%d", widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta));
D
dapan1121 已提交
1223 1224
  }

D
dapan1121 已提交
1225
  void *orig = taosArraySearch(slot->meta, &id, searchCompare, TD_EQ);
D
dapan1121 已提交
1226
  if (NULL == orig) {
D
dapan1121 已提交
1227
    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 已提交
1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246
    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 已提交
1247
int32_t ctgMetaRentRemove(SCtgRentMgmt *mgmt, int64_t id, __compar_fn_t sortCompare, __compar_fn_t searchCompare) {
1248
  int16_t widx = abs((int)(id % mgmt->slotNum));
D
dapan1121 已提交
1249

D
dapan1121 已提交
1250
  SCtgRentSlot *slot = &mgmt->slots[widx];
D
dapan1121 已提交
1251 1252 1253 1254 1255 1256 1257 1258 1259
  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 已提交
1260
    taosArraySort(slot->meta, sortCompare);
D
dapan1121 已提交
1261 1262 1263 1264
    slot->needSort = false;
    qDebug("meta slot sorted, slot idx:%d, type:%d", widx, mgmt->type);
  }

D
dapan1121 已提交
1265
  int32_t idx = taosArraySearchIdx(slot->meta, &id, searchCompare, TD_EQ);
D
dapan1121 已提交
1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282
  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 已提交
1283
int32_t ctgMetaRentGetImpl(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size) {
D
dapan1121 已提交
1284 1285 1286 1287
  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 已提交
1288
  }
D
dapan1121 已提交
1289

D
dapan1121 已提交
1290
  SCtgRentSlot *slot = &mgmt->slots[ridx];
D
dapan1121 已提交
1291
  int32_t code = 0;
D
dapan1121 已提交
1292
  
D
dapan1121 已提交
1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307
  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;
wafwerar's avatar
wafwerar 已提交
1308
  *res = taosMemoryMalloc(msize);
D
dapan1121 已提交
1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328
  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 已提交
1329
int32_t ctgMetaRentGet(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size) {
D
dapan1121 已提交
1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348
  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 已提交
1349 1350 1351
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
1352
int32_t ctgAddNewDBCache(SCatalog *pCtg, const char *dbFName, uint64_t dbId) {
D
dapan1121 已提交
1353
  int32_t code = 0;
D
dapan1121 已提交
1354

D
dapan1121 已提交
1355 1356 1357
  SCtgDBCache newDBCache = {0};
  newDBCache.dbId = dbId;

D
dapan 已提交
1358
  newDBCache.tbCache.metaCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
D
dapan1121 已提交
1359
  if (NULL == newDBCache.tbCache.metaCache) {
D
dapan 已提交
1360
    ctgError("taosHashInit %d metaCache failed", gCtgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
1361 1362 1363
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

D
dapan 已提交
1364
  newDBCache.tbCache.stbCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), true, HASH_ENTRY_LOCK);
D
dapan1121 已提交
1365
  if (NULL == newDBCache.tbCache.stbCache) {
D
dapan 已提交
1366
    ctgError("taosHashInit %d stbCache failed", gCtgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
1367 1368 1369 1370
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
  }

  code = taosHashPut(pCtg->dbCache, dbFName, strlen(dbFName), &newDBCache, sizeof(SCtgDBCache));
D
dapan1121 已提交
1371 1372 1373
  if (code) {
    if (HASH_NODE_EXIST(code)) {
      ctgDebug("db already in cache, dbFName:%s", dbFName);
D
dapan1121 已提交
1374
      goto _return;
D
dapan1121 已提交
1375 1376 1377
    }
    
    ctgError("taosHashPut db to cache failed, dbFName:%s", dbFName);
D
dapan1121 已提交
1378 1379
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
  }
D
dapan1121 已提交
1380
 
D
dapan1121 已提交
1381
  SDbVgVersion vgVersion = {.dbId = newDBCache.dbId, .vgVersion = -1};
D
dapan1121 已提交
1382 1383
  strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));

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

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

D
dapan1121 已提交
1390
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1391

D
dapan1121 已提交
1392
_return:
D
dapan1121 已提交
1393

D
dapan1121 已提交
1394
  ctgFreeDbCache(&newDBCache);
D
dapan1121 已提交
1395

D
dapan1121 已提交
1396 1397
  CTG_RET(code);
}
D
dapan1121 已提交
1398

D
dapan1121 已提交
1399

D
dapan1121 已提交
1400
void ctgRemoveStbRent(SCatalog* pCtg, SCtgTbMetaCache *cache) {
D
dapan1121 已提交
1401 1402 1403 1404 1405
  CTG_LOCK(CTG_WRITE, &cache->stbLock);
  if (cache->stbCache) {
    void *pIter = taosHashIterate(cache->stbCache, NULL);
    while (pIter) {
      uint64_t *suid = NULL;
H
Haojun Liao 已提交
1406
      suid = taosHashGetKey(pIter, NULL);
D
dapan1121 已提交
1407

D
dapan1121 已提交
1408
      if (TSDB_CODE_SUCCESS == ctgMetaRentRemove(&pCtg->stbRent, *suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare)) {
D
dapan1121 已提交
1409 1410 1411 1412 1413 1414 1415 1416 1417 1418
        ctgDebug("stb removed from rent, suid:%"PRIx64, *suid);
      }
          
      pIter = taosHashIterate(cache->stbCache, pIter);
    }
  }
  CTG_UNLOCK(CTG_WRITE, &cache->stbLock);
}


D
dapan1121 已提交
1419
int32_t ctgRemoveDB(SCatalog* pCtg, SCtgDBCache *dbCache, const char* dbFName) {
D
dapan 已提交
1420 1421 1422 1423
  uint64_t dbId = dbCache->dbId;
  
  ctgInfo("start to remove db from cache, dbFName:%s, dbId:%"PRIx64, dbFName, dbCache->dbId);

D
dapan1121 已提交
1424 1425 1426 1427 1428 1429
  atomic_store_8(&dbCache->deleted, 1);

  ctgRemoveStbRent(pCtg, &dbCache->tbCache);

  ctgFreeDbCache(dbCache);

D
dapan1121 已提交
1430
  CTG_ERR_RET(ctgMetaRentRemove(&pCtg->dbRent, dbCache->dbId, ctgDbVgVersionSortCompare, ctgDbVgVersionSearchCompare));
D
dapan1121 已提交
1431 1432 1433 1434
  
  ctgDebug("db removed from rent, dbFName:%s, dbId:%"PRIx64, dbFName, dbCache->dbId);

  if (taosHashRemove(pCtg->dbCache, dbFName, strlen(dbFName))) {
D
dapan1121 已提交
1435 1436
    ctgInfo("taosHashRemove from dbCache failed, may be removed, dbFName:%s", dbFName);
    CTG_ERR_RET(TSDB_CODE_CTG_DB_DROPPED);
D
dapan1121 已提交
1437
  }
D
dapan 已提交
1438 1439

  ctgInfo("db removed from cache, dbFName:%s, dbId:%"PRIx64, dbFName, dbId);
D
dapan1121 已提交
1440 1441 1442
  
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
1443 1444


D
dapan1121 已提交
1445
int32_t ctgGetAddDBCache(SCatalog* pCtg, const char *dbFName, uint64_t dbId, SCtgDBCache **pCache) {
D
dapan1121 已提交
1446 1447
  int32_t code = 0;
  SCtgDBCache *dbCache = NULL;
D
dapan1121 已提交
1448
  ctgGetDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
1449
  
D
dapan1121 已提交
1450 1451
  if (dbCache) {
  // TODO OPEN IT
D
dapan1121 已提交
1452
#if 0    
D
dapan1121 已提交
1453 1454 1455 1456
    if (dbCache->dbId == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
D
dapan1121 已提交
1457
#else
D
dapan1121 已提交
1458 1459 1460
    if (0 == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1461 1462
    }

D
dapan1121 已提交
1463 1464 1465 1466 1467
    if (dbId && (dbCache->dbId == 0)) {
      dbCache->dbId = dbId;
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
D
dapan1121 已提交
1468
    
D
dapan1121 已提交
1469 1470 1471 1472 1473 1474
    if (dbCache->dbId == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
#endif
    CTG_ERR_RET(ctgRemoveDB(pCtg, dbCache, dbFName));
D
dapan1121 已提交
1475
  }
D
dapan1121 已提交
1476 1477
  
  CTG_ERR_RET(ctgAddNewDBCache(pCtg, dbFName, dbId));
D
dapan1121 已提交
1478

D
dapan1121 已提交
1479
  ctgGetDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
1480

D
dapan1121 已提交
1481
  *pCache = dbCache;
D
dapan1121 已提交
1482

D
dapan1121 已提交
1483
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1484 1485 1486
}


D
dapan1121 已提交
1487 1488 1489
int32_t ctgUpdateDBVgInfo(SCatalog* pCtg, const char* dbFName, uint64_t dbId, SDBVgInfo** pDbInfo) {
  int32_t code = 0;
  SDBVgInfo* dbInfo = *pDbInfo;
D
dapan1121 已提交
1490 1491 1492 1493

  if (NULL == dbInfo->vgHash) {
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
1494
  
D
dapan1121 已提交
1495
  if (dbInfo->vgVersion < 0 || taosHashGetSize(dbInfo->vgHash) <= 0) {
D
dapan1121 已提交
1496 1497
    ctgError("invalid db vgInfo, dbFName:%s, vgHash:%p, vgVersion:%d, vgHashSize:%d", 
      dbFName, dbInfo->vgHash, dbInfo->vgVersion, taosHashGetSize(dbInfo->vgHash));
D
dapan1121 已提交
1498
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1499 1500
  }

D
dapan1121 已提交
1501
  bool newAdded = false;
D
dapan 已提交
1502
  SDbVgVersion vgVersion = {.dbId = dbId, .vgVersion = dbInfo->vgVersion, .numOfTable = dbInfo->numOfTable};
D
dapan1121 已提交
1503 1504 1505 1506 1507 1508 1509 1510 1511 1512

  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 已提交
1513
  
D
dapan1121 已提交
1514
  if (dbCache->vgInfo) {
D
dapan 已提交
1515 1516 1517 1518 1519 1520 1521 1522 1523
    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 已提交
1524
      ctgWReleaseVgInfo(dbCache);
D
dapan1121 已提交
1525
      
D
dapan1121 已提交
1526
      return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1527
    }
D
dapan1121 已提交
1528 1529

    ctgFreeVgInfo(dbCache->vgInfo);
D
dapan1121 已提交
1530 1531
  }

D
dapan1121 已提交
1532
  dbCache->vgInfo = dbInfo;
D
dapan1121 已提交
1533

D
dapan1121 已提交
1534
  *pDbInfo = NULL;
D
dapan1121 已提交
1535

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

D
dapan1121 已提交
1538
  ctgWReleaseVgInfo(dbCache);
D
dapan1121 已提交
1539

D
dapan1121 已提交
1540
  dbCache = NULL;
D
dapan1121 已提交
1541

D
dapan1121 已提交
1542
  strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));
D
dapan1121 已提交
1543
  CTG_ERR_RET(ctgMetaRentUpdate(&pCtg->dbRent, &vgVersion, vgVersion.dbId, sizeof(SDbVgVersion), ctgDbVgVersionSortCompare, ctgDbVgVersionSearchCompare));
D
dapan1121 已提交
1544
  
D
dapan1121 已提交
1545 1546 1547 1548
  CTG_RET(code);
}


D
dapan1121 已提交
1549 1550
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 已提交
1551

D
dapan1121 已提交
1552 1553 1554 1555 1556
  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 已提交
1557 1558
  }

D
dapan1121 已提交
1559 1560 1561 1562 1563 1564 1565
  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 已提交
1566 1567 1568 1569 1570 1571 1572 1573 1574 1575
    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 已提交
1576
        ctgMetaRentRemove(&pCtg->stbRent, orig->suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare);
D
dapan1121 已提交
1577
      }
D
dapan1121 已提交
1578

D
dapan1121 已提交
1579
      origSuid = orig->suid;
D
dapan1121 已提交
1580
    }
D
dapan1121 已提交
1581
  }
D
dapan1121 已提交
1582

D
dapan1121 已提交
1583 1584
  if (isStb) {
    CTG_LOCK(CTG_WRITE, &tbCache->stbLock);
D
dapan1121 已提交
1585
  }
D
dapan1121 已提交
1586
  
D
dapan1121 已提交
1587 1588 1589 1590 1591 1592 1593 1594 1595
  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 已提交
1596

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

D
dapan1121 已提交
1600 1601 1602
  if (!isStb) {
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);  
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1603
  }
D
dapan1121 已提交
1604

D
dapan1121 已提交
1605
  if (origType == TSDB_SUPER_TABLE && origSuid == meta->suid) {
D
dapan1121 已提交
1606 1607 1608 1609
    CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);  
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
1610

D
dapan1121 已提交
1611 1612 1613 1614
  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);    
H
Haojun Liao 已提交
1615
    ctgError("taosHashPut stable to stable cache failed, suid:%"PRIx64, meta->suid);
D
dapan1121 已提交
1616
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1617
  }
D
dapan1121 已提交
1618 1619
  
  CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);
D
dapan1121 已提交
1620

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

D
dapan 已提交
1623
  ctgDebug("stb updated to stbCache, dbFName:%s, tbName:%s, tbType:%d", dbFName, tbName, meta->tableType);
D
dapan1121 已提交
1624 1625 1626 1627 1628 1629 1630

  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 已提交
1631 1632
}

D
dapan 已提交
1633
int32_t ctgCloneVgInfo(SDBVgInfo *src, SDBVgInfo **dst) {
wafwerar's avatar
wafwerar 已提交
1634
  *dst = taosMemoryMalloc(sizeof(SDBVgInfo));
D
dapan 已提交
1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645
  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);
wafwerar's avatar
wafwerar 已提交
1646
    taosMemoryFreeClear(*dst);
D
dapan 已提交
1647 1648 1649 1650 1651 1652
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

  int32_t *vgId = NULL;
  void *pIter = taosHashIterate(src->vgHash, NULL);
  while (pIter) {
H
Haojun Liao 已提交
1653
    vgId = taosHashGetKey(pIter, NULL);
D
dapan 已提交
1654 1655 1656 1657 1658

    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);
wafwerar's avatar
wafwerar 已提交
1659
      taosMemoryFreeClear(*dst);
D
dapan 已提交
1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }
    
    pIter = taosHashIterate(src->vgHash, pIter);
  }


  return TSDB_CODE_SUCCESS;
}



D
dapan1121 已提交
1672
int32_t ctgGetDBVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* dbFName, SCtgDBCache** dbCache, SDBVgInfo **pInfo) {
D
dapan1121 已提交
1673
  bool inCache = false;
D
dapan1121 已提交
1674
  int32_t code = 0;
D
dapan1121 已提交
1675 1676 1677

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

D
dapan1121 已提交
1678
  if (inCache) {
D
dapan1121 已提交
1679
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1680 1681 1682 1683 1684
  }

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

D
dapan1121 已提交
1685
  tstrncpy(input.db, dbFName, tListLen(input.db));
D
dapan1121 已提交
1686
  input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
H
Haojun Liao 已提交
1687

D
dapan1121 已提交
1688 1689 1690 1691 1692 1693 1694 1695 1696
  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 已提交
1697

D
dapan 已提交
1698
  CTG_ERR_JRET(ctgCloneVgInfo(DbOut.dbVgroup, pInfo));
D
dapan1121 已提交
1699

D
dapan1121 已提交
1700
  CTG_ERR_RET(ctgPushUpdateVgMsgInQueue(pCtg, dbFName, DbOut.dbId, DbOut.dbVgroup, false));
D
dapan 已提交
1701

D
dapan1121 已提交
1702
  return TSDB_CODE_SUCCESS;
D
dapan 已提交
1703 1704 1705

_return:

wafwerar's avatar
wafwerar 已提交
1706
  taosMemoryFreeClear(*pInfo);
D
dapan 已提交
1707 1708 1709
  *pInfo = DbOut.dbVgroup;
  
  CTG_RET(code);
D
dapan1121 已提交
1710 1711
}

D
dapan1121 已提交
1712 1713 1714
int32_t ctgRefreshDBVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* dbFName) {
  bool inCache = false;
  int32_t code = 0;
D
dapan1121 已提交
1715
  SCtgDBCache* dbCache = NULL;
D
dapan1121 已提交
1716

D
dapan1121 已提交
1717
  CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache, &inCache));
D
dapan1121 已提交
1718 1719 1720 1721 1722 1723

  SUseDbOutput DbOut = {0};
  SBuildUseDBInput input = {0};
  tstrncpy(input.db, dbFName, tListLen(input.db));

  if (inCache) {
D
dapan1121 已提交
1724
    input.dbId = dbCache->dbId;
D
dapan1121 已提交
1725

D
dapan1121 已提交
1726 1727
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
1728
  }
D
dapan1121 已提交
1729 1730 1731
  
  input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
  input.numOfTable = 0;
D
dapan1121 已提交
1732 1733 1734

  code = ctgGetDBVgInfoFromMnode(pCtg, pRpc, pMgmtEps, &input, &DbOut);
  if (code) {
D
dapan1121 已提交
1735
    if (CTG_DB_NOT_EXIST(code) && inCache) {
D
dapan1121 已提交
1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748
      ctgDebug("db no longer exist, dbFName:%s, dbId:%" PRIx64, input.db, input.dbId);
      ctgPushRmDBMsgInQueue(pCtg, input.db, input.dbId);
    }

    CTG_ERR_RET(code);
  }

  CTG_ERR_RET(ctgPushUpdateVgMsgInQueue(pCtg, dbFName, DbOut.dbId, DbOut.dbVgroup, true));

  return TSDB_CODE_SUCCESS;
}


D
dapan 已提交
1749

D
dapan1121 已提交
1750
int32_t ctgCloneMetaOutput(STableMetaOutput *output, STableMetaOutput **pOutput) {
wafwerar's avatar
wafwerar 已提交
1751
  *pOutput = taosMemoryMalloc(sizeof(STableMetaOutput));
D
dapan1121 已提交
1752 1753 1754
  if (NULL == *pOutput) {
    qError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan 已提交
1755 1756
  }

D
dapan1121 已提交
1757 1758 1759 1760
  memcpy(*pOutput, output, sizeof(STableMetaOutput));

  if (output->tbMeta) {
    int32_t metaSize = CTG_META_SIZE(output->tbMeta);
wafwerar's avatar
wafwerar 已提交
1761
    (*pOutput)->tbMeta = taosMemoryMalloc(metaSize);
D
dapan1121 已提交
1762
    if (NULL == (*pOutput)->tbMeta) {
D
dapan 已提交
1763
      qError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
wafwerar's avatar
wafwerar 已提交
1764
      taosMemoryFreeClear(*pOutput);
D
dapan1121 已提交
1765 1766 1767 1768
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }

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

D
dapan1121 已提交
1771 1772 1773
  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
1774 1775


D
dapan1121 已提交
1776
int32_t ctgRefreshTblMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, int32_t flag, STableMetaOutput **pOutput, bool syncReq) {
D
dapan1121 已提交
1777
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pTableName) {
D
dapan1121 已提交
1778 1779 1780 1781 1782 1783
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }

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

D
dapan1121 已提交
1784
  if (!CTG_FLAG_IS_INF_DB(flag)) {
D
dapan1121 已提交
1785 1786
    CTG_ERR_RET(catalogGetTableHashVgroup(pCtg, pTrans, pMgmtEps, pTableName, &vgroupInfo));
  }
D
dapan1121 已提交
1787

D
dapan1121 已提交
1788
  STableMetaOutput  moutput = {0};
wafwerar's avatar
wafwerar 已提交
1789
  STableMetaOutput *output = taosMemoryCalloc(1, sizeof(STableMetaOutput));
D
dapan1121 已提交
1790 1791 1792 1793
  if (NULL == output) {
    ctgError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }
D
dapan1121 已提交
1794

D
dapan1121 已提交
1795
  if (CTG_FLAG_IS_INF_DB(flag)) {
D
dapan1121 已提交
1796 1797 1798
    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 已提交
1799
  } else if (CTG_FLAG_IS_STB(flag)) {
D
dapan 已提交
1800
    ctgDebug("will refresh tbmeta, supposed to be stb, tbName:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
1801 1802

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

D
dapan1121 已提交
1805
    if (CTG_IS_META_NULL(output->metaType)) {
D
dapan1121 已提交
1806
      CTG_ERR_JRET(ctgGetTableMetaFromVnode(pCtg, pTrans, pMgmtEps, pTableName, &vgroupInfo, output));
D
dapan1121 已提交
1807 1808
    }
  } else {
D
dapan1121 已提交
1809
    ctgDebug("will refresh tbmeta, not supposed to be stb, tbName:%s, flag:%d", tNameGetTableName(pTableName), flag);
D
dapan1121 已提交
1810 1811

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

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

wafwerar's avatar
wafwerar 已提交
1817
      taosMemoryFreeClear(output->tbMeta);
D
dapan1121 已提交
1818
      
D
dapan1121 已提交
1819
      CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, output->dbFName, output->tbName, output));
D
dapan1121 已提交
1820
    } else if (CTG_IS_META_BOTH(output->metaType)) {
D
dapan1121 已提交
1821
      int32_t exist = 0;
D
dapan1121 已提交
1822 1823 1824
      if (!CTG_FLAG_IS_FORCE_UPDATE(flag)) {
        CTG_ERR_JRET(ctgIsTableMetaExistInCache(pCtg, output->dbFName, output->tbName, &exist));
      }
H
Haojun Liao 已提交
1825

D
dapan1121 已提交
1826
      if (0 == exist) {
D
dapan1121 已提交
1827
        CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, output->dbFName, output->tbName, &moutput));
D
dapan1121 已提交
1828

D
dapan1121 已提交
1829
        if (CTG_IS_META_NULL(moutput.metaType)) {
D
dapan1121 已提交
1830
          SET_META_TYPE_NULL(output->metaType);
D
dapan1121 已提交
1831 1832
        }
        
wafwerar's avatar
wafwerar 已提交
1833
        taosMemoryFreeClear(output->tbMeta);
D
dapan1121 已提交
1834
        output->tbMeta = moutput.tbMeta;
D
dapan1121 已提交
1835 1836
        moutput.tbMeta = NULL;
      } else {
wafwerar's avatar
wafwerar 已提交
1837
        taosMemoryFreeClear(output->tbMeta);
D
dapan1121 已提交
1838
        
D
dapan1121 已提交
1839
        SET_META_TYPE_CTABLE(output->metaType); 
D
dapan1121 已提交
1840
      }
D
dapan1121 已提交
1841 1842 1843
    }
  }

D
dapan1121 已提交
1844
  if (CTG_IS_META_NULL(output->metaType)) {
D
dapan 已提交
1845
    ctgError("no tbmeta got, tbNmae:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
1846 1847 1848
    CTG_ERR_JRET(CTG_ERR_CODE_TABLE_NOT_EXIST);
  }

D
dapan 已提交
1849 1850 1851 1852 1853 1854
  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 已提交
1855 1856 1857 1858
  if (pOutput) {
    CTG_ERR_JRET(ctgCloneMetaOutput(output, pOutput));
  }

D
dapan1121 已提交
1859
  CTG_ERR_JRET(ctgPushUpdateTblMsgInQueue(pCtg, output, syncReq));
D
dapan 已提交
1860

D
dapan1121 已提交
1861
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1862 1863 1864

_return:

wafwerar's avatar
wafwerar 已提交
1865 1866
  taosMemoryFreeClear(output->tbMeta);
  taosMemoryFreeClear(output);
D
dapan1121 已提交
1867 1868 1869 1870
  
  CTG_RET(code);
}

D
dapan1121 已提交
1871
int32_t ctgGetTableMeta(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t flag) {
D
dapan1121 已提交
1872
  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == pTableMeta) {
D
dapan1121 已提交
1873 1874 1875 1876
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }
  
  int32_t exist = 0;
D
dapan1121 已提交
1877
  int32_t code = 0;
D
dapan1121 已提交
1878 1879 1880
  uint64_t dbId = 0;
  uint64_t suid = 0;
  STableMetaOutput *output = NULL;
D
dapan1121 已提交
1881

D
dapan1121 已提交
1882
  if (CTG_IS_INF_DBNAME(pTableName->dbname)) {
D
dapan1121 已提交
1883
    CTG_FLAG_SET_INF_DB(flag);
D
dapan1121 已提交
1884 1885
  }

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

D
dapan1121 已提交
1888 1889 1890 1891 1892
  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 已提交
1893
    }
D
dapan1121 已提交
1894

D
dapan1121 已提交
1895 1896
    tbType = (*pTableMeta)->tableType;
    suid = (*pTableMeta)->suid;
D
dapan1121 已提交
1897

wafwerar's avatar
wafwerar 已提交
1898
    taosMemoryFreeClear(*pTableMeta);
D
dapan1121 已提交
1899
  }
D
dapan1121 已提交
1900

D
dapan1121 已提交
1901 1902
  if (CTG_FLAG_IS_UNKNOWN_STB(flag)) {
    CTG_FLAG_SET_STB(flag, tbType);
D
dapan1121 已提交
1903 1904
  }

D
dapan1121 已提交
1905

D
dapan 已提交
1906
  while (true) {
D
dapan1121 已提交
1907
    CTG_ERR_JRET(ctgRefreshTblMeta(pCtg, pRpc, pMgmtEps, pTableName, flag, &output, false));
D
dapan1121 已提交
1908

D
dapan 已提交
1909 1910 1911 1912
    if (CTG_IS_META_TABLE(output->metaType)) {
      *pTableMeta = output->tbMeta;
      goto _return;
    }
D
dapan1121 已提交
1913

D
dapan 已提交
1914 1915 1916 1917 1918 1919
    if (CTG_IS_META_BOTH(output->metaType)) {
      memcpy(output->tbMeta, &output->ctbMeta, sizeof(output->ctbMeta));
      
      *pTableMeta = output->tbMeta;
      goto _return;
    }
D
dapan1121 已提交
1920

D
dapan 已提交
1921 1922
    if ((!CTG_IS_META_CTABLE(output->metaType)) || output->tbMeta) {
      ctgError("invalid metaType:%d", output->metaType);
wafwerar's avatar
wafwerar 已提交
1923
      taosMemoryFreeClear(output->tbMeta);
D
dapan 已提交
1924 1925
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
    }
D
dapan1121 已提交
1926

D
dapan 已提交
1927
    // HANDLE ONLY CHILD TABLE META
D
dapan1121 已提交
1928

D
dapan 已提交
1929 1930 1931
    SName stbName = *pTableName;
    strcpy(stbName.tname, output->tbName);
    
D
dapan1121 已提交
1932
    CTG_ERR_JRET(ctgGetTableMetaFromCache(pCtg, &stbName, pTableMeta, &exist, flag, NULL));
D
dapan 已提交
1933 1934 1935 1936
    if (0 == exist) {
      ctgDebug("stb no longer exist, dbFName:%s, tbName:%s", output->dbFName, pTableName->tname);
      continue;
    }
D
dapan1121 已提交
1937

D
dapan 已提交
1938 1939 1940 1941
    memcpy(*pTableMeta, &output->ctbMeta, sizeof(output->ctbMeta));

    break;
  }
D
dapan1121 已提交
1942 1943 1944

_return:

D
dapan1121 已提交
1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959
  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);
    }
  }

wafwerar's avatar
wafwerar 已提交
1960
  taosMemoryFreeClear(output);
D
dapan1121 已提交
1961

D
dapan 已提交
1962 1963
  if (*pTableMeta) {
    ctgDebug("tbmeta returned, tbName:%s, tbType:%d", pTableName->tname, (*pTableMeta)->tableType);
D
dapan1121 已提交
1964
    ctgDbgShowTableMeta(pCtg, pTableName->tname, *pTableMeta);
D
dapan 已提交
1965 1966
  }

D
dapan1121 已提交
1967 1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979
  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 已提交
1980
  ctgFreeVgInfo(msg->dbInfo);
wafwerar's avatar
wafwerar 已提交
1981
  taosMemoryFreeClear(msg);
D
dapan1121 已提交
1982 1983 1984 1985 1986 1987 1988 1989 1990 1991
  
  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 已提交
1992
  ctgGetDBCache(msg->pCtg, msg->dbFName, &dbCache);
D
dapan1121 已提交
1993 1994 1995 1996 1997 1998 1999 2000 2001 2002 2003 2004 2005
  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:

wafwerar's avatar
wafwerar 已提交
2006
  taosMemoryFreeClear(msg);
D
dapan1121 已提交
2007 2008 2009 2010 2011 2012 2013 2014 2015 2016 2017 2018 2019 2020
  
  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 已提交
2021
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
2022 2023 2024 2025 2026 2027
  }

  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 已提交
2028
  
D
dapan1121 已提交
2029 2030 2031 2032 2033 2034 2035 2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046
  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 已提交
2047
  if (output) {
wafwerar's avatar
wafwerar 已提交
2048 2049
    taosMemoryFreeClear(output->tbMeta);
    taosMemoryFreeClear(output);
D
dapan1121 已提交
2050
  }
D
dapan 已提交
2051
  
wafwerar's avatar
wafwerar 已提交
2052
  taosMemoryFreeClear(msg);
D
dapan1121 已提交
2053 2054 2055 2056 2057 2058 2059 2060 2061 2062 2063 2064 2065 2066 2067 2068
  
  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;
  }

D
dapan 已提交
2069
  if (msg->dbId && (dbCache->dbId != msg->dbId)) {
D
dapan1121 已提交
2070 2071 2072 2073 2074 2075 2076 2077 2078 2079 2080 2081 2082 2083 2084 2085 2086 2087 2088
    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))) {
    ctgDebug("stb not exist in stbCache, may be removed, dbFName:%s, stb:%s, suid:%"PRIx64, msg->dbFName, msg->stbName, msg->suid);
  }

  CTG_LOCK(CTG_READ, &dbCache->tbCache.metaLock);
  if (taosHashRemove(dbCache->tbCache.metaCache, msg->stbName, strlen(msg->stbName))) {  
    ctgError("stb not exist in cache, dbFName:%s, stb:%s, suid:%"PRIx64, msg->dbFName, msg->stbName, msg->suid);
  }  
  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 已提交
2089
  CTG_ERR_JRET(ctgMetaRentRemove(&msg->pCtg->stbRent, msg->suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare));
D
dapan1121 已提交
2090 2091 2092 2093 2094
  
  ctgDebug("stb removed from rent, dbFName:%s, stbName:%s, suid:%"PRIx64, msg->dbFName, msg->stbName, msg->suid);
  
_return:

wafwerar's avatar
wafwerar 已提交
2095
  taosMemoryFreeClear(msg);
D
dapan1121 已提交
2096 2097 2098 2099 2100
  
  CTG_RET(code);
}

int32_t ctgActRemoveTbl(SCtgMetaAction *action) {
D
dapan1121 已提交
2101 2102 2103 2104 2105 2106 2107 2108 2109 2110 2111 2112 2113 2114
  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;
  }
H
Haojun Liao 已提交
2115

D
dapan1121 已提交
2116
  CTG_LOCK(CTG_READ, &dbCache->tbCache.metaLock);
H
Haojun Liao 已提交
2117
  if (taosHashRemove(dbCache->tbCache.metaCache, msg->tbName, strlen(msg->tbName))) {
D
dapan1121 已提交
2118 2119 2120
    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);
H
Haojun Liao 已提交
2121
  }
D
dapan1121 已提交
2122
  CTG_UNLOCK(CTG_READ, &dbCache->tbCache.metaLock);
D
dapan1121 已提交
2123

D
dapan1121 已提交
2124
  ctgInfo("table removed from cache, dbFName:%s, tbName:%s", msg->dbFName, msg->tbName);
H
Haojun Liao 已提交
2125

D
dapan1121 已提交
2126
_return:
D
dapan1121 已提交
2127

wafwerar's avatar
wafwerar 已提交
2128
  taosMemoryFreeClear(msg);
H
Haojun Liao 已提交
2129

D
dapan1121 已提交
2130
  CTG_RET(code);
D
dapan1121 已提交
2131 2132 2133 2134 2135 2136 2137 2138
}


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

  qInfo("catalog update thread started");

D
dapan 已提交
2139
  CTG_LOCK(CTG_READ, &gCtgMgmt.lock);
D
dapan1121 已提交
2140 2141
  
  while (true) {
D
dapan1121 已提交
2142
    tsem_wait(&gCtgMgmt.queue.reqSem);
D
dapan1121 已提交
2143
    
wafwerar's avatar
wafwerar 已提交
2144
    if (atomic_load_8((int8_t*)&gCtgMgmt.exit)) {
D
dapan1121 已提交
2145
      tsem_post(&gCtgMgmt.queue.rspSem);
D
dapan1121 已提交
2146 2147 2148 2149 2150
      break;
    }

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

D
dapan1121 已提交
2153
    ctgDebug("process [%s] action", gCtgAction[action->act].name);
D
dapan 已提交
2154 2155 2156
    
    (*gCtgAction[action->act].func)(action);

D
dapan1121 已提交
2157 2158 2159 2160 2161 2162
    gCtgMgmt.queue.seqDone = action->seqId;

    if (action->syncReq) {
      tsem_post(&gCtgMgmt.queue.rspSem);
    }

D
dapan 已提交
2163
    CTG_STAT_ADD(gCtgMgmt.stat.runtime.qDoneNum); 
D
dapan1121 已提交
2164 2165

    ctgDbgShowClusterCache(pCtg);
D
dapan1121 已提交
2166 2167
  }

D
dapan 已提交
2168
  CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock);
D
dapan1121 已提交
2169 2170 2171 2172 2173 2174 2175 2176

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


int32_t ctgStartUpdateThread() {
wafwerar's avatar
wafwerar 已提交
2177 2178 2179
  TdThreadAttr thAttr;
  taosThreadAttrInit(&thAttr);
  taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
D
dapan1121 已提交
2180

wafwerar's avatar
wafwerar 已提交
2181
  if (taosThreadCreate(&gCtgMgmt.updateThread, &thAttr, ctgUpdateThreadFunc, NULL) != 0) {
D
dapan1121 已提交
2182 2183
    terrno = TAOS_SYSTEM_ERROR(errno);
    CTG_ERR_RET(terrno);
D
dapan1121 已提交
2184 2185
  }
  
wafwerar's avatar
wafwerar 已提交
2186
  taosThreadAttrDestroy(&thAttr);
D
dapan1121 已提交
2187 2188 2189
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
2190 2191 2192 2193 2194 2195 2196 2197 2198 2199 2200 2201 2202 2203 2204 2205
int32_t ctgGetTableDistVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgList) {
  STableMeta *tbMeta = NULL;
  int32_t code = 0;
  SVgroupInfo vgroupInfo = {0};
  SCtgDBCache* dbCache = NULL;
  SArray *vgList = NULL;
  SDBVgInfo *vgInfo = NULL;

  *pVgList = NULL;
  
  CTG_ERR_JRET(ctgGetTableMeta(pCtg, pRpc, pMgmtEps, pTableName, &tbMeta, CTG_FLAG_UNKNOWN_STB));

  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);

  SHashObj *vgHash = NULL;  
D
dapan1121 已提交
2206
  CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pRpc, pMgmtEps, db, &dbCache, &vgInfo));
D
dapan1121 已提交
2207 2208 2209 2210 2211 2212 2213 2214 2215 2216 2217 2218 2219 2220 2221 2222 2223 2224 2225 2226 2227 2228 2229 2230 2231 2232 2233 2234 2235 2236 2237 2238 2239 2240 2241 2242 2243 2244 2245 2246 2247 2248 2249 2250

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

  if (tbMeta->tableType == TSDB_SUPER_TABLE) {
    CTG_ERR_JRET(ctgGenerateVgList(pCtg, vgHash, pVgList));
  } else {
    // USE HASH METHOD INSTEAD OF VGID IN TBMETA
    ctgError("invalid method to get none stb vgInfo, tbType:%d", tbMeta->tableType);
    CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
    
#if 0  
    int32_t vgId = tbMeta->vgId;
    if (taosHashGetDup(vgHash, &vgId, sizeof(vgId), &vgroupInfo) != 0) {
      ctgWarn("table's vgId not found in vgroup list, vgId:%d, tbName:%s", vgId, tNameGetTableName(pTableName));
      CTG_ERR_JRET(TSDB_CODE_CTG_VG_META_MISMATCH);
    }

    vgList = taosArrayInit(1, sizeof(SVgroupInfo));
    if (NULL == vgList) {
      ctgError("taosArrayInit %d failed", (int32_t)sizeof(SVgroupInfo));
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);    
    }

    if (NULL == taosArrayPush(vgList, &vgroupInfo)) {
      ctgError("taosArrayPush vgroupInfo to array failed, vgId:%d, tbName:%s", vgId, tNameGetTableName(pTableName));
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
    }

    *pVgList = vgList;
    vgList = NULL;
#endif    
  }

_return:

  if (dbCache) {
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
  }

wafwerar's avatar
wafwerar 已提交
2251
  taosMemoryFreeClear(tbMeta);
D
dapan1121 已提交
2252 2253 2254

  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
wafwerar's avatar
wafwerar 已提交
2255
    taosMemoryFreeClear(vgInfo);
D
dapan1121 已提交
2256 2257 2258 2259 2260 2261 2262 2263 2264 2265
  }

  if (vgList) {
    taosArrayDestroy(vgList);
    vgList = NULL;
  }

  CTG_RET(code);
}

D
dapan1121 已提交
2266

D
dapan1121 已提交
2267
int32_t catalogInit(SCatalogCfg *cfg) {
D
dapan 已提交
2268
  if (gCtgMgmt.pCluster) {
D
dapan 已提交
2269
    qError("catalog already initialized");
D
dapan1121 已提交
2270
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2271 2272
  }

wafwerar's avatar
wafwerar 已提交
2273
  atomic_store_8((int8_t*)&gCtgMgmt.exit, false);
D
dapan1121 已提交
2274

D
dapan1121 已提交
2275
  if (cfg) {
D
dapan 已提交
2276
    memcpy(&gCtgMgmt.cfg, cfg, sizeof(*cfg));
H
Haojun Liao 已提交
2277

D
dapan 已提交
2278 2279
    if (gCtgMgmt.cfg.maxDBCacheNum == 0) {
      gCtgMgmt.cfg.maxDBCacheNum = CTG_DEFAULT_CACHE_DB_NUMBER;
D
dapan1121 已提交
2280 2281
    }

D
dapan 已提交
2282 2283
    if (gCtgMgmt.cfg.maxTblCacheNum == 0) {
      gCtgMgmt.cfg.maxTblCacheNum = CTG_DEFAULT_CACHE_TBLMETA_NUMBER;
D
dapan1121 已提交
2284
    }
D
dapan1121 已提交
2285

D
dapan 已提交
2286 2287
    if (gCtgMgmt.cfg.dbRentSec == 0) {
      gCtgMgmt.cfg.dbRentSec = CTG_DEFAULT_RENT_SECOND;
D
dapan1121 已提交
2288 2289
    }

D
dapan 已提交
2290 2291
    if (gCtgMgmt.cfg.stbRentSec == 0) {
      gCtgMgmt.cfg.stbRentSec = CTG_DEFAULT_RENT_SECOND;
D
dapan1121 已提交
2292
    }
D
dapan1121 已提交
2293
  } else {
D
dapan 已提交
2294 2295 2296 2297
    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 已提交
2298 2299
  }

D
dapan 已提交
2300 2301
  gCtgMgmt.pCluster = taosHashInit(CTG_DEFAULT_CACHE_CLUSTER_NUMBER, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
  if (NULL == gCtgMgmt.pCluster) {
D
dapan1121 已提交
2302 2303
    qError("taosHashInit %d cluster cache failed", CTG_DEFAULT_CACHE_CLUSTER_NUMBER);
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
2304 2305
  }

D
dapan1121 已提交
2306 2307
  CTG_ERR_RET(ctgStartUpdateThread());

D
dapan1121 已提交
2308 2309
  tsem_init(&gCtgMgmt.queue.reqSem, 0, 0);
  tsem_init(&gCtgMgmt.queue.rspSem, 0, 0);
D
dapan1121 已提交
2310

wafwerar's avatar
wafwerar 已提交
2311
  gCtgMgmt.queue.head = taosMemoryCalloc(1, sizeof(SCtgQNode));
D
dapan1121 已提交
2312
  if (NULL == gCtgMgmt.queue.head) {
D
dapan1121 已提交
2313 2314 2315
    qError("calloc %d failed", (int32_t)sizeof(SCtgQNode));
    CTG_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }
D
dapan1121 已提交
2316
  gCtgMgmt.queue.tail = gCtgMgmt.queue.head;
D
dapan1121 已提交
2317

D
dapan 已提交
2318
  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 已提交
2319

D
dapan 已提交
2320
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
2321 2322
}

D
dapan1121 已提交
2323
int32_t catalogGetHandle(uint64_t clusterId, SCatalog** catalogHandle) {
2324
  if (NULL == catalogHandle) {
D
dapan1121 已提交
2325
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
2326 2327
  }

D
dapan 已提交
2328
  if (NULL == gCtgMgmt.pCluster) {
D
dapan 已提交
2329
    qError("catalog cluster cache are not ready, clusterId:%"PRIx64, clusterId);
D
dapan1121 已提交
2330
    CTG_ERR_RET(TSDB_CODE_CTG_NOT_READY);
D
dapan 已提交
2331 2332
  }

D
dapan1121 已提交
2333 2334
  int32_t code = 0;
  SCatalog *clusterCtg = NULL;
D
dapan 已提交
2335

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

D
dapan1121 已提交
2339 2340 2341 2342 2343
    if (ctg && (*ctg)) {
      *catalogHandle = *ctg;
      qDebug("got catalog handle from cache, clusterId:%"PRIx64", CTG:%p", clusterId, *ctg);
      return TSDB_CODE_SUCCESS;
    }
D
dapan 已提交
2344

wafwerar's avatar
wafwerar 已提交
2345
    clusterCtg = taosMemoryCalloc(1, sizeof(SCatalog));
D
dapan1121 已提交
2346 2347 2348 2349 2350
    if (NULL == clusterCtg) {
      qError("calloc %d failed", (int32_t)sizeof(SCatalog));
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }

D
dapan1121 已提交
2351 2352
    clusterCtg->clusterId = clusterId;

D
dapan 已提交
2353 2354
    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 已提交
2355

D
dapan 已提交
2356
    clusterCtg->dbCache = taosHashInit(gCtgMgmt.cfg.maxDBCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
D
dapan1121 已提交
2357 2358 2359 2360 2361
    if (NULL == clusterCtg->dbCache) {
      qError("taosHashInit %d dbCache failed", CTG_DEFAULT_CACHE_DB_NUMBER);
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
    }

D
dapan 已提交
2362
    SHashObj *metaCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
D
dapan1121 已提交
2363
    if (NULL == metaCache) {
D
dapan 已提交
2364
      qError("taosHashInit failed, num:%d", gCtgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
2365 2366 2367
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }
    
D
dapan 已提交
2368
    code = taosHashPut(gCtgMgmt.pCluster, &clusterId, sizeof(clusterId), &clusterCtg, POINTER_BYTES);
D
dapan1121 已提交
2369 2370 2371 2372 2373 2374 2375 2376 2377 2378 2379 2380 2381
    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 已提交
2382
  }
D
dapan1121 已提交
2383 2384

  *catalogHandle = clusterCtg;
D
dapan 已提交
2385
  
D
dapan1121 已提交
2386
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
2387 2388 2389 2390 2391 2392 2393 2394

_return:

  ctgFreeHandle(clusterCtg);
  
  CTG_RET(code);
}

D
dapan1121 已提交
2395 2396
void catalogFreeHandle(SCatalog* pCtg) {
  if (NULL == pCtg) {
D
dapan1121 已提交
2397 2398
    return;
  }
D
dapan1121 已提交
2399

D
dapan 已提交
2400
  if (taosHashRemove(gCtgMgmt.pCluster, &pCtg->clusterId, sizeof(pCtg->clusterId))) {
D
dapan1121 已提交
2401
    ctgWarn("taosHashRemove from cluster failed, may already be freed, clusterId:%"PRIx64, pCtg->clusterId);
D
dapan1121 已提交
2402 2403 2404
    return;
  }

D
dapan1121 已提交
2405
  uint64_t clusterId = pCtg->clusterId;
D
dapan1121 已提交
2406
  
D
dapan1121 已提交
2407
  ctgFreeHandle(pCtg);
D
dapan1121 已提交
2408 2409

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

D
dapan1121 已提交
2412
int32_t catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* version, int64_t* dbId, int32_t *tableNum) {
D
dapan1121 已提交
2413 2414
  CTG_API_ENTER();

D
dapan1121 已提交
2415
  if (NULL == pCtg || NULL == dbFName || NULL == version || NULL == dbId) {
D
dapan1121 已提交
2416 2417 2418 2419
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2420
    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
2421
    ctgInfo("empty db cache, dbFName:%s", dbFName);
D
dapan1121 已提交
2422
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2423 2424
  }

D
dapan1121 已提交
2425 2426 2427
  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
  if (NULL == dbCache) {
D
dapan1121 已提交
2428
    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
2429
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2430 2431
  }

D
dapan 已提交
2432 2433 2434
  bool inCache = false;
  ctgAcquireVgInfo(pCtg, dbCache, &inCache);
  if (!inCache) {
D
dapan1121 已提交
2435
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
2436 2437

    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
2438
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2439 2440
  }

D
dapan1121 已提交
2441
  *version = dbCache->vgInfo->vgVersion;
D
dapan1121 已提交
2442
  *dbId = dbCache->dbId;
D
dapan1121 已提交
2443
  *tableNum = dbCache->vgInfo->numOfTable;
D
dapan1121 已提交
2444 2445 2446

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

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

D
dapan1121 已提交
2450
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2451 2452
}

D
dapan1121 已提交
2453
int32_t catalogGetDBVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* dbFName, SArray** vgroupList) {
D
dapan1121 已提交
2454 2455
  CTG_API_ENTER();

D
dapan1121 已提交
2456 2457 2458
  if (NULL == pCtg || NULL == dbFName || NULL == pRpc || NULL == pMgmtEps || NULL == vgroupList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
2459

D
dapan1121 已提交
2460
  SCtgDBCache* dbCache = NULL;
2461
  int32_t code = 0;
D
dapan1121 已提交
2462
  SArray *vgList = NULL;
D
dapan1121 已提交
2463 2464
  SHashObj *vgHash = NULL;
  SDBVgInfo *vgInfo = NULL;
D
dapan1121 已提交
2465
  CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pRpc, pMgmtEps, dbFName, &dbCache, &vgInfo));
D
dapan1121 已提交
2466 2467 2468 2469
  if (dbCache) {
    vgHash = dbCache->vgInfo->vgHash;
  } else {
    vgHash = vgInfo->vgHash;
D
dapan1121 已提交
2470 2471
  }

D
dapan1121 已提交
2472
  CTG_ERR_JRET(ctgGenerateVgList(pCtg, vgHash, &vgList));
D
dapan1121 已提交
2473 2474 2475 2476 2477

  *vgroupList = vgList;
  vgList = NULL;

_return:
D
dapan1121 已提交
2478 2479

  if (dbCache) {
D
dapan1121 已提交
2480 2481
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
2482 2483
  }

D
dapan1121 已提交
2484 2485
  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
wafwerar's avatar
wafwerar 已提交
2486
    taosMemoryFreeClear(vgInfo);
D
dapan1121 已提交
2487 2488
  }

D
dapan1121 已提交
2489
  CTG_API_LEAVE(code);  
D
dapan1121 已提交
2490 2491 2492
}


D
dapan1121 已提交
2493
int32_t catalogUpdateDBVgInfo(SCatalog* pCtg, const char* dbFName, uint64_t dbId, SDBVgInfo* dbInfo) {
D
dapan1121 已提交
2494
  CTG_API_ENTER();
D
dapan1121 已提交
2495 2496

  int32_t code = 0;
D
dapan1121 已提交
2497
  
D
dapan1121 已提交
2498
  if (NULL == pCtg || NULL == dbFName || NULL == dbInfo) {
D
dapan1121 已提交
2499
    ctgFreeVgInfo(dbInfo);
D
dapan1121 已提交
2500 2501 2502
    CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2503
  code = ctgPushUpdateVgMsgInQueue(pCtg, dbFName, dbId, dbInfo, false);
D
dapan1121 已提交
2504

D
dapan1121 已提交
2505 2506
_return:

D
dapan1121 已提交
2507
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2508 2509 2510
}


D
dapan1121 已提交
2511 2512 2513
int32_t catalogRemoveDB(SCatalog* pCtg, const char* dbFName, uint64_t dbId) {
  CTG_API_ENTER();

D
dapan1121 已提交
2514 2515
  int32_t code = 0;
  
D
dapan1121 已提交
2516 2517
  if (NULL == pCtg || NULL == dbFName) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2518 2519
  }

D
dapan1121 已提交
2520
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2521
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2522
  }
D
dapan1121 已提交
2523

D
dapan1121 已提交
2524
  CTG_ERR_JRET(ctgPushRmDBMsgInQueue(pCtg, dbFName, dbId));
D
dapan 已提交
2525

D
dapan1121 已提交
2526
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2527
  
D
dapan1121 已提交
2528 2529
_return:

D
dapan1121 已提交
2530
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2531 2532
}

D
dapan1121 已提交
2533 2534 2535
int32_t catalogUpdateVgEpSet(SCatalog* pCtg, const char* dbFName, int32_t vgId, SEpSet *epSet) {

}
D
dapan1121 已提交
2536

D
dapan 已提交
2537 2538 2539 2540 2541 2542 2543 2544 2545 2546 2547 2548 2549 2550 2551 2552 2553 2554 2555 2556 2557 2558 2559 2560 2561 2562 2563 2564 2565 2566 2567 2568 2569 2570 2571
int32_t catalogRemoveTableMeta(SCatalog* pCtg, SName* pTableName) {
  CTG_API_ENTER();

  int32_t code = 0;
  
  if (NULL == pCtg || NULL == pTableName) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

  if (NULL == pCtg->dbCache) {
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
  }

  STableMeta *tblMeta = NULL;
  int32_t exist = 0;
  uint64_t dbId = 0;
  CTG_ERR_JRET(ctgGetTableMetaFromCache(pCtg, pTableName, &tblMeta, &exist, 0, &dbId));

  if (0 == exist) {
    ctgDebug("table already not in cache, db:%s, tblName:%s", pTableName->dbname, pTableName->tname);
    goto _return;
  }

  char dbFName[TSDB_DB_FNAME_LEN];
  tNameGetFullDbName(pTableName, dbFName);
  
  if (TSDB_SUPER_TABLE == tblMeta->tableType) {
    CTG_ERR_JRET(ctgPushRmStbMsgInQueue(pCtg, dbFName, dbId, pTableName->tname, tblMeta->suid));
  } else {
    CTG_ERR_JRET(ctgPushRmTblMsgInQueue(pCtg, dbFName, dbId, pTableName->tname));
  }

 
_return:

wafwerar's avatar
wafwerar 已提交
2572
  taosMemoryFreeClear(tblMeta);
D
dapan 已提交
2573 2574 2575 2576 2577

  CTG_API_LEAVE(code);
}


D
dapan1121 已提交
2578 2579 2580
int32_t catalogRemoveStbMeta(SCatalog* pCtg, const char* dbFName, uint64_t dbId, const char* stbName, uint64_t suid) {
  CTG_API_ENTER();

D
dapan 已提交
2581 2582
  int32_t code = 0;
  
D
dapan1121 已提交
2583 2584
  if (NULL == pCtg || NULL == dbFName || NULL == stbName) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
2585 2586
  }

D
dapan1121 已提交
2587
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2588
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2589
  }
D
dapan1121 已提交
2590

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

D
dapan1121 已提交
2593
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2594
  
D
dapan1121 已提交
2595 2596
_return:

D
dapan1121 已提交
2597
  CTG_API_LEAVE(code);
D
dapan 已提交
2598 2599
}

D
dapan1121 已提交
2600 2601 2602
int32_t catalogGetIndexMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, const char *pIndexName, SIndexMeta** pIndexMeta) {

}
D
dapan1121 已提交
2603

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

D
dapan1121 已提交
2607
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_UNKNOWN_STB));
D
dapan1121 已提交
2608
}
D
dapan1121 已提交
2609

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

D
dapan1121 已提交
2613
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_STB));
D
dapan1121 已提交
2614 2615
}

D
dapan1121 已提交
2616 2617 2618 2619 2620 2621 2622
int32_t catalogUpdateSTableMeta(SCatalog* pCtg, STableMetaRsp *rspMsg) {
  CTG_API_ENTER();

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

wafwerar's avatar
wafwerar 已提交
2623
  STableMetaOutput *output = taosMemoryCalloc(1, sizeof(STableMetaOutput));
D
dapan1121 已提交
2624 2625 2626 2627 2628
  if (NULL == output) {
    ctgError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
    CTG_API_LEAVE(TSDB_CODE_CTG_MEM_ERROR);
  }
  
D
dapan1121 已提交
2629 2630
  int32_t code = 0;

D
dapan1121 已提交
2631 2632
  strcpy(output->dbFName, rspMsg->dbFName);
  strcpy(output->tbName, rspMsg->tbName);
D
dapan1121 已提交
2633

D
dapan1121 已提交
2634
  output->dbId = rspMsg->dbId;
D
dapan1121 已提交
2635
  
D
dapan1121 已提交
2636
  SET_META_TYPE_TABLE(output->metaType);
D
dapan1121 已提交
2637
  
D
dapan1121 已提交
2638
  CTG_ERR_JRET(queryCreateTableMetaFromMsg(rspMsg, true, &output->tbMeta));
D
dapan1121 已提交
2639

D
dapan1121 已提交
2640
  CTG_ERR_JRET(ctgPushUpdateTblMsgInQueue(pCtg, output, false));
D
dapan 已提交
2641

D
dapan1121 已提交
2642 2643
  CTG_API_LEAVE(code);
  
D
dapan1121 已提交
2644 2645
_return:

wafwerar's avatar
wafwerar 已提交
2646 2647
  taosMemoryFreeClear(output->tbMeta);
  taosMemoryFreeClear(output);
D
dapan1121 已提交
2648
  
D
dapan1121 已提交
2649
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2650 2651
}

D
dapan1121 已提交
2652 2653 2654 2655 2656 2657 2658 2659 2660
int32_t catalogRefreshDBVgInfo(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const char* dbFName) {
  CTG_API_ENTER();

  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == dbFName) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

  CTG_API_LEAVE(ctgRefreshDBVgInfo(pCtg, pTrans, pMgmtEps, dbFName));
}
D
dapan1121 已提交
2661

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

D
dapan1121 已提交
2665
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pTableName) {
D
dapan1121 已提交
2666 2667 2668
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2669
  CTG_API_LEAVE(ctgRefreshTblMeta(pCtg, pTrans, pMgmtEps, pTableName, CTG_FLAG_FORCE_UPDATE | CTG_FLAG_MAKE_STB(isSTable), NULL, true));
2670
}
2671

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

D
dapan1121 已提交
2675
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_FORCE_UPDATE | CTG_FLAG_MAKE_STB(isSTable)));
D
dapan1121 已提交
2676 2677
}

D
dapan1121 已提交
2678
int32_t catalogGetTableDistVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgList) {
D
dapan1121 已提交
2679
  CTG_API_ENTER();
D
dapan1121 已提交
2680 2681 2682 2683

  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == pVgList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
2684 2685 2686 2687 2688

  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 已提交
2689

D
dapan1121 已提交
2690
  int32_t code = 0;
D
dapan1121 已提交
2691

D
dapan1121 已提交
2692 2693 2694 2695 2696
  while (true) {
    code = ctgGetTableDistVgInfo(pCtg, pRpc, pMgmtEps, pTableName, pVgList);
    if (code) {
      if (TSDB_CODE_CTG_VG_META_MISMATCH == code) {
        CTG_ERR_JRET(ctgRefreshTblMeta(pCtg, pRpc, pMgmtEps, pTableName, CTG_FLAG_FORCE_UPDATE | CTG_FLAG_MAKE_STB(CTG_FLAG_UNKNOWN_STB), NULL, true));
D
dapan 已提交
2697

D
dapan1121 已提交
2698 2699 2700 2701 2702 2703
        char dbFName[TSDB_DB_FNAME_LEN] = {0};
        tNameGetFullDbName(pTableName, dbFName);        
        CTG_ERR_JRET(ctgRefreshDBVgInfo(pCtg, pRpc, pMgmtEps, dbFName));
        
        continue;
      }
D
dapan1121 已提交
2704
    }
D
dapan 已提交
2705

D
dapan1121 已提交
2706
    break;
D
dapan1121 已提交
2707
  }
D
dapan 已提交
2708

D
dapan1121 已提交
2709
_return:
D
dapan 已提交
2710

D
dapan1121 已提交
2711
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2712 2713 2714
}


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

D
dapan1121 已提交
2718 2719 2720 2721 2722
  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 已提交
2723 2724
  SCtgDBCache* dbCache = NULL;
  int32_t code = 0;
H
Haojun Liao 已提交
2725 2726
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);
D
dapan1121 已提交
2727

D
dapan1121 已提交
2728
  SDBVgInfo *vgInfo = NULL;
D
dapan1121 已提交
2729
  CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pTrans, pMgmtEps, db, &dbCache, &vgInfo));
D
dapan1121 已提交
2730

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

D
dapan1121 已提交
2733
_return:
D
dapan1121 已提交
2734

D
dapan1121 已提交
2735
  if (dbCache) {
D
dapan1121 已提交
2736 2737 2738 2739 2740 2741
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
  }

  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
wafwerar's avatar
wafwerar 已提交
2742
    taosMemoryFreeClear(vgInfo);
D
dapan1121 已提交
2743
  }
D
dapan1121 已提交
2744

D
dapan1121 已提交
2745
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2746 2747 2748
}


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

D
dapan1121 已提交
2752
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pReq || NULL == pRsp) {
D
dapan1121 已提交
2753 2754 2755
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2756
  int32_t code = 0;
D
dapan1121 已提交
2757
  pRsp->pTableMeta = NULL;
D
dapan1121 已提交
2758 2759 2760

  if (pReq->pTableName) {
    int32_t tbNum = (int32_t)taosArrayGetSize(pReq->pTableName);
D
dapan1121 已提交
2761
    if (tbNum <= 0) {
D
dapan1121 已提交
2762
      ctgError("empty table name list, tbNum:%d", tbNum);
D
dapan1121 已提交
2763
      CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2764
    }
H
Haojun Liao 已提交
2765

D
dapan1121 已提交
2766 2767
    pRsp->pTableMeta = taosArrayInit(tbNum, POINTER_BYTES);
    if (NULL == pRsp->pTableMeta) {
D
dapan1121 已提交
2768
      ctgError("taosArrayInit %d failed", tbNum);
D
dapan1121 已提交
2769
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
2770 2771 2772 2773 2774 2775
    }
    
    for (int32_t i = 0; i < tbNum; ++i) {
      SName *name = taosArrayGet(pReq->pTableName, i);
      STableMeta *pTableMeta = NULL;
      
D
dapan1121 已提交
2776
      CTG_ERR_JRET(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, name, &pTableMeta, CTG_FLAG_UNKNOWN_STB));
D
dapan1121 已提交
2777 2778 2779

      if (NULL == taosArrayPush(pRsp->pTableMeta, &pTableMeta)) {
        ctgError("taosArrayPush failed, idx:%d", i);
wafwerar's avatar
wafwerar 已提交
2780
        taosMemoryFreeClear(pTableMeta);
D
dapan1121 已提交
2781 2782 2783 2784 2785
        CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
      }
    }
  }

D
dapan1121 已提交
2786 2787 2788 2789
  if (pReq->qNodeRequired) {
    CTG_ERR_JRET(ctgGetQnodeListFromMnode(pCtg, pTrans, pMgmtEps, &pRsp->pEpSetList));
  }

D
dapan1121 已提交
2790
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2791 2792

_return:  
D
dapan1121 已提交
2793

D
dapan1121 已提交
2794 2795 2796 2797
  if (pRsp->pTableMeta) {
    int32_t aSize = taosArrayGetSize(pRsp->pTableMeta);
    for (int32_t i = 0; i < aSize; ++i) {
      STableMeta *pMeta = taosArrayGetP(pRsp->pTableMeta, i);
wafwerar's avatar
wafwerar 已提交
2798
      taosMemoryFreeClear(pMeta);
D
dapan1121 已提交
2799 2800 2801
    }
    
    taosArrayDestroy(pRsp->pTableMeta);
D
dapan1121 已提交
2802
    pRsp->pTableMeta = NULL;
D
dapan1121 已提交
2803
  }
D
dapan 已提交
2804
  
D
dapan1121 已提交
2805
  CTG_API_LEAVE(code);
2806
}
D
dapan 已提交
2807

D
dapan1121 已提交
2808
int32_t catalogGetQnodeList(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SArray* pQnodeList) {
D
dapan1121 已提交
2809
  CTG_API_ENTER();
D
dapan1121 已提交
2810 2811
  
  int32_t code = 0;
D
dapan1121 已提交
2812 2813 2814 2815
  if (NULL == pCtg || NULL == pRpc  || NULL == pMgmtEps || NULL == pQnodeList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2816 2817 2818
  CTG_ERR_JRET(ctgGetQnodeListFromMnode(pCtg, pRpc, pMgmtEps, &pQnodeList));

_return:
D
dapan 已提交
2819

D
dapan1121 已提交
2820
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2821 2822
}

D
dapan1121 已提交
2823
int32_t catalogGetExpiredSTables(SCatalog* pCtg, SSTableMetaVersion **stables, uint32_t *num) {
D
dapan1121 已提交
2824 2825
  CTG_API_ENTER();

D
dapan1121 已提交
2826 2827
  if (NULL == pCtg || NULL == stables || NULL == num) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2828 2829
  }

D
dapan1121 已提交
2830 2831 2832 2833
  CTG_API_LEAVE(ctgMetaRentGet(&pCtg->stbRent, (void **)stables, num, sizeof(SSTableMetaVersion)));
}

int32_t catalogGetExpiredDBs(SCatalog* pCtg, SDbVgVersion **dbs, uint32_t *num) {
D
dapan1121 已提交
2834
  CTG_API_ENTER();
D
dapan1121 已提交
2835 2836 2837 2838
  
  if (NULL == pCtg || NULL == dbs || NULL == num) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
2839

D
dapan1121 已提交
2840
  CTG_API_LEAVE(ctgMetaRentGet(&pCtg->dbRent, (void **)dbs, num, sizeof(SDbVgVersion)));
D
dapan1121 已提交
2841 2842
}

D
dapan 已提交
2843

D
dapan 已提交
2844
void catalogDestroy(void) {
D
dapan1121 已提交
2845 2846
  qInfo("start to destroy catalog");
  
wafwerar's avatar
wafwerar 已提交
2847
  if (NULL == gCtgMgmt.pCluster || atomic_load_8((int8_t*)&gCtgMgmt.exit)) {
D
dapan1121 已提交
2848 2849 2850
    return;
  }

wafwerar's avatar
wafwerar 已提交
2851
  atomic_store_8((int8_t*)&gCtgMgmt.exit, true);
D
dapan 已提交
2852

D
dapan1121 已提交
2853 2854
  tsem_post(&gCtgMgmt.queue.reqSem);
  tsem_post(&gCtgMgmt.queue.rspSem);
D
dapan1121 已提交
2855

D
dapan1121 已提交
2856
  while (CTG_IS_LOCKED(&gCtgMgmt.lock)) {
wafwerar's avatar
wafwerar 已提交
2857
    taosUsleep(1);
D
dapan1121 已提交
2858 2859
  }
  
D
dapan 已提交
2860
  CTG_LOCK(CTG_WRITE, &gCtgMgmt.lock);
D
dapan1121 已提交
2861

D
dapan1121 已提交
2862
  SCatalog *pCtg = NULL;
D
dapan 已提交
2863
  void *pIter = taosHashIterate(gCtgMgmt.pCluster, NULL);
D
dapan1121 已提交
2864
  while (pIter) {
D
dapan1121 已提交
2865
    pCtg = *(SCatalog **)pIter;
D
dapan1121 已提交
2866

D
dapan1121 已提交
2867 2868
    if (pCtg) {
      catalogFreeHandle(pCtg);
D
dapan1121 已提交
2869 2870
    }
    
D
dapan 已提交
2871
    pIter = taosHashIterate(gCtgMgmt.pCluster, pIter);
D
dapan 已提交
2872
  }
D
dapan1121 已提交
2873
  
D
dapan 已提交
2874 2875
  taosHashCleanup(gCtgMgmt.pCluster);
  gCtgMgmt.pCluster = NULL;
D
dapan1121 已提交
2876

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

D
dapan1121 已提交
2879
  qInfo("catalog destroyed");
D
dapan 已提交
2880 2881 2882 2883
}