catalog.c 90.8 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
#include "systable.h"
21

D
dapan 已提交
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
int32_t ctgActUpdateUser(SCtgMetaAction *action);
D
dapan 已提交
28

D
dapan1121 已提交
29
extern SCtgDebug gCTGDebug;
D
dapan 已提交
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
SCatalogMgmt gCtgMgmt = {0};
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
D
dapan 已提交
55 56 57 58 59
                          },
                          {
                            CTG_ACT_UPDATE_USER,
                            "update user",
                            ctgActUpdateUser
D
dapan 已提交
60 61 62
                          }
};

D
dapan1121 已提交
63 64 65 66 67 68 69 70 71 72 73 74 75
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 已提交
76
  taosMemoryFreeClear(mgmt->slots);
D
dapan1121 已提交
77 78 79 80 81 82
}


void ctgFreeTableMetaCache(SCtgTbMetaCache *cache) {
  CTG_LOCK(CTG_WRITE, &cache->stbLock);
  if (cache->stbCache) {
D
dapan1121 已提交
83
    int32_t stblNum = taosHashGetSize(cache->stbCache);  
D
dapan1121 已提交
84 85
    taosHashCleanup(cache->stbCache);
    cache->stbCache = NULL;
D
dapan1121 已提交
86
    CTG_CACHE_STAT_SUB(stblNum, stblNum);
D
dapan1121 已提交
87 88 89 90 91
  }
  CTG_UNLOCK(CTG_WRITE, &cache->stbLock);

  CTG_LOCK(CTG_WRITE, &cache->metaLock);
  if (cache->metaCache) {
D
dapan1121 已提交
92
    int32_t tblNum = taosHashGetSize(cache->metaCache);
D
dapan1121 已提交
93 94
    taosHashCleanup(cache->metaCache);
    cache->metaCache = NULL;
D
dapan1121 已提交
95
    CTG_CACHE_STAT_SUB(tblNum, tblNum);
D
dapan1121 已提交
96 97 98 99 100 101 102 103 104 105 106 107 108 109
  }
  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 已提交
110
  taosMemoryFreeClear(vgInfo);
D
dapan1121 已提交
111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130
}

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) {
D
dapan1121 已提交
131 132
    int32_t dbNum = taosHashGetSize(pCtg->dbCache);
    
D
dapan1121 已提交
133 134 135 136 137 138 139 140 141 142 143 144
    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);
D
dapan1121 已提交
145 146
    
    CTG_CACHE_STAT_SUB(dbNum, dbNum);
D
dapan1121 已提交
147 148
  }
  
wafwerar's avatar
wafwerar 已提交
149
  taosMemoryFree(pCtg);
D
dapan1121 已提交
150 151 152 153 154 155 156 157
}



void ctgWaitAction(SCtgMetaAction *action) {
  while (true) {
    tsem_wait(&gCtgMgmt.queue.rspSem);
    
X
Xiaoyu Wang 已提交
158
    if (atomic_load_8((int8_t*)&gCtgMgmt.exit)) {
D
dapan1121 已提交
159 160 161 162 163 164 165 166 167 168 169 170
      tsem_post(&gCtgMgmt.queue.rspSem);
      break;
    }

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

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

void ctgPopAction(SCtgMetaAction **action) {
D
dapan1121 已提交
173
  SCtgQNode *orig = gCtgMgmt.queue.head;
D
dapan 已提交
174
  
D
dapan1121 已提交
175 176
  SCtgQNode *node = gCtgMgmt.queue.head->next;
  gCtgMgmt.queue.head = gCtgMgmt.queue.head->next;
D
dapan 已提交
177 178 179

  CTG_QUEUE_SUB();
  
wafwerar's avatar
wafwerar 已提交
180
  taosMemoryFreeClear(orig);
D
dapan 已提交
181 182 183 184 185

  *action = &node->action;
}


D
dapan1121 已提交
186
int32_t ctgPushAction(SCatalog* pCtg, SCtgMetaAction *action) {
wafwerar's avatar
wafwerar 已提交
187
  SCtgQNode *node = taosMemoryCalloc(1, sizeof(SCtgQNode));
D
dapan 已提交
188 189 190 191
  if (NULL == node) {
    qError("calloc %d failed", (int32_t)sizeof(SCtgQNode));
    CTG_RET(TSDB_CODE_CTG_MEM_ERROR);
  }
D
dapan1121 已提交
192 193

  action->seqId = atomic_add_fetch_64(&gCtgMgmt.queue.seqId, 1);
D
dapan 已提交
194 195 196
  
  node->action = *action;

D
dapan1121 已提交
197 198 199 200
  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 已提交
201 202

  CTG_QUEUE_ADD();
D
dapan1121 已提交
203
  CTG_RUNTIME_STAT_ADD(qNum, 1);
D
dapan 已提交
204

D
dapan1121 已提交
205
  tsem_post(&gCtgMgmt.queue.reqSem);
D
dapan 已提交
206

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

D
dapan1121 已提交
209 210
  if (action->syncReq) {
    ctgWaitAction(action);
D
dapan1121 已提交
211 212
  }

D
dapan1121 已提交
213
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
214 215 216
}


D
dapan1121 已提交
217 218 219
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 已提交
220
  SCtgRemoveDBMsg *msg = taosMemoryMalloc(sizeof(SCtgRemoveDBMsg));
D
dapan1121 已提交
221 222 223 224 225
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgRemoveDBMsg));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

D
dapan1121 已提交
226
  char *p = strchr(dbFName, '.');
D
dapan1121 已提交
227
  if (p && CTG_IS_SYS_DBNAME(p + 1)) {
D
dapan1121 已提交
228 229 230
    dbFName = p + 1;
  }

D
dapan1121 已提交
231 232 233 234 235 236
  msg->pCtg = pCtg;
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  msg->dbId = dbId;

  action.data = msg;

D
dapan1121 已提交
237
  CTG_ERR_JRET(ctgPushAction(pCtg, &action));
D
dapan1121 已提交
238 239 240 241

  return TSDB_CODE_SUCCESS;

_return:
H
Haojun Liao 已提交
242

wafwerar's avatar
wafwerar 已提交
243
  taosMemoryFreeClear(action.data);
H
Haojun Liao 已提交
244
  CTG_RET(code);
D
dapan1121 已提交
245 246 247
}


D
dapan1121 已提交
248
int32_t ctgPushRmStbMsgInQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid, bool syncReq) {
D
dapan1121 已提交
249
  int32_t code = 0;
D
dapan1121 已提交
250
  SCtgMetaAction action= {.act = CTG_ACT_REMOVE_STB, .syncReq = syncReq};
wafwerar's avatar
wafwerar 已提交
251
  SCtgRemoveStbMsg *msg = taosMemoryMalloc(sizeof(SCtgRemoveStbMsg));
D
dapan1121 已提交
252 253 254 255 256 257 258 259 260 261 262 263 264
  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 已提交
265
  CTG_ERR_JRET(ctgPushAction(pCtg, &action));
D
dapan1121 已提交
266 267 268 269

  return TSDB_CODE_SUCCESS;

_return:
H
Haojun Liao 已提交
270

wafwerar's avatar
wafwerar 已提交
271
  taosMemoryFreeClear(action.data);
H
Haojun Liao 已提交
272
  CTG_RET(code);
D
dapan1121 已提交
273 274 275 276
}



D
dapan1121 已提交
277
int32_t ctgPushRmTblMsgInQueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *tbName, bool syncReq) {
D
dapan1121 已提交
278
  int32_t code = 0;
D
dapan1121 已提交
279
  SCtgMetaAction action= {.act = CTG_ACT_REMOVE_TBL, .syncReq = syncReq};
wafwerar's avatar
wafwerar 已提交
280
  SCtgRemoveTblMsg *msg = taosMemoryMalloc(sizeof(SCtgRemoveTblMsg));
D
dapan1121 已提交
281 282 283 284 285 286 287 288 289 290 291 292
  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 已提交
293
  CTG_ERR_JRET(ctgPushAction(pCtg, &action));
D
dapan1121 已提交
294 295 296 297

  return TSDB_CODE_SUCCESS;

_return:
H
Haojun Liao 已提交
298

wafwerar's avatar
wafwerar 已提交
299
  taosMemoryFreeClear(action.data);
H
Haojun Liao 已提交
300
  CTG_RET(code);
D
dapan1121 已提交
301 302
}

D
dapan1121 已提交
303 304 305
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 已提交
306
  SCtgUpdateVgMsg *msg = taosMemoryMalloc(sizeof(SCtgUpdateVgMsg));
D
dapan1121 已提交
307 308 309 310
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateVgMsg));
    ctgFreeVgInfo(dbInfo);
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
311 312
  }

D
dapan1121 已提交
313
  char *p = strchr(dbFName, '.');
D
dapan1121 已提交
314
  if (p && CTG_IS_SYS_DBNAME(p + 1)) {
D
dapan1121 已提交
315 316 317
    dbFName = p + 1;
  }

D
dapan1121 已提交
318 319 320 321
  strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
  msg->pCtg = pCtg;
  msg->dbId = dbId;
  msg->dbInfo = dbInfo;
D
dapan1121 已提交
322

D
dapan1121 已提交
323
  action.data = msg;
D
dapan1121 已提交
324

D
dapan1121 已提交
325
  CTG_ERR_JRET(ctgPushAction(pCtg, &action));
D
dapan1121 已提交
326

D
dapan1121 已提交
327
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
328

D
dapan1121 已提交
329
_return:
D
dapan1121 已提交
330

D
dapan1121 已提交
331
  ctgFreeVgInfo(dbInfo);
wafwerar's avatar
wafwerar 已提交
332
  taosMemoryFreeClear(action.data);
D
dapan1121 已提交
333
  CTG_RET(code);
D
dapan1121 已提交
334 335
}

D
dapan1121 已提交
336 337
int32_t ctgPushUpdateTblMsgInQueue(SCatalog* pCtg, STableMetaOutput *output, bool syncReq) {
  int32_t code = 0;
D
dapan1121 已提交
338
  SCtgMetaAction action= {.act = CTG_ACT_UPDATE_TBL, .syncReq = syncReq};
wafwerar's avatar
wafwerar 已提交
339
  SCtgUpdateTblMsg *msg = taosMemoryMalloc(sizeof(SCtgUpdateTblMsg));
D
dapan1121 已提交
340 341 342 343 344
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateTblMsg));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

D
dapan1121 已提交
345
  char *p = strchr(output->dbFName, '.');
D
dapan1121 已提交
346
  if (p && CTG_IS_SYS_DBNAME(p + 1)) {
D
dapan1121 已提交
347 348 349
    memmove(output->dbFName, p + 1, strlen(p + 1));
  }

D
dapan1121 已提交
350 351 352 353 354 355 356 357 358 359 360
  msg->pCtg = pCtg;
  msg->output = output;

  action.data = msg;

  CTG_ERR_JRET(ctgPushAction(pCtg, &action));

  return TSDB_CODE_SUCCESS;
  
_return:

wafwerar's avatar
wafwerar 已提交
361
  taosMemoryFreeClear(msg);
D
dapan1121 已提交
362 363 364 365
  
  CTG_RET(code);
}

D
dapan 已提交
366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385
int32_t ctgPushUpdateUserMsgInQueue(SCatalog* pCtg, SGetUserAuthRsp *pAuth, bool syncReq) {
  int32_t code = 0;
  SCtgMetaAction action= {.act = CTG_ACT_UPDATE_USER, .syncReq = syncReq};
  SCtgUpdateUserMsg *msg = taosMemoryMalloc(sizeof(SCtgUpdateUserMsg));
  if (NULL == msg) {
    ctgError("malloc %d failed", (int32_t)sizeof(SCtgUpdateUserMsg));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

  msg->pCtg = pCtg;
  msg->userAuth = *pAuth;

  action.data = msg;

  CTG_ERR_JRET(ctgPushAction(pCtg, &action));

  return TSDB_CODE_SUCCESS;
  
_return:

D
dapan 已提交
386
  tFreeSGetUserAuthRsp(pAuth);
D
dapan 已提交
387 388 389 390
  taosMemoryFreeClear(msg);
  
  CTG_RET(code);
}
D
dapan1121 已提交
391

D
dapan 已提交
392
int32_t ctgAcquireVgInfo(SCatalog *pCtg, SCtgDBCache *dbCache, bool *inCache) {
D
dapan1121 已提交
393 394
  CTG_LOCK(CTG_READ, &dbCache->vgLock);
  
D
dapan 已提交
395 396 397 398 399 400 401 402 403 404
  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 已提交
405 406 407
  if (NULL == dbCache->vgInfo) {
    CTG_UNLOCK(CTG_READ, &dbCache->vgLock);

D
dapan 已提交
408
    *inCache = false;
D
dapan1121 已提交
409
    ctgDebug("db vgInfo is empty, dbId:%"PRIx64, dbCache->dbId);
D
dapan1121 已提交
410 411 412
    return TSDB_CODE_SUCCESS;
  }

D
dapan 已提交
413 414
  *inCache = true;
  
D
dapan1121 已提交
415 416 417 418 419 420 421 422 423 424 425 426 427 428
  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 已提交
429

D
dapan1121 已提交
430 431 432
void ctgReleaseDBCache(SCatalog *pCtg, SCtgDBCache *dbCache) {
  taosHashRelease(pCtg->dbCache, dbCache);
}
D
dapan1121 已提交
433

D
dapan1121 已提交
434 435 436
void ctgReleaseVgInfo(SCtgDBCache *dbCache) {
  CTG_UNLOCK(CTG_READ, &dbCache->vgLock);
}
D
dapan1121 已提交
437

D
dapan1121 已提交
438 439 440
void ctgWReleaseVgInfo(SCtgDBCache *dbCache) {
  CTG_UNLOCK(CTG_WRITE, &dbCache->vgLock);
}
D
dapan1121 已提交
441

D
dapan1121 已提交
442

D
dapan 已提交
443
int32_t ctgAcquireDBCacheImpl(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache, bool acquire) {
D
dapan1121 已提交
444
  char *p = strchr(dbFName, '.');
D
dapan1121 已提交
445
  if (p && CTG_IS_SYS_DBNAME(p + 1)) {
D
dapan1121 已提交
446 447 448
    dbFName = p + 1;
  }

D
dapan 已提交
449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485
  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 已提交
486
int32_t ctgAcquireVgInfoFromCache(SCatalog* pCtg, const char *dbFName, SCtgDBCache **pCache, bool *inCache) {
D
dapan1121 已提交
487 488
  SCtgDBCache *dbCache = NULL;

D
dapan1121 已提交
489
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
490 491
    ctgDebug("empty db cache, dbFName:%s", dbFName);
    goto _return;
D
dapan1121 已提交
492 493 494 495
  }

  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
  if (NULL == dbCache) {  
D
dapan1121 已提交
496 497
    ctgDebug("db %s not in cache", dbFName);
    goto _return;
D
dapan1121 已提交
498 499
  }
  
D
dapan 已提交
500 501
  ctgAcquireVgInfo(pCtg, dbCache, inCache);
  if (!(*inCache)) {
D
dapan1121 已提交
502 503
    ctgDebug("vgInfo of db %s not in cache", dbFName);
    goto _return;
D
dapan1121 已提交
504
  }
D
dapan1121 已提交
505

D
dapan1121 已提交
506
  *pCache = dbCache;
D
dapan1121 已提交
507
  *inCache = true;
D
dapan1121 已提交
508

D
dapan1121 已提交
509 510
  CTG_CACHE_STAT_ADD(vgHitNum, 1);

D
dapan1121 已提交
511
  ctgDebug("Got db vgInfo from cache, dbFName:%s", dbFName);
D
dapan1121 已提交
512
  
D
dapan1121 已提交
513 514 515 516 517 518 519 520 521 522 523 524 525
  return TSDB_CODE_SUCCESS;

_return:

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

  *pCache = NULL;
  *inCache = false;

  CTG_CACHE_STAT_ADD(vgMissNum, 1);
  
D
dapan1121 已提交
526 527 528
  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
529
int32_t ctgGetQnodeListFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SArray *out) {
D
dapan1121 已提交
530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550
  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 已提交
551
    ctgError("error rsp for qnode list, error:%s", tstrerror(rpcRsp.code));
D
dapan1121 已提交
552 553 554 555 556 557 558 559 560
    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
dapan 已提交
561
  ctgDebug("Got qnode list from mnode, listNum:%d", (int32_t)taosArrayGetSize(out));
D
dapan1121 已提交
562 563 564

  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
565

566 567 568 569
//typedef void __taos_async_internal_fn_t(void* param, )
void ctgGetDBVgInfoFromMnode_a(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SBuildUseDBInput *input, __taos_async_fn_t fp) {

}
D
dapan1121 已提交
570

D
dapan1121 已提交
571
int32_t ctgGetDBVgInfoFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SBuildUseDBInput *input, SUseDbOutput *out) {
D
dapan1121 已提交
572 573 574
  char *msg = NULL;
  int32_t msgLen = 0;

D
dapan 已提交
575
  ctgDebug("try to get db vgInfo from mnode, dbFName:%s", input->db);
D
dapan1121 已提交
576 577 578 579 580 581

  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 已提交
582
  
D
dapan1121 已提交
583
  SRpcMsg rpcMsg = {
H
Hongze Cheng 已提交
584
      .msgType = TDMT_MND_USE_DB,
D
catalog  
dapan1121 已提交
585
      .pCont   = msg,
D
dapan1121 已提交
586 587 588 589 590 591
      .contLen = msgLen,
  };

  SRpcMsg rpcRsp = {0};

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

D
dapan1121 已提交
597 598 599 600 601
  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 已提交
602

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

D
dapan1121 已提交
605 606
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
607

D
dapan1121 已提交
608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644
int32_t ctgGetDBCfgFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char *dbFName, SDbCfgInfo *out) {
  char *msg = NULL;
  int32_t msgLen = 0;

  ctgDebug("try to get db cfg from mnode, dbFName:%s", dbFName);

  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_MND_GET_DB_CFG)]((void *)dbFName, &msg, 0, &msgLen);
  if (code) {
    ctgError("Build get db cfg msg failed, code:%x, db:%s", code, dbFName);
    CTG_ERR_RET(code);
  }
  
  SRpcMsg rpcMsg = {
      .msgType = TDMT_MND_GET_DB_CFG,
      .pCont   = msg,
      .contLen = msgLen,
  };

  SRpcMsg rpcRsp = {0};

  rpcSendRecv(pRpc, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
    ctgError("error rsp for get db cfg, error:%s, db:%s", tstrerror(rpcRsp.code), dbFName);
    CTG_ERR_RET(rpcRsp.code);
  }

  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_GET_DB_CFG)](out, rpcRsp.pCont, rpcRsp.contLen);
  if (code) {
    ctgError("Process get db cfg rsp failed, code:%x, db:%s", code, dbFName);
    CTG_ERR_RET(code);
  }

  ctgDebug("Got db cfg from mnode, dbFName:%s", dbFName);

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681
int32_t ctgGetIndexInfoFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char *indexName, SIndexInfo *out) {
  char *msg = NULL;
  int32_t msgLen = 0;

  ctgDebug("try to get index from mnode, indexName:%s", indexName);

  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_MND_GET_INDEX)]((void *)indexName, &msg, 0, &msgLen);
  if (code) {
    ctgError("Build get index msg failed, code:%x, db:%s", code, indexName);
    CTG_ERR_RET(code);
  }
  
  SRpcMsg rpcMsg = {
      .msgType = TDMT_MND_GET_INDEX,
      .pCont   = msg,
      .contLen = msgLen,
  };

  SRpcMsg rpcRsp = {0};

  rpcSendRecv(pRpc, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
    ctgError("error rsp for get index, error:%s, indexName:%s", tstrerror(rpcRsp.code), indexName);
    CTG_ERR_RET(rpcRsp.code);
  }

  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_GET_INDEX)](out, rpcRsp.pCont, rpcRsp.contLen);
  if (code) {
    ctgError("Process get index rsp failed, code:%x, indexName:%s", code, indexName);
    CTG_ERR_RET(code);
  }

  ctgDebug("Got index from mnode, indexName:%s", indexName);

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
682
int32_t ctgGetUdfInfoFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char *funcName, SFuncInfo **out) {
D
dapan1121 已提交
683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703
  char *msg = NULL;
  int32_t msgLen = 0;

  ctgDebug("try to get udf info from mnode, funcName:%s", funcName);

  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_MND_RETRIEVE_FUNC)]((void *)funcName, &msg, 0, &msgLen);
  if (code) {
    ctgError("Build get udf msg failed, code:%x, db:%s", code, funcName);
    CTG_ERR_RET(code);
  }
  
  SRpcMsg rpcMsg = {
      .msgType = TDMT_MND_RETRIEVE_FUNC,
      .pCont   = msg,
      .contLen = msgLen,
  };

  SRpcMsg rpcRsp = {0};

  rpcSendRecv(pRpc, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
D
dapan1121 已提交
704 705 706 707 708 709
    if (TSDB_CODE_MND_FUNC_NOT_EXIST == rpcRsp.code) {
      ctgDebug("funcName %s not exist in mnode", funcName);
      taosMemoryFreeClear(*out);
      CTG_RET(TSDB_CODE_SUCCESS);
    }
    
D
dapan1121 已提交
710 711 712 713
    ctgError("error rsp for get udf, error:%s, funcName:%s", tstrerror(rpcRsp.code), funcName);
    CTG_ERR_RET(rpcRsp.code);
  }

D
dapan1121 已提交
714
  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_RETRIEVE_FUNC)](*out, rpcRsp.pCont, rpcRsp.contLen);
D
dapan1121 已提交
715 716 717 718 719 720 721 722 723 724
  if (code) {
    ctgError("Process get udf rsp failed, code:%x, funcName:%s", code, funcName);
    CTG_ERR_RET(code);
  }

  ctgDebug("Got udf from mnode, funcName:%s", funcName);

  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761
int32_t ctgGetUserDbAuthFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char *user, SGetUserAuthRsp *authRsp) {
  char *msg = NULL;
  int32_t msgLen = 0;

  ctgDebug("try to get user auth from mnode, user:%s", user);

  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_MND_GET_USER_AUTH)]((void *)user, &msg, 0, &msgLen);
  if (code) {
    ctgError("Build get user auth msg failed, code:%x, db:%s", code, user);
    CTG_ERR_RET(code);
  }
  
  SRpcMsg rpcMsg = {
      .msgType = TDMT_MND_GET_USER_AUTH,
      .pCont   = msg,
      .contLen = msgLen,
  };

  SRpcMsg rpcRsp = {0};

  rpcSendRecv(pRpc, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
    ctgError("error rsp for get user auth, error:%s, user:%s", tstrerror(rpcRsp.code), user);
    CTG_ERR_RET(rpcRsp.code);
  }

  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_GET_USER_AUTH)](authRsp, rpcRsp.pCont, rpcRsp.contLen);
  if (code) {
    ctgError("Process get user auth rsp failed, code:%x, user:%s", code, user);
    CTG_ERR_RET(code);
  }

  ctgDebug("Got user auth from mnode, user:%s", user);

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
762

D
dapan1121 已提交
763 764
int32_t ctgIsTableMetaExistInCache(SCatalog* pCtg, char *dbFName, char* tbName, int32_t *exist) {
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
765
    *exist = 0;
D
dapan1121 已提交
766 767 768 769
    ctgWarn("empty db cache, dbFName:%s, tbName:%s", dbFName, tbName);
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
770 771
  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
772 773
  if (NULL == dbCache) {
    *exist = 0;
D
dapan1121 已提交
774 775 776
    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
777
  size_t sz = 0;
D
dapan1121 已提交
778 779 780 781
  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 已提交
782
  if (NULL == tbMeta) {
D
dapan 已提交
783
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
784
    
D
dapan1121 已提交
785
    *exist = 0;
D
dapan1121 已提交
786
    ctgDebug("tbmeta not in cache, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
787 788 789 790
    return TSDB_CODE_SUCCESS;
  }

  *exist = 1;
D
dapan1121 已提交
791

D
dapan 已提交
792
  ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
793
  
D
dapan1121 已提交
794
  ctgDebug("tbmeta is in cache, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
795 796 797 798
  
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
799

D
dapan1121 已提交
800
int32_t ctgGetTableMetaFromCache(SCatalog* pCtg, const SName* pTableName, STableMeta** pTableMeta, bool *inCache, int32_t flag, uint64_t *dbId) {
D
dapan1121 已提交
801
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
802 803
    ctgDebug("empty tbmeta cache, tbName:%s", pTableName->tname);
    goto _return;
D
dapan1121 已提交
804 805
  }

D
dapan1121 已提交
806
  char dbFName[TSDB_DB_FNAME_LEN] = {0};
D
dapan1121 已提交
807
  if (CTG_FLAG_IS_SYS_DB(flag)) {
D
dapan1121 已提交
808 809 810 811
    strcpy(dbFName, pTableName->dbname);
  } else {
    tNameGetFullDbName(pTableName, dbFName);
  }
D
dapan1121 已提交
812

D
dapan1121 已提交
813 814
  *pTableMeta = NULL;

D
dapan1121 已提交
815 816
  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
817
  if (NULL == dbCache) {
D
dapan1121 已提交
818 819
    ctgDebug("db %s not in cache", pTableName->tname);
    goto _return;
D
dapan1121 已提交
820 821
  }
  
H
Haojun Liao 已提交
822
  int32_t sz = 0;
D
dapan1121 已提交
823
  CTG_LOCK(CTG_READ, &dbCache->tbCache.metaLock);
H
Haojun Liao 已提交
824
  int32_t code = taosHashGetDup_m(dbCache->tbCache.metaCache, pTableName->tname, strlen(pTableName->tname), (void **)pTableMeta, &sz);
D
dapan1121 已提交
825 826
  CTG_UNLOCK(CTG_READ, &dbCache->tbCache.metaLock);

D
dapan1121 已提交
827
  if (NULL == *pTableMeta) {
D
dapan1121 已提交
828 829
    ctgReleaseDBCache(pCtg, dbCache);
    ctgDebug("tbl not in cache, dbFName:%s, tbName:%s", dbFName, pTableName->tname);
D
dapan1121 已提交
830
    goto _return;
D
dapan1121 已提交
831 832
  }

D
dapan1121 已提交
833 834 835
  if (dbId) {
    *dbId = dbCache->dbId;
  }
H
Haojun Liao 已提交
836

H
Haojun Liao 已提交
837
  STableMeta* tbMeta = *pTableMeta;
D
dapan1121 已提交
838

D
dapan1121 已提交
839
  if (tbMeta->tableType != TSDB_CHILD_TABLE) {
D
dapan1121 已提交
840
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
841
    ctgDebug("Got meta from cache, type:%d, dbFName:%s, tbName:%s", tbMeta->tableType, dbFName, pTableName->tname);
D
dapan1121 已提交
842 843 844 845

    *inCache = true;
    CTG_CACHE_STAT_ADD(tblHitNum, 1);
    
D
dapan1121 已提交
846 847 848
    return TSDB_CODE_SUCCESS;
  }
  
D
dapan1121 已提交
849
  CTG_LOCK(CTG_READ, &dbCache->tbCache.stbLock);
D
dapan1121 已提交
850
  
D
dapan1121 已提交
851
  STableMeta **stbMeta = taosHashGet(dbCache->tbCache.stbCache, &tbMeta->suid, sizeof(tbMeta->suid));
D
dapan1121 已提交
852
  if (NULL == stbMeta || NULL == *stbMeta) {
D
dapan1121 已提交
853
    CTG_UNLOCK(CTG_READ, &dbCache->tbCache.stbLock);
D
dapan1121 已提交
854 855
    ctgReleaseDBCache(pCtg, dbCache);
    ctgError("stb not in stbCache, suid:%"PRIx64, tbMeta->suid);
wafwerar's avatar
wafwerar 已提交
856
    taosMemoryFreeClear(*pTableMeta);
D
dapan1121 已提交
857
    goto _return;
D
dapan1121 已提交
858
  }
D
dapan1121 已提交
859

D
dapan1121 已提交
860
  if ((*stbMeta)->suid != tbMeta->suid) {    
D
dapan1121 已提交
861
    CTG_UNLOCK(CTG_READ, &dbCache->tbCache.stbLock);
D
dapan1121 已提交
862
    ctgReleaseDBCache(pCtg, dbCache);
wafwerar's avatar
wafwerar 已提交
863
    taosMemoryFreeClear(*pTableMeta);
D
dapan1121 已提交
864
    ctgError("stable suid in stbCache mis-match, expected suid:%"PRIx64 ",actual suid:%"PRIx64, tbMeta->suid, (*stbMeta)->suid);
D
dapan1121 已提交
865 866
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
  }
D
dapan1121 已提交
867

D
dapan1121 已提交
868
  int32_t metaSize = CTG_META_SIZE(*stbMeta);
wafwerar's avatar
wafwerar 已提交
869
  *pTableMeta = taosMemoryRealloc(*pTableMeta, metaSize);
D
dapan1121 已提交
870
  if (NULL == *pTableMeta) {    
D
dapan1121 已提交
871
    CTG_UNLOCK(CTG_READ, &dbCache->tbCache.stbLock);
D
dapan1121 已提交
872
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
873
    ctgError("realloc size[%d] failed", metaSize);
D
dapan1121 已提交
874
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
875 876
  }

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

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

D
dapan1121 已提交
881
  ctgReleaseDBCache(pCtg, dbCache);
D
dapan 已提交
882

D
dapan1121 已提交
883 884 885
  *inCache = true;
  CTG_CACHE_STAT_ADD(tblHitNum, 1);

D
dapan1121 已提交
886
  ctgDebug("Got tbmeta from cache, dbFName:%s, tbName:%s", dbFName, pTableName->tname);
D
dapan1121 已提交
887
  
D
dapan1121 已提交
888 889 890 891 892 893 894
  return TSDB_CODE_SUCCESS;

_return:

  *inCache = false;
  CTG_CACHE_STAT_ADD(tblMissNum, 1);
  
D
dapan1121 已提交
895 896 897
  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
898
int32_t ctgGetTableTypeFromCache(SCatalog* pCtg, const char* dbFName, const char *tableName, int32_t *tbType) {
D
dapan1121 已提交
899
  if (NULL == pCtg->dbCache) {
D
dapan 已提交
900
    ctgWarn("empty db cache, dbFName:%s, tbName:%s", dbFName, tableName);  
D
dapan1121 已提交
901 902
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
903
  
D
dapan 已提交
904 905
  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
906 907 908
  if (NULL == dbCache) {
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
909

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

D
dapan1121 已提交
913
  if (NULL == pTableMeta) {
D
dapan1121 已提交
914
    CTG_UNLOCK(CTG_READ, &dbCache->tbCache.metaLock);
D
dapan 已提交
915
    ctgWarn("tbl not in cache, dbFName:%s, tbName:%s", dbFName, tableName);  
D
dapan 已提交
916
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
917
    
D
dapan1121 已提交
918 919 920
    return TSDB_CODE_SUCCESS;
  }

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

D
dapan1121 已提交
923 924 925 926
  taosHashRelease(dbCache->tbCache.metaCache, pTableMeta);

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

D
dapan 已提交
927
  ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
928

D
dapan 已提交
929
  ctgDebug("Got tbtype from cache, dbFName:%s, tbName:%s, type:%d", dbFName, tableName, *tbType);  
D
dapan1121 已提交
930 931 932 933
  
  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
934
int32_t ctgChkAuthFromCache(SCatalog* pCtg, const char* user, const char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass) {
D
dapan 已提交
935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951
  if (NULL == pCtg->userCache) {
    ctgDebug("empty user auth cache, user:%s", user);
    goto _return;
  }
  
  SCtgUserAuth *pUser = (SCtgUserAuth *)taosHashGet(pCtg->userCache, user, strlen(user));
  if (NULL == pUser) {
    ctgDebug("user not in cache, user:%s", user);
    goto _return;
  }

  *inCache = true;

  ctgDebug("Got user from cache, user:%s", user);
  CTG_CACHE_STAT_ADD(userHitNum, 1);
  
  if (pUser->superUser) {
D
dapan 已提交
952
    *pass = true;
D
dapan 已提交
953 954 955 956 957
    return TSDB_CODE_SUCCESS;
  }

  CTG_LOCK(CTG_READ, &pUser->lock);
  if (pUser->createdDbs && taosHashGet(pUser->createdDbs, dbFName, strlen(dbFName))) {
D
dapan 已提交
958
    *pass = true;
D
dapan 已提交
959 960 961 962
    CTG_UNLOCK(CTG_READ, &pUser->lock);
    return TSDB_CODE_SUCCESS;
  }
  
D
dapan 已提交
963 964
  if (pUser->readDbs && taosHashGet(pUser->readDbs, dbFName, strlen(dbFName)) && type == AUTH_TYPE_READ) {
    *pass = true;
D
dapan 已提交
965 966
  }
  
D
dapan 已提交
967 968
  if (pUser->writeDbs && taosHashGet(pUser->writeDbs, dbFName, strlen(dbFName)) && type == AUTH_TYPE_WRITE) {
    *pass = true;
D
dapan 已提交
969 970 971 972 973 974 975 976 977 978 979 980 981 982
  }

  CTG_UNLOCK(CTG_READ, &pUser->lock);
  
  return TSDB_CODE_SUCCESS;

_return:

  *inCache = false;
  CTG_CACHE_STAT_ADD(userMissNum, 1);
  
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
983
int32_t ctgGetTableMetaFromMnodeImpl(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, char *dbFName, char* tbName, STableMetaOutput* output) {
D
dapan1121 已提交
984
  SBuildTableMetaInput bInput = {.vgId = 0, .dbFName = dbFName, .tbName = tbName};
D
dapan1121 已提交
985 986 987 988
  char *msg = NULL;
  SEpSet *pVnodeEpSet = NULL;
  int32_t msgLen = 0;

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

D
dapan1121 已提交
991
  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_MND_TABLE_META)](&bInput, &msg, 0, &msgLen);
D
dapan1121 已提交
992 993 994 995
  if (code) {
    ctgError("Build mnode stablemeta msg failed, code:%x", code);
    CTG_ERR_RET(code);
  }
D
dapan1121 已提交
996 997

  SRpcMsg rpcMsg = {
D
dapan1121 已提交
998
      .msgType = TDMT_MND_TABLE_META,
D
dapan1121 已提交
999 1000 1001
      .pCont   = msg,
      .contLen = msgLen,
  };
D
dapan1121 已提交
1002

D
dapan1121 已提交
1003 1004
  SRpcMsg rpcRsp = {0};

D
dapan1121 已提交
1005
  rpcSendRecv(pTrans, (SEpSet*)pMgmtEps, &rpcMsg, &rpcRsp);
D
dapan1121 已提交
1006 1007
  
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
D
dapan1121 已提交
1008
    if (CTG_TABLE_NOT_EXIST(rpcRsp.code)) {
D
dapan1121 已提交
1009
      SET_META_TYPE_NULL(output->metaType);
D
dapan1121 已提交
1010
      ctgDebug("stablemeta not exist in mnode, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
1011 1012 1013
      return TSDB_CODE_SUCCESS;
    }
    
D
dapan1121 已提交
1014
    ctgError("error rsp for stablemeta from mnode, code:%s, dbFName:%s, tbName:%s", tstrerror(rpcRsp.code), dbFName, tbName);
D
dapan1121 已提交
1015 1016 1017
    CTG_ERR_RET(rpcRsp.code);
  }

D
dapan1121 已提交
1018
  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_TABLE_META)](output, rpcRsp.pCont, rpcRsp.contLen);
D
dapan1121 已提交
1019
  if (code) {
D
dapan1121 已提交
1020
    ctgError("Process mnode stablemeta rsp failed, code:%x, dbFName:%s, tbName:%s", code, dbFName, tbName);
D
dapan1121 已提交
1021 1022 1023
    CTG_ERR_RET(code);
  }

D
dapan1121 已提交
1024
  ctgDebug("Got table meta from mnode, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
1025 1026 1027 1028

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
1029
int32_t ctgGetTableMetaFromMnode(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, STableMetaOutput* output) {
D
dapan1121 已提交
1030 1031
  char dbFName[TSDB_DB_FNAME_LEN];
  tNameGetFullDbName(pTableName, dbFName);
D
dapan1121 已提交
1032

D
dapan1121 已提交
1033
  return ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, dbFName, (char *)pTableName->tname, output);
D
dapan1121 已提交
1034
}
D
dapan1121 已提交
1035

D
dapan1121 已提交
1036
int32_t ctgGetTableMetaFromVnodeImpl(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, SVgroupInfo *vgroupInfo, STableMetaOutput* output) {
D
dapan1121 已提交
1037
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pTableName || NULL == vgroupInfo || NULL == output) {
D
dapan1121 已提交
1038
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
1039 1040
  }

D
dapan1121 已提交
1041 1042
  char dbFName[TSDB_DB_FNAME_LEN];
  tNameGetFullDbName(pTableName, dbFName);
D
dapan1121 已提交
1043

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

D
dapan1121 已提交
1046
  SBuildTableMetaInput bInput = {.vgId = vgroupInfo->vgId, .dbFName = dbFName, .tbName = (char *)tNameGetTableName(pTableName)};
D
dapan1121 已提交
1047 1048 1049
  char *msg = NULL;
  int32_t msgLen = 0;

D
dapan1121 已提交
1050 1051
  int32_t code = queryBuildMsg[TMSG_INDEX(TDMT_VND_TABLE_META)](&bInput, &msg, 0, &msgLen);
  if (code) {
D
dapan1121 已提交
1052
    ctgError("Build vnode tablemeta msg failed, code:%x, dbFName:%s, tbName:%s", code, dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
1053 1054
    CTG_ERR_RET(code);
  }
D
dapan1121 已提交
1055 1056

  SRpcMsg rpcMsg = {
H
Hongze Cheng 已提交
1057
      .msgType = TDMT_VND_TABLE_META,
D
dapan1121 已提交
1058 1059 1060 1061 1062
      .pCont   = msg,
      .contLen = msgLen,
  };

  SRpcMsg rpcRsp = {0};
L
Liu Jicong 已提交
1063
  rpcSendRecv(pTrans, &vgroupInfo->epSet, &rpcMsg, &rpcRsp);
D
dapan1121 已提交
1064
  
D
dapan1121 已提交
1065
  if (TSDB_CODE_SUCCESS != rpcRsp.code) {
D
dapan1121 已提交
1066
    if (CTG_TABLE_NOT_EXIST(rpcRsp.code)) {
D
dapan1121 已提交
1067
      SET_META_TYPE_NULL(output->metaType);
D
dapan1121 已提交
1068
      ctgDebug("tablemeta not exist in vnode, dbFName:%s, tbName:%s", dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
1069 1070 1071
      return TSDB_CODE_SUCCESS;
    }
  
D
dapan1121 已提交
1072
    ctgError("error rsp for table meta from vnode, code:%s, dbFName:%s, tbName:%s", tstrerror(rpcRsp.code), dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
1073
    CTG_ERR_RET(rpcRsp.code);
D
dapan1121 已提交
1074 1075
  }

D
dapan1121 已提交
1076 1077
  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_VND_TABLE_META)](output, rpcRsp.pCont, rpcRsp.contLen);
  if (code) {
D
dapan1121 已提交
1078
    ctgError("Process vnode tablemeta rsp failed, code:%s, dbFName:%s, tbName:%s", tstrerror(code), dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
1079 1080 1081
    CTG_ERR_RET(code);
  }

D
dapan1121 已提交
1082
  ctgDebug("Got table meta from vnode, dbFName:%s, tbName:%s", dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
1083 1084 1085
  return TSDB_CODE_SUCCESS;
}

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

1113 1114
int32_t ctgGetHashFunction(int8_t hashMethod, tableNameHashFp *fp) {
  switch (hashMethod) {
D
dapan1121 已提交
1115 1116 1117 1118 1119 1120 1121 1122
    default:
      *fp = MurmurHash3_32;
      break;
  }

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
1123
int32_t ctgGenerateVgList(SCatalog *pCtg, SHashObj *vgHash, SArray** pList) {
D
dapan1121 已提交
1124
  SHashObj *vgroupHash = NULL;
1125
  SVgroupInfo *vgInfo = NULL;
D
dapan1121 已提交
1126 1127
  SArray *vgList = NULL;
  int32_t code = 0;
D
dapan1121 已提交
1128
  int32_t vgNum = taosHashGetSize(vgHash);
1129

D
dapan1121 已提交
1130
  vgList = taosArrayInit(vgNum, sizeof(SVgroupInfo));
D
dapan1121 已提交
1131
  if (NULL == vgList) {
D
dapan1121 已提交
1132
    ctgError("taosArrayInit failed, num:%d", vgNum);
D
dapan 已提交
1133 1134 1135
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);    
  }

D
dapan1121 已提交
1136
  void *pIter = taosHashIterate(vgHash, NULL);
1137 1138
  while (pIter) {
    vgInfo = pIter;
D
dapan1121 已提交
1139

D
dapan1121 已提交
1140
    if (NULL == taosArrayPush(vgList, vgInfo)) {
D
dapan1121 已提交
1141
      ctgError("taosArrayPush failed, vgId:%d", vgInfo->vgId);
D
dapan1121 已提交
1142
      taosHashCancelIterate(vgHash, pIter);      
D
dapan1121 已提交
1143
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
1144 1145
    }
    
D
dapan1121 已提交
1146
    pIter = taosHashIterate(vgHash, pIter);
1147
    vgInfo = NULL;
D
dapan1121 已提交
1148 1149
  }

D
dapan1121 已提交
1150
  *pList = vgList;
D
dapan1121 已提交
1151

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

D
dapan1121 已提交
1154
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1155 1156 1157 1158 1159 1160 1161 1162

_return:

  if (vgList) {
    taosArrayDestroy(vgList);
  }

  CTG_RET(code);
D
dapan1121 已提交
1163 1164
}

D
dapan1121 已提交
1165
int32_t ctgGetVgInfoFromHashValue(SCatalog *pCtg, SDBVgInfo *dbInfo, const SName *pTableName, SVgroupInfo *pVgroup) {
D
dapan1121 已提交
1166 1167
  int32_t code = 0;
  
D
dapan1121 已提交
1168
  int32_t vgNum = taosHashGetSize(dbInfo->vgHash);
H
Haojun Liao 已提交
1169 1170 1171
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);

1172
  if (vgNum <= 0) {
D
dapan1121 已提交
1173
    ctgError("db vgroup cache invalid, db:%s, vgroup number:%d", db, vgNum);
D
dapan1121 已提交
1174
    CTG_ERR_RET(TSDB_CODE_TSC_DB_NOT_SELECTED);
D
dapan1121 已提交
1175 1176
  }

1177 1178
  tableNameHashFp fp = NULL;
  SVgroupInfo *vgInfo = NULL;
D
dapan1121 已提交
1179

D
dapan1121 已提交
1180
  CTG_ERR_RET(ctgGetHashFunction(dbInfo->hashMethod, &fp));
1181 1182

  char tbFullName[TSDB_TABLE_FNAME_LEN];
H
Haojun Liao 已提交
1183
  tNameExtractFullName(pTableName, tbFullName);
1184 1185 1186

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

D
dapan1121 已提交
1187
  void *pIter = taosHashIterate(dbInfo->vgHash, NULL);
1188 1189 1190
  while (pIter) {
    vgInfo = pIter;
    if (hashValue >= vgInfo->hashBegin && hashValue <= vgInfo->hashEnd) {
D
dapan1121 已提交
1191
      taosHashCancelIterate(dbInfo->vgHash, pIter);
1192
      break;
D
dapan1121 已提交
1193
    }
1194
    
D
dapan1121 已提交
1195
    pIter = taosHashIterate(dbInfo->vgHash, pIter);
1196
    vgInfo = NULL;
D
dapan1121 已提交
1197 1198
  }

1199
  if (NULL == vgInfo) {
D
dapan1121 已提交
1200 1201
    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);
1202 1203 1204 1205
  }

  *pVgroup = *vgInfo;

1206
  CTG_RET(code);
D
dapan1121 已提交
1207 1208
}

D
dapan1121 已提交
1209
int32_t ctgStbVersionSearchCompare(const void* key1, const void* key2) {
D
dapan 已提交
1210
  if (*(uint64_t *)key1 < ((SSTableMetaVersion*)key2)->suid) {
D
dapan1121 已提交
1211
    return -1;
D
dapan 已提交
1212
  } else if (*(uint64_t *)key1 > ((SSTableMetaVersion*)key2)->suid) {
D
dapan1121 已提交
1213 1214 1215 1216 1217 1218
    return 1;
  } else {
    return 0;
  }
}

D
dapan1121 已提交
1219
int32_t ctgDbVgVersionSearchCompare(const void* key1, const void* key2) {
D
dapan1121 已提交
1220
  if (*(int64_t *)key1 < ((SDbVgVersion*)key2)->dbId) {
D
dapan1121 已提交
1221
    return -1;
D
dapan1121 已提交
1222
  } else if (*(int64_t *)key1 > ((SDbVgVersion*)key2)->dbId) {
D
dapan1121 已提交
1223 1224 1225
    return 1;
  } else {
    return 0;
D
dapan1121 已提交
1226
  }
D
dapan1121 已提交
1227 1228
}

D
dapan1121 已提交
1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249
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 已提交
1250
int32_t ctgMetaRentInit(SCtgRentMgmt *mgmt, uint32_t rentSec, int8_t type) {
D
dapan1121 已提交
1251 1252 1253 1254
  mgmt->slotRIdx = 0;
  mgmt->slotNum = rentSec / CTG_RENT_SLOT_SECOND;
  mgmt->type = type;

D
dapan1121 已提交
1255
  size_t msgSize = sizeof(SCtgRentSlot) * mgmt->slotNum;
D
dapan1121 已提交
1256
  
wafwerar's avatar
wafwerar 已提交
1257
  mgmt->slots = taosMemoryCalloc(1, msgSize);
D
dapan1121 已提交
1258 1259
  if (NULL == mgmt->slots) {
    qError("calloc %d failed", (int32_t)msgSize);
D
dapan 已提交
1260
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1261
  }
D
dapan1121 已提交
1262

D
dapan1121 已提交
1263 1264 1265 1266
  qDebug("meta rent initialized, type:%d, slotNum:%d", type, mgmt->slotNum);
  
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
1267

D
dapan1121 已提交
1268

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

D
dapan1121 已提交
1272
  SCtgRentSlot *slot = &mgmt->slots[widx];
D
dapan1121 已提交
1273 1274 1275 1276 1277 1278 1279 1280
  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 已提交
1281
    }
D
dapan1121 已提交
1282
  }
D
dapan1121 已提交
1283

D
dapan1121 已提交
1284 1285 1286
  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 已提交
1287 1288
  }

D
dapan1121 已提交
1289
  slot->needSort = true;
D
dapan1121 已提交
1290

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

D
dapan1121 已提交
1293 1294 1295 1296 1297 1298
_return:

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

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

D
dapan1121 已提交
1302
  SCtgRentSlot *slot = &mgmt->slots[widx];
D
dapan1121 已提交
1303
  int32_t code = 0;
1304

D
dapan1121 已提交
1305 1306
  CTG_LOCK(CTG_WRITE, &slot->lock);
  if (NULL == slot->meta) {
D
dapan1121 已提交
1307 1308
    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 已提交
1309 1310 1311
  }

  if (slot->needSort) {
D
dapan1121 已提交
1312
    qDebug("meta slot before sorte, slot idx:%d, type:%d, size:%d", widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta));
D
dapan1121 已提交
1313
    taosArraySort(slot->meta, sortCompare);
D
dapan1121 已提交
1314
    slot->needSort = false;
D
dapan1121 已提交
1315
    qDebug("meta slot sorted, slot idx:%d, type:%d, size:%d", widx, mgmt->type, (int32_t)taosArrayGetSize(slot->meta));
D
dapan1121 已提交
1316 1317
  }

D
dapan1121 已提交
1318
  void *orig = taosArraySearch(slot->meta, &id, searchCompare, TD_EQ);
D
dapan1121 已提交
1319
  if (NULL == orig) {
D
dapan1121 已提交
1320
    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 已提交
1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339
    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 已提交
1340
int32_t ctgMetaRentRemove(SCtgRentMgmt *mgmt, int64_t id, __compar_fn_t sortCompare, __compar_fn_t searchCompare) {
1341
  int16_t widx = abs((int)(id % mgmt->slotNum));
D
dapan1121 已提交
1342

D
dapan1121 已提交
1343
  SCtgRentSlot *slot = &mgmt->slots[widx];
D
dapan1121 已提交
1344 1345 1346 1347 1348 1349 1350 1351 1352
  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 已提交
1353
    taosArraySort(slot->meta, sortCompare);
D
dapan1121 已提交
1354 1355 1356 1357
    slot->needSort = false;
    qDebug("meta slot sorted, slot idx:%d, type:%d", widx, mgmt->type);
  }

D
dapan1121 已提交
1358
  int32_t idx = taosArraySearchIdx(slot->meta, &id, searchCompare, TD_EQ);
D
dapan1121 已提交
1359 1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375
  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 已提交
1376
int32_t ctgMetaRentGetImpl(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size) {
D
dapan1121 已提交
1377 1378 1379 1380
  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 已提交
1381
  }
D
dapan1121 已提交
1382

D
dapan1121 已提交
1383
  SCtgRentSlot *slot = &mgmt->slots[ridx];
D
dapan1121 已提交
1384
  int32_t code = 0;
D
dapan1121 已提交
1385
  
D
dapan1121 已提交
1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400
  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 已提交
1401
  *res = taosMemoryMalloc(msize);
D
dapan1121 已提交
1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421
  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 已提交
1422
int32_t ctgMetaRentGet(SCtgRentMgmt *mgmt, void **res, uint32_t *num, int32_t size) {
D
dapan1121 已提交
1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441
  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 已提交
1442 1443 1444
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
1445
int32_t ctgAddNewDBCache(SCatalog *pCtg, const char *dbFName, uint64_t dbId) {
D
dapan1121 已提交
1446
  int32_t code = 0;
D
dapan1121 已提交
1447

D
dapan1121 已提交
1448 1449 1450
  SCtgDBCache newDBCache = {0};
  newDBCache.dbId = dbId;

D
dapan 已提交
1451
  newDBCache.tbCache.metaCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
D
dapan1121 已提交
1452
  if (NULL == newDBCache.tbCache.metaCache) {
D
dapan 已提交
1453
    ctgError("taosHashInit %d metaCache failed", gCtgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
1454 1455 1456
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

D
dapan 已提交
1457
  newDBCache.tbCache.stbCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), true, HASH_ENTRY_LOCK);
D
dapan1121 已提交
1458
  if (NULL == newDBCache.tbCache.stbCache) {
D
dapan 已提交
1459
    ctgError("taosHashInit %d stbCache failed", gCtgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
1460 1461 1462 1463
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
  }

  code = taosHashPut(pCtg->dbCache, dbFName, strlen(dbFName), &newDBCache, sizeof(SCtgDBCache));
D
dapan1121 已提交
1464 1465 1466
  if (code) {
    if (HASH_NODE_EXIST(code)) {
      ctgDebug("db already in cache, dbFName:%s", dbFName);
D
dapan1121 已提交
1467
      goto _return;
D
dapan1121 已提交
1468 1469 1470
    }
    
    ctgError("taosHashPut db to cache failed, dbFName:%s", dbFName);
D
dapan1121 已提交
1471 1472
    CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
  }
D
dapan1121 已提交
1473 1474

  CTG_CACHE_STAT_ADD(dbNum, 1);
D
dapan1121 已提交
1475
 
D
dapan1121 已提交
1476
  SDbVgVersion vgVersion = {.dbId = newDBCache.dbId, .vgVersion = -1};
D
dapan1121 已提交
1477 1478
  strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));

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

D
dapan1121 已提交
1481 1482 1483
  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 已提交
1484

D
dapan1121 已提交
1485
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1486

D
dapan1121 已提交
1487
_return:
D
dapan1121 已提交
1488

D
dapan1121 已提交
1489
  ctgFreeDbCache(&newDBCache);
D
dapan1121 已提交
1490

D
dapan1121 已提交
1491 1492
  CTG_RET(code);
}
D
dapan1121 已提交
1493

D
dapan1121 已提交
1494

D
dapan1121 已提交
1495
void ctgRemoveStbRent(SCatalog* pCtg, SCtgTbMetaCache *cache) {
D
dapan1121 已提交
1496 1497 1498 1499 1500
  CTG_LOCK(CTG_WRITE, &cache->stbLock);
  if (cache->stbCache) {
    void *pIter = taosHashIterate(cache->stbCache, NULL);
    while (pIter) {
      uint64_t *suid = NULL;
H
Haojun Liao 已提交
1501
      suid = taosHashGetKey(pIter, NULL);
D
dapan1121 已提交
1502

D
dapan1121 已提交
1503
      if (TSDB_CODE_SUCCESS == ctgMetaRentRemove(&pCtg->stbRent, *suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare)) {
D
dapan1121 已提交
1504 1505 1506 1507 1508 1509 1510 1511 1512 1513
        ctgDebug("stb removed from rent, suid:%"PRIx64, *suid);
      }
          
      pIter = taosHashIterate(cache->stbCache, pIter);
    }
  }
  CTG_UNLOCK(CTG_WRITE, &cache->stbLock);
}


D
dapan1121 已提交
1514
int32_t ctgRemoveDB(SCatalog* pCtg, SCtgDBCache *dbCache, const char* dbFName) {
D
dapan 已提交
1515 1516 1517 1518
  uint64_t dbId = dbCache->dbId;
  
  ctgInfo("start to remove db from cache, dbFName:%s, dbId:%"PRIx64, dbFName, dbCache->dbId);

D
dapan1121 已提交
1519 1520 1521 1522 1523 1524
  atomic_store_8(&dbCache->deleted, 1);

  ctgRemoveStbRent(pCtg, &dbCache->tbCache);

  ctgFreeDbCache(dbCache);

D
dapan1121 已提交
1525
  CTG_ERR_RET(ctgMetaRentRemove(&pCtg->dbRent, dbCache->dbId, ctgDbVgVersionSortCompare, ctgDbVgVersionSearchCompare));
D
dapan1121 已提交
1526 1527 1528 1529
  
  ctgDebug("db removed from rent, dbFName:%s, dbId:%"PRIx64, dbFName, dbCache->dbId);

  if (taosHashRemove(pCtg->dbCache, dbFName, strlen(dbFName))) {
D
dapan1121 已提交
1530 1531
    ctgInfo("taosHashRemove from dbCache failed, may be removed, dbFName:%s", dbFName);
    CTG_ERR_RET(TSDB_CODE_CTG_DB_DROPPED);
D
dapan1121 已提交
1532
  }
D
dapan 已提交
1533

D
dapan1121 已提交
1534 1535
  CTG_CACHE_STAT_SUB(dbNum, 1);

D
dapan 已提交
1536
  ctgInfo("db removed from cache, dbFName:%s, dbId:%"PRIx64, dbFName, dbId);
D
dapan1121 已提交
1537 1538 1539
  
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
1540 1541


D
dapan1121 已提交
1542
int32_t ctgGetAddDBCache(SCatalog* pCtg, const char *dbFName, uint64_t dbId, SCtgDBCache **pCache) {
D
dapan1121 已提交
1543 1544
  int32_t code = 0;
  SCtgDBCache *dbCache = NULL;
D
dapan1121 已提交
1545
  ctgGetDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
1546
  
D
dapan1121 已提交
1547 1548
  if (dbCache) {
  // TODO OPEN IT
D
dapan1121 已提交
1549
#if 0    
D
dapan1121 已提交
1550 1551 1552 1553
    if (dbCache->dbId == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
D
dapan1121 已提交
1554
#else
D
dapan1121 已提交
1555 1556 1557
    if (0 == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1558 1559
    }

D
dapan1121 已提交
1560 1561 1562 1563 1564
    if (dbId && (dbCache->dbId == 0)) {
      dbCache->dbId = dbId;
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
D
dapan1121 已提交
1565
    
D
dapan1121 已提交
1566 1567 1568 1569 1570 1571
    if (dbCache->dbId == dbId) {
      *pCache = dbCache;
      return TSDB_CODE_SUCCESS;
    }
#endif
    CTG_ERR_RET(ctgRemoveDB(pCtg, dbCache, dbFName));
D
dapan1121 已提交
1572
  }
D
dapan1121 已提交
1573 1574
  
  CTG_ERR_RET(ctgAddNewDBCache(pCtg, dbFName, dbId));
D
dapan1121 已提交
1575

D
dapan1121 已提交
1576
  ctgGetDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
1577

D
dapan1121 已提交
1578
  *pCache = dbCache;
D
dapan1121 已提交
1579

D
dapan1121 已提交
1580
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1581 1582 1583
}


D
dapan1121 已提交
1584 1585 1586
int32_t ctgUpdateDBVgInfo(SCatalog* pCtg, const char* dbFName, uint64_t dbId, SDBVgInfo** pDbInfo) {
  int32_t code = 0;
  SDBVgInfo* dbInfo = *pDbInfo;
D
dapan1121 已提交
1587 1588 1589 1590

  if (NULL == dbInfo->vgHash) {
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
1591
  
D
dapan1121 已提交
1592
  if (dbInfo->vgVersion < 0 || taosHashGetSize(dbInfo->vgHash) <= 0) {
D
dapan1121 已提交
1593 1594
    ctgError("invalid db vgInfo, dbFName:%s, vgHash:%p, vgVersion:%d, vgHashSize:%d", 
      dbFName, dbInfo->vgHash, dbInfo->vgVersion, taosHashGetSize(dbInfo->vgHash));
D
dapan1121 已提交
1595
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1596 1597
  }

D
dapan1121 已提交
1598
  bool newAdded = false;
D
dapan 已提交
1599
  SDbVgVersion vgVersion = {.dbId = dbId, .vgVersion = dbInfo->vgVersion, .numOfTable = dbInfo->numOfTable};
D
dapan1121 已提交
1600 1601 1602 1603 1604 1605 1606 1607 1608 1609

  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 已提交
1610
  
D
dapan1121 已提交
1611
  if (dbCache->vgInfo) {
D
dapan 已提交
1612 1613 1614 1615 1616 1617 1618 1619 1620
    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 已提交
1621
      ctgWReleaseVgInfo(dbCache);
D
dapan1121 已提交
1622
      
D
dapan1121 已提交
1623
      return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1624
    }
D
dapan1121 已提交
1625 1626

    ctgFreeVgInfo(dbCache->vgInfo);
D
dapan1121 已提交
1627 1628
  }

D
dapan1121 已提交
1629
  dbCache->vgInfo = dbInfo;
D
dapan1121 已提交
1630

D
dapan1121 已提交
1631
  *pDbInfo = NULL;
D
dapan1121 已提交
1632

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

D
dapan1121 已提交
1635
  ctgWReleaseVgInfo(dbCache);
D
dapan1121 已提交
1636

D
dapan1121 已提交
1637
  dbCache = NULL;
D
dapan1121 已提交
1638

D
dapan1121 已提交
1639
  strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));
D
dapan1121 已提交
1640
  CTG_ERR_RET(ctgMetaRentUpdate(&pCtg->dbRent, &vgVersion, vgVersion.dbId, sizeof(SDbVgVersion), ctgDbVgVersionSortCompare, ctgDbVgVersionSearchCompare));
D
dapan1121 已提交
1641
  
D
dapan1121 已提交
1642 1643 1644 1645
  CTG_RET(code);
}


D
dapan1121 已提交
1646 1647
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 已提交
1648

D
dapan1121 已提交
1649 1650 1651 1652 1653
  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 已提交
1654 1655
  }

D
dapan1121 已提交
1656 1657 1658 1659 1660 1661 1662
  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 已提交
1663 1664 1665 1666 1667
    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);
D
dapan1121 已提交
1668 1669
        } else {
          CTG_CACHE_STAT_SUB(stblNum, 1);
D
dapan1121 已提交
1670 1671 1672 1673 1674
        }
        CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);

        ctgDebug("stb removed from stbCache, dbFName:%s, stb:%s, suid:%"PRIx64, dbFName, tbName, orig->suid);
        
D
dapan1121 已提交
1675
        ctgMetaRentRemove(&pCtg->stbRent, orig->suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare);
D
dapan1121 已提交
1676
      }
D
dapan1121 已提交
1677

D
dapan1121 已提交
1678
      origSuid = orig->suid;
D
dapan1121 已提交
1679
    }
D
dapan1121 已提交
1680
  }
D
dapan1121 已提交
1681

D
dapan1121 已提交
1682 1683
  if (isStb) {
    CTG_LOCK(CTG_WRITE, &tbCache->stbLock);
D
dapan1121 已提交
1684
  }
D
dapan1121 已提交
1685
  
D
dapan1121 已提交
1686 1687 1688 1689 1690 1691 1692 1693 1694
  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 已提交
1695

D
dapan1121 已提交
1696 1697 1698 1699
  if (NULL == orig) {
    CTG_CACHE_STAT_ADD(tblNum, 1);
  }

D
dapan 已提交
1700
  ctgDebug("tbmeta updated to cache, dbFName:%s, tbName:%s, tbType:%d", dbFName, tbName, meta->tableType);
D
dapan1121 已提交
1701
  ctgdShowTableMeta(pCtg, tbName, meta);
D
dapan 已提交
1702

D
dapan1121 已提交
1703 1704 1705
  if (!isStb) {
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);  
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1706
  }
D
dapan1121 已提交
1707

D
dapan1121 已提交
1708
  if (origType == TSDB_SUPER_TABLE && origSuid == meta->suid) {
D
dapan1121 已提交
1709 1710 1711 1712
    CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);  
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
1713

D
dapan1121 已提交
1714 1715 1716 1717
  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 已提交
1718
    ctgError("taosHashPut stable to stable cache failed, suid:%"PRIx64, meta->suid);
D
dapan1121 已提交
1719
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
1720
  }
D
dapan1121 已提交
1721 1722

  CTG_CACHE_STAT_ADD(stblNum, 1);
D
dapan1121 已提交
1723 1724
  
  CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);
D
dapan1121 已提交
1725

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

D
dapan 已提交
1728
  ctgDebug("stb updated to stbCache, dbFName:%s, tbName:%s, tbType:%d", dbFName, tbName, meta->tableType);
D
dapan1121 已提交
1729 1730 1731 1732 1733 1734 1735

  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 已提交
1736 1737
}

D
dapan 已提交
1738
int32_t ctgCloneVgInfo(SDBVgInfo *src, SDBVgInfo **dst) {
wafwerar's avatar
wafwerar 已提交
1739
  *dst = taosMemoryMalloc(sizeof(SDBVgInfo));
D
dapan 已提交
1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750
  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 已提交
1751
    taosMemoryFreeClear(*dst);
D
dapan 已提交
1752 1753 1754 1755 1756 1757
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }

  int32_t *vgId = NULL;
  void *pIter = taosHashIterate(src->vgHash, NULL);
  while (pIter) {
H
Haojun Liao 已提交
1758
    vgId = taosHashGetKey(pIter, NULL);
D
dapan 已提交
1759 1760 1761 1762 1763

    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 已提交
1764
      taosMemoryFreeClear(*dst);
D
dapan 已提交
1765 1766 1767 1768 1769 1770 1771 1772 1773 1774
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }
    
    pIter = taosHashIterate(src->vgHash, pIter);
  }


  return TSDB_CODE_SUCCESS;
}

1775
void ctgGetDBVgInfo_a();
D
dapan 已提交
1776

D
dapan1121 已提交
1777
int32_t ctgGetDBVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* dbFName, SCtgDBCache** dbCache, SDBVgInfo **pInfo) {
D
dapan1121 已提交
1778
  bool inCache = false;
D
dapan1121 已提交
1779
  int32_t code = 0;
D
dapan1121 已提交
1780 1781 1782

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

D
dapan1121 已提交
1783
  if (inCache) {
D
dapan1121 已提交
1784
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1785 1786 1787 1788 1789
  }

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

D
dapan1121 已提交
1790
  tstrncpy(input.db, dbFName, tListLen(input.db));
D
dapan1121 已提交
1791
  input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
H
Haojun Liao 已提交
1792

D
dapan1121 已提交
1793 1794 1795 1796 1797 1798 1799 1800 1801
  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 已提交
1802

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

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

D
dapan1121 已提交
1807
  return TSDB_CODE_SUCCESS;
D
dapan 已提交
1808 1809 1810

_return:

wafwerar's avatar
wafwerar 已提交
1811
  taosMemoryFreeClear(*pInfo);
D
dapan 已提交
1812 1813 1814
  *pInfo = DbOut.dbVgroup;
  
  CTG_RET(code);
D
dapan1121 已提交
1815 1816
}

D
dapan1121 已提交
1817 1818 1819
int32_t ctgRefreshDBVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* dbFName) {
  bool inCache = false;
  int32_t code = 0;
D
dapan1121 已提交
1820
  SCtgDBCache* dbCache = NULL;
D
dapan1121 已提交
1821

D
dapan1121 已提交
1822
  CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache, &inCache));
D
dapan1121 已提交
1823 1824 1825 1826 1827 1828

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

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

D
dapan1121 已提交
1831 1832
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
1833
  }
D
dapan1121 已提交
1834 1835 1836
  
  input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
  input.numOfTable = 0;
D
dapan1121 已提交
1837 1838 1839

  code = ctgGetDBVgInfoFromMnode(pCtg, pRpc, pMgmtEps, &input, &DbOut);
  if (code) {
D
dapan1121 已提交
1840
    if (CTG_DB_NOT_EXIST(code) && inCache) {
D
dapan1121 已提交
1841 1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853
      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 已提交
1854

D
dapan1121 已提交
1855
int32_t ctgCloneMetaOutput(STableMetaOutput *output, STableMetaOutput **pOutput) {
wafwerar's avatar
wafwerar 已提交
1856
  *pOutput = taosMemoryMalloc(sizeof(STableMetaOutput));
D
dapan1121 已提交
1857 1858 1859
  if (NULL == *pOutput) {
    qError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan 已提交
1860 1861
  }

D
dapan1121 已提交
1862 1863 1864 1865
  memcpy(*pOutput, output, sizeof(STableMetaOutput));

  if (output->tbMeta) {
    int32_t metaSize = CTG_META_SIZE(output->tbMeta);
wafwerar's avatar
wafwerar 已提交
1866
    (*pOutput)->tbMeta = taosMemoryMalloc(metaSize);
D
dapan1121 已提交
1867
    if (NULL == (*pOutput)->tbMeta) {
D
dapan 已提交
1868
      qError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
wafwerar's avatar
wafwerar 已提交
1869
      taosMemoryFreeClear(*pOutput);
D
dapan1121 已提交
1870 1871 1872 1873
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }

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

D
dapan1121 已提交
1876 1877 1878
  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
1879 1880


D
dapan1121 已提交
1881
int32_t ctgRefreshTblMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, int32_t flag, STableMetaOutput **pOutput, bool syncReq) {
D
dapan1121 已提交
1882
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pTableName) {
D
dapan1121 已提交
1883 1884 1885 1886 1887 1888
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }

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

D
dapan1121 已提交
1889
  if (!CTG_FLAG_IS_SYS_DB(flag)) {
D
dapan1121 已提交
1890 1891
    CTG_ERR_RET(catalogGetTableHashVgroup(pCtg, pTrans, pMgmtEps, pTableName, &vgroupInfo));
  }
D
dapan1121 已提交
1892

D
dapan1121 已提交
1893
  STableMetaOutput  moutput = {0};
wafwerar's avatar
wafwerar 已提交
1894
  STableMetaOutput *output = taosMemoryCalloc(1, sizeof(STableMetaOutput));
D
dapan1121 已提交
1895 1896 1897 1898
  if (NULL == output) {
    ctgError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
  }
D
dapan1121 已提交
1899

D
dapan1121 已提交
1900
  if (CTG_FLAG_IS_SYS_DB(flag)) {
D
dapan1121 已提交
1901 1902 1903
    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 已提交
1904
  } else if (CTG_FLAG_IS_STB(flag)) {
D
dapan 已提交
1905
    ctgDebug("will refresh tbmeta, supposed to be stb, tbName:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
1906 1907

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

D
dapan1121 已提交
1910
    if (CTG_IS_META_NULL(output->metaType)) {
D
dapan1121 已提交
1911
      CTG_ERR_JRET(ctgGetTableMetaFromVnode(pCtg, pTrans, pMgmtEps, pTableName, &vgroupInfo, output));
D
dapan1121 已提交
1912 1913
    }
  } else {
D
dapan1121 已提交
1914
    ctgDebug("will refresh tbmeta, not supposed to be stb, tbName:%s, flag:%d", tNameGetTableName(pTableName), flag);
D
dapan1121 已提交
1915 1916

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

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

wafwerar's avatar
wafwerar 已提交
1922
      taosMemoryFreeClear(output->tbMeta);
D
dapan1121 已提交
1923
      
D
dapan1121 已提交
1924
      CTG_ERR_JRET(ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, output->dbFName, output->tbName, output));
D
dapan1121 已提交
1925
    } else if (CTG_IS_META_BOTH(output->metaType)) {
D
dapan1121 已提交
1926
      int32_t exist = 0;
D
dapan1121 已提交
1927 1928 1929
      if (!CTG_FLAG_IS_FORCE_UPDATE(flag)) {
        CTG_ERR_JRET(ctgIsTableMetaExistInCache(pCtg, output->dbFName, output->tbName, &exist));
      }
H
Haojun Liao 已提交
1930

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

D
dapan1121 已提交
1934
        if (CTG_IS_META_NULL(moutput.metaType)) {
D
dapan1121 已提交
1935
          SET_META_TYPE_NULL(output->metaType);
D
dapan1121 已提交
1936 1937
        }
        
wafwerar's avatar
wafwerar 已提交
1938
        taosMemoryFreeClear(output->tbMeta);
D
dapan1121 已提交
1939
        output->tbMeta = moutput.tbMeta;
D
dapan1121 已提交
1940 1941
        moutput.tbMeta = NULL;
      } else {
wafwerar's avatar
wafwerar 已提交
1942
        taosMemoryFreeClear(output->tbMeta);
D
dapan1121 已提交
1943
        
D
dapan1121 已提交
1944
        SET_META_TYPE_CTABLE(output->metaType); 
D
dapan1121 已提交
1945
      }
D
dapan1121 已提交
1946 1947 1948
    }
  }

D
dapan1121 已提交
1949
  if (CTG_IS_META_NULL(output->metaType)) {
D
dapan 已提交
1950
    ctgError("no tbmeta got, tbNmae:%s", tNameGetTableName(pTableName));
D
dapan1121 已提交
1951
    catalogRemoveTableMeta(pCtg, pTableName);
D
dapan1121 已提交
1952 1953 1954
    CTG_ERR_JRET(CTG_ERR_CODE_TABLE_NOT_EXIST);
  }

D
dapan 已提交
1955 1956 1957 1958 1959 1960
  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 已提交
1961 1962 1963 1964
  if (pOutput) {
    CTG_ERR_JRET(ctgCloneMetaOutput(output, pOutput));
  }

D
dapan1121 已提交
1965
  CTG_ERR_JRET(ctgPushUpdateTblMsgInQueue(pCtg, output, syncReq));
D
dapan 已提交
1966

D
dapan1121 已提交
1967
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1968 1969 1970

_return:

wafwerar's avatar
wafwerar 已提交
1971 1972
  taosMemoryFreeClear(output->tbMeta);
  taosMemoryFreeClear(output);
D
dapan1121 已提交
1973 1974 1975 1976
  
  CTG_RET(code);
}

D
dapan1121 已提交
1977
int32_t ctgGetTableMeta(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, STableMeta** pTableMeta, int32_t flag) {
D
dapan1121 已提交
1978
  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == pTableMeta) {
D
dapan1121 已提交
1979 1980
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
1981

D
dapan1121 已提交
1982
  bool inCache = false;
D
dapan1121 已提交
1983
  int32_t code = 0;
D
dapan1121 已提交
1984 1985 1986
  uint64_t dbId = 0;
  uint64_t suid = 0;
  STableMetaOutput *output = NULL;
D
dapan1121 已提交
1987

D
dapan1121 已提交
1988 1989
  if (CTG_IS_SYS_DBNAME(pTableName->dbname)) {
    CTG_FLAG_SET_SYS_DB(flag);
D
dapan1121 已提交
1990 1991
  }

D
dapan1121 已提交
1992
  CTG_ERR_RET(ctgGetTableMetaFromCache(pCtg, pTableName, pTableMeta, &inCache, flag, &dbId));
D
dapan1121 已提交
1993

D
dapan1121 已提交
1994 1995
  int32_t tbType = 0;

D
dapan1121 已提交
1996
  if (inCache) {
D
dapan1121 已提交
1997
    if (CTG_FLAG_MATCH_STB(flag, (*pTableMeta)->tableType) && ((!CTG_FLAG_IS_FORCE_UPDATE(flag)) || (CTG_FLAG_IS_SYS_DB(flag)))) {
D
dapan1121 已提交
1998
      goto _return;
D
dapan1121 已提交
1999
    }
D
dapan1121 已提交
2000

D
dapan1121 已提交
2001 2002
    tbType = (*pTableMeta)->tableType;
    suid = (*pTableMeta)->suid;
D
dapan1121 已提交
2003

wafwerar's avatar
wafwerar 已提交
2004
    taosMemoryFreeClear(*pTableMeta);
D
dapan1121 已提交
2005
  }
D
dapan1121 已提交
2006

D
dapan1121 已提交
2007 2008
  if (CTG_FLAG_IS_UNKNOWN_STB(flag)) {
    CTG_FLAG_SET_STB(flag, tbType);
D
dapan1121 已提交
2009 2010
  }

D
dapan1121 已提交
2011

D
dapan 已提交
2012
  while (true) {
D
dapan1121 已提交
2013
    CTG_ERR_JRET(ctgRefreshTblMeta(pCtg, pRpc, pMgmtEps, pTableName, flag, &output, false));
D
dapan1121 已提交
2014

D
dapan 已提交
2015 2016 2017 2018
    if (CTG_IS_META_TABLE(output->metaType)) {
      *pTableMeta = output->tbMeta;
      goto _return;
    }
D
dapan1121 已提交
2019

D
dapan 已提交
2020 2021 2022 2023 2024 2025
    if (CTG_IS_META_BOTH(output->metaType)) {
      memcpy(output->tbMeta, &output->ctbMeta, sizeof(output->ctbMeta));
      
      *pTableMeta = output->tbMeta;
      goto _return;
    }
D
dapan1121 已提交
2026

D
dapan 已提交
2027 2028
    if ((!CTG_IS_META_CTABLE(output->metaType)) || output->tbMeta) {
      ctgError("invalid metaType:%d", output->metaType);
wafwerar's avatar
wafwerar 已提交
2029
      taosMemoryFreeClear(output->tbMeta);
D
dapan 已提交
2030 2031
      CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
    }
D
dapan1121 已提交
2032

D
dapan 已提交
2033
    // HANDLE ONLY CHILD TABLE META
D
dapan1121 已提交
2034

D
dapan 已提交
2035 2036 2037
    SName stbName = *pTableName;
    strcpy(stbName.tname, output->tbName);
    
D
dapan1121 已提交
2038 2039
    CTG_ERR_JRET(ctgGetTableMetaFromCache(pCtg, &stbName, pTableMeta, &inCache, flag, NULL));
    if (!inCache) {
D
dapan 已提交
2040 2041 2042
      ctgDebug("stb no longer exist, dbFName:%s, tbName:%s", output->dbFName, pTableName->tname);
      continue;
    }
D
dapan1121 已提交
2043

D
dapan 已提交
2044 2045 2046 2047
    memcpy(*pTableMeta, &output->ctbMeta, sizeof(output->ctbMeta));

    break;
  }
D
dapan1121 已提交
2048 2049 2050

_return:

D
dapan1121 已提交
2051
  if (CTG_TABLE_NOT_EXIST(code) && inCache) {
D
dapan1121 已提交
2052
    char dbFName[TSDB_DB_FNAME_LEN] = {0};
D
dapan1121 已提交
2053
    if (CTG_FLAG_IS_SYS_DB(flag)) {
D
dapan1121 已提交
2054 2055 2056 2057 2058 2059
      strcpy(dbFName, pTableName->dbname);
    } else {
      tNameGetFullDbName(pTableName, dbFName);
    }

    if (TSDB_SUPER_TABLE == tbType) {
D
dapan1121 已提交
2060
      ctgPushRmStbMsgInQueue(pCtg, dbFName, dbId, pTableName->tname, suid, false);
D
dapan1121 已提交
2061
    } else {
D
dapan1121 已提交
2062
      ctgPushRmTblMsgInQueue(pCtg, dbFName, dbId, pTableName->tname, false);
D
dapan1121 已提交
2063 2064 2065
    }
  }

wafwerar's avatar
wafwerar 已提交
2066
  taosMemoryFreeClear(output);
D
dapan1121 已提交
2067

D
dapan 已提交
2068 2069
  if (*pTableMeta) {
    ctgDebug("tbmeta returned, tbName:%s, tbType:%d", pTableName->tname, (*pTableMeta)->tableType);
D
dapan1121 已提交
2070
    ctgdShowTableMeta(pCtg, pTableName->tname, *pTableMeta);
D
dapan 已提交
2071 2072
  }

D
dapan1121 已提交
2073 2074 2075
  CTG_RET(code);
}

D
dapan 已提交
2076
int32_t ctgChkAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* user, const char* dbFName, AUTH_TYPE type, bool *pass) {
D
dapan 已提交
2077 2078 2079
  bool inCache = false;
  int32_t code = 0;
  
D
dapan 已提交
2080 2081 2082
  *pass = false;
  
  CTG_ERR_RET(ctgChkAuthFromCache(pCtg, user, dbFName, type, &inCache, pass));
D
dapan 已提交
2083 2084 2085 2086 2087 2088 2089 2090 2091

  if (inCache) {
    return TSDB_CODE_SUCCESS;
  }

  SGetUserAuthRsp authRsp = {0};
  CTG_ERR_RET(ctgGetUserDbAuthFromMnode(pCtg, pRpc, pMgmtEps, user, &authRsp));
  
  if (authRsp.superAuth) {
D
dapan 已提交
2092
    *pass = true;
D
dapan 已提交
2093 2094 2095 2096
    goto _return;
  }

  if (authRsp.createdDbs && taosHashGet(authRsp.createdDbs, dbFName, strlen(dbFName))) {
D
dapan 已提交
2097
    *pass = true;
D
dapan 已提交
2098 2099 2100
    goto _return;
  }

D
dapan 已提交
2101 2102
  if (authRsp.readDbs && taosHashGet(authRsp.readDbs, dbFName, strlen(dbFName)) && type == AUTH_TYPE_READ) {
    *pass = true;
D
dapan 已提交
2103 2104
  }

D
dapan 已提交
2105 2106
  if (authRsp.writeDbs && taosHashGet(authRsp.writeDbs, dbFName, strlen(dbFName)) && type == AUTH_TYPE_WRITE) {
    *pass = true;
D
dapan 已提交
2107 2108 2109 2110 2111 2112 2113 2114
  }

_return:

  ctgPushUpdateUserMsgInQueue(pCtg, &authRsp, false);

  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
2115 2116 2117 2118 2119 2120 2121 2122 2123 2124


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 已提交
2125
  ctgFreeVgInfo(msg->dbInfo);
wafwerar's avatar
wafwerar 已提交
2126
  taosMemoryFreeClear(msg);
D
dapan1121 已提交
2127 2128 2129 2130 2131 2132 2133 2134 2135 2136
  
  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 已提交
2137
  ctgGetDBCache(msg->pCtg, msg->dbFName, &dbCache);
D
dapan1121 已提交
2138 2139 2140 2141 2142 2143 2144 2145 2146 2147 2148 2149 2150
  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 已提交
2151
  taosMemoryFreeClear(msg);
D
dapan1121 已提交
2152 2153 2154 2155 2156 2157 2158 2159 2160 2161 2162 2163 2164 2165
  
  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 已提交
2166
    CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
2167 2168 2169 2170 2171 2172
  }

  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 已提交
2173
  
D
dapan1121 已提交
2174 2175 2176 2177 2178 2179 2180 2181 2182 2183 2184 2185 2186 2187 2188 2189 2190 2191
  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 已提交
2192
  if (output) {
wafwerar's avatar
wafwerar 已提交
2193 2194
    taosMemoryFreeClear(output->tbMeta);
    taosMemoryFreeClear(output);
D
dapan1121 已提交
2195
  }
D
dapan 已提交
2196
  
wafwerar's avatar
wafwerar 已提交
2197
  taosMemoryFreeClear(msg);
D
dapan1121 已提交
2198 2199 2200 2201 2202 2203 2204 2205 2206 2207 2208 2209 2210 2211 2212 2213
  
  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 已提交
2214
  if (msg->dbId && (dbCache->dbId != msg->dbId)) {
D
dapan1121 已提交
2215 2216 2217 2218 2219 2220 2221
    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);
D
dapan1121 已提交
2222 2223
  } else {
    CTG_CACHE_STAT_SUB(stblNum, 1);
D
dapan1121 已提交
2224 2225 2226 2227 2228
  }

  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);
D
dapan1121 已提交
2229 2230 2231
  } else {
    CTG_CACHE_STAT_SUB(tblNum, 1);
  }
D
dapan1121 已提交
2232 2233 2234 2235 2236 2237
  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 已提交
2238
  CTG_ERR_JRET(ctgMetaRentRemove(&msg->pCtg->stbRent, msg->suid, ctgStbVersionSortCompare, ctgStbVersionSearchCompare));
D
dapan1121 已提交
2239 2240 2241 2242 2243
  
  ctgDebug("stb removed from rent, dbFName:%s, stbName:%s, suid:%"PRIx64, msg->dbFName, msg->stbName, msg->suid);
  
_return:

wafwerar's avatar
wafwerar 已提交
2244
  taosMemoryFreeClear(msg);
D
dapan1121 已提交
2245 2246 2247 2248 2249
  
  CTG_RET(code);
}

int32_t ctgActRemoveTbl(SCtgMetaAction *action) {
D
dapan1121 已提交
2250 2251 2252 2253 2254 2255 2256 2257 2258 2259 2260 2261 2262 2263
  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 已提交
2264

D
dapan1121 已提交
2265
  CTG_LOCK(CTG_READ, &dbCache->tbCache.metaLock);
H
Haojun Liao 已提交
2266
  if (taosHashRemove(dbCache->tbCache.metaCache, msg->tbName, strlen(msg->tbName))) {
D
dapan1121 已提交
2267 2268 2269
    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);
D
dapan1121 已提交
2270 2271
  } else {
    CTG_CACHE_STAT_SUB(tblNum, 1);
H
Haojun Liao 已提交
2272
  }
D
dapan1121 已提交
2273
  CTG_UNLOCK(CTG_READ, &dbCache->tbCache.metaLock);
D
dapan1121 已提交
2274

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

D
dapan1121 已提交
2277
_return:
D
dapan1121 已提交
2278

wafwerar's avatar
wafwerar 已提交
2279
  taosMemoryFreeClear(msg);
H
Haojun Liao 已提交
2280

D
dapan1121 已提交
2281
  CTG_RET(code);
D
dapan1121 已提交
2282 2283
}

D
dapan 已提交
2284 2285 2286 2287 2288 2289 2290 2291 2292 2293 2294 2295 2296 2297 2298 2299 2300 2301 2302 2303 2304 2305 2306 2307 2308 2309 2310 2311 2312 2313 2314 2315 2316 2317 2318 2319 2320 2321 2322 2323 2324 2325 2326 2327 2328 2329 2330 2331 2332 2333 2334 2335 2336 2337 2338 2339 2340 2341 2342 2343 2344
int32_t ctgActUpdateUser(SCtgMetaAction *action) {
  int32_t code = 0;
  SCtgUpdateUserMsg *msg = action->data;
  SCatalog* pCtg = msg->pCtg;

  if (NULL == pCtg->userCache) {
    pCtg->userCache = taosHashInit(gCtgMgmt.cfg.maxUserCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
    if (NULL == pCtg->userCache) {
      ctgError("taosHashInit %d user cache failed", gCtgMgmt.cfg.maxUserCacheNum);
      CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
    }
  }
  
  SCtgUserAuth *pUser = (SCtgUserAuth *)taosHashGet(pCtg->userCache, msg->userAuth.user, strlen(msg->userAuth.user));
  if (NULL == pUser) {
    SCtgUserAuth userAuth = {0};

    userAuth.version = msg->userAuth.version;
    userAuth.superUser = msg->userAuth.superAuth;
    userAuth.createdDbs = msg->userAuth.createdDbs;
    userAuth.readDbs = msg->userAuth.readDbs;
    userAuth.writeDbs = msg->userAuth.writeDbs;

    if (taosHashPut(pCtg->userCache, msg->userAuth.user, strlen(msg->userAuth.user), &userAuth, sizeof(userAuth))) {
      ctgError("taosHashPut user %s to cache failed", msg->userAuth.user);
      CTG_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY);
    }

    return TSDB_CODE_SUCCESS;
  }

  pUser->version = msg->userAuth.version;

  CTG_LOCK(CTG_WRITE, &pUser->lock);

  taosHashCleanup(pUser->createdDbs);
  pUser->createdDbs = msg->userAuth.createdDbs;
  msg->userAuth.createdDbs = NULL;

  taosHashCleanup(pUser->readDbs);
  pUser->readDbs = msg->userAuth.readDbs;
  msg->userAuth.readDbs = NULL;

  taosHashCleanup(pUser->writeDbs);
  pUser->writeDbs = msg->userAuth.writeDbs;
  msg->userAuth.writeDbs = NULL;

  CTG_UNLOCK(CTG_WRITE, &pUser->lock);

_return:


  taosHashCleanup(msg->userAuth.createdDbs);
  taosHashCleanup(msg->userAuth.readDbs);
  taosHashCleanup(msg->userAuth.writeDbs);
 
  taosMemoryFreeClear(msg);
  
  CTG_RET(code);
}

D
dapan1121 已提交
2345 2346 2347 2348 2349 2350

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

  qInfo("catalog update thread started");

D
dapan 已提交
2351
  CTG_LOCK(CTG_READ, &gCtgMgmt.lock);
D
dapan1121 已提交
2352 2353
  
  while (true) {
D
dapan1121 已提交
2354 2355 2356
    if (tsem_wait(&gCtgMgmt.queue.reqSem)) {
      qError("ctg tsem_wait failed, error:%s", tstrerror(TAOS_SYSTEM_ERROR(errno)));
    }
D
dapan1121 已提交
2357
    
wafwerar's avatar
wafwerar 已提交
2358
    if (atomic_load_8((int8_t*)&gCtgMgmt.exit)) {
D
dapan1121 已提交
2359
      tsem_post(&gCtgMgmt.queue.rspSem);
D
dapan1121 已提交
2360 2361 2362 2363 2364
      break;
    }

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

D
dapan1121 已提交
2367
    ctgDebug("process [%s] action", gCtgAction[action->act].name);
D
dapan 已提交
2368 2369 2370
    
    (*gCtgAction[action->act].func)(action);

D
dapan1121 已提交
2371 2372 2373 2374 2375 2376
    gCtgMgmt.queue.seqDone = action->seqId;

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

D
dapan1121 已提交
2377
    CTG_RUNTIME_STAT_ADD(qDoneNum, 1); 
D
dapan1121 已提交
2378

D
dapan1121 已提交
2379
    ctgdShowClusterCache(pCtg);
D
dapan1121 已提交
2380 2381
  }

D
dapan 已提交
2382
  CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock);
D
dapan1121 已提交
2383 2384 2385 2386 2387 2388 2389 2390

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


int32_t ctgStartUpdateThread() {
wafwerar's avatar
wafwerar 已提交
2391 2392 2393
  TdThreadAttr thAttr;
  taosThreadAttrInit(&thAttr);
  taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
D
dapan1121 已提交
2394

wafwerar's avatar
wafwerar 已提交
2395
  if (taosThreadCreate(&gCtgMgmt.updateThread, &thAttr, ctgUpdateThreadFunc, NULL) != 0) {
D
dapan1121 已提交
2396 2397
    terrno = TAOS_SYSTEM_ERROR(errno);
    CTG_ERR_RET(terrno);
D
dapan1121 已提交
2398 2399
  }
  
wafwerar's avatar
wafwerar 已提交
2400
  taosThreadAttrDestroy(&thAttr);
D
dapan1121 已提交
2401 2402 2403
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
2404 2405 2406 2407 2408 2409 2410 2411 2412 2413 2414 2415 2416 2417 2418 2419
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 已提交
2420
  CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pRpc, pMgmtEps, db, &dbCache, &vgInfo));
D
dapan1121 已提交
2421 2422 2423 2424 2425 2426 2427 2428 2429 2430 2431 2432 2433 2434 2435 2436 2437 2438 2439 2440 2441 2442 2443 2444 2445 2446 2447 2448 2449 2450 2451 2452 2453 2454 2455 2456 2457 2458 2459 2460 2461 2462 2463 2464

  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 已提交
2465
  taosMemoryFreeClear(tbMeta);
D
dapan1121 已提交
2466 2467 2468

  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
wafwerar's avatar
wafwerar 已提交
2469
    taosMemoryFreeClear(vgInfo);
D
dapan1121 已提交
2470 2471 2472 2473 2474 2475 2476 2477 2478 2479
  }

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

  CTG_RET(code);
}

D
dapan1121 已提交
2480
int32_t catalogInit(SCatalogCfg *cfg) {
D
dapan 已提交
2481
  if (gCtgMgmt.pCluster) {
D
dapan 已提交
2482
    qError("catalog already initialized");
D
dapan1121 已提交
2483
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2484 2485
  }

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

D
dapan1121 已提交
2488
  if (cfg) {
D
dapan 已提交
2489
    memcpy(&gCtgMgmt.cfg, cfg, sizeof(*cfg));
H
Haojun Liao 已提交
2490

D
dapan 已提交
2491 2492
    if (gCtgMgmt.cfg.maxDBCacheNum == 0) {
      gCtgMgmt.cfg.maxDBCacheNum = CTG_DEFAULT_CACHE_DB_NUMBER;
D
dapan1121 已提交
2493 2494
    }

D
dapan 已提交
2495 2496
    if (gCtgMgmt.cfg.maxTblCacheNum == 0) {
      gCtgMgmt.cfg.maxTblCacheNum = CTG_DEFAULT_CACHE_TBLMETA_NUMBER;
D
dapan1121 已提交
2497
    }
D
dapan1121 已提交
2498

D
dapan 已提交
2499 2500
    if (gCtgMgmt.cfg.dbRentSec == 0) {
      gCtgMgmt.cfg.dbRentSec = CTG_DEFAULT_RENT_SECOND;
D
dapan1121 已提交
2501 2502
    }

D
dapan 已提交
2503 2504
    if (gCtgMgmt.cfg.stbRentSec == 0) {
      gCtgMgmt.cfg.stbRentSec = CTG_DEFAULT_RENT_SECOND;
D
dapan1121 已提交
2505
    }
D
dapan1121 已提交
2506
  } else {
D
dapan 已提交
2507 2508 2509 2510
    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 已提交
2511 2512
  }

D
dapan 已提交
2513 2514
  gCtgMgmt.pCluster = taosHashInit(CTG_DEFAULT_CACHE_CLUSTER_NUMBER, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_ENTRY_LOCK);
  if (NULL == gCtgMgmt.pCluster) {
D
dapan1121 已提交
2515 2516
    qError("taosHashInit %d cluster cache failed", CTG_DEFAULT_CACHE_CLUSTER_NUMBER);
    CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
D
dapan1121 已提交
2517 2518
  }

D
dapan1121 已提交
2519 2520 2521 2522 2523 2524 2525 2526 2527
  if (tsem_init(&gCtgMgmt.queue.reqSem, 0, 0)) {
    qError("tsem_init failed, error:%s", tstrerror(TAOS_SYSTEM_ERROR(errno)));
    CTG_ERR_RET(TSDB_CODE_CTG_SYS_ERROR);
  }
  
  if (tsem_init(&gCtgMgmt.queue.rspSem, 0, 0)) {
    qError("tsem_init failed, error:%s", tstrerror(TAOS_SYSTEM_ERROR(errno)));
    CTG_ERR_RET(TSDB_CODE_CTG_SYS_ERROR);
  }
D
dapan1121 已提交
2528

wafwerar's avatar
wafwerar 已提交
2529
  gCtgMgmt.queue.head = taosMemoryCalloc(1, sizeof(SCtgQNode));
D
dapan1121 已提交
2530
  if (NULL == gCtgMgmt.queue.head) {
D
dapan1121 已提交
2531 2532 2533
    qError("calloc %d failed", (int32_t)sizeof(SCtgQNode));
    CTG_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }
D
dapan1121 已提交
2534
  gCtgMgmt.queue.tail = gCtgMgmt.queue.head;
D
dapan1121 已提交
2535

D
dapan1121 已提交
2536 2537
  CTG_ERR_RET(ctgStartUpdateThread());

D
dapan 已提交
2538
  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 已提交
2539

D
dapan 已提交
2540
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
2541 2542
}

D
dapan1121 已提交
2543
int32_t catalogGetHandle(uint64_t clusterId, SCatalog** catalogHandle) {
2544
  if (NULL == catalogHandle) {
D
dapan1121 已提交
2545
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
2546 2547
  }

D
dapan 已提交
2548
  if (NULL == gCtgMgmt.pCluster) {
D
dapan 已提交
2549
    qError("catalog cluster cache are not ready, clusterId:%"PRIx64, clusterId);
D
dapan1121 已提交
2550
    CTG_ERR_RET(TSDB_CODE_CTG_NOT_READY);
D
dapan 已提交
2551 2552
  }

D
dapan1121 已提交
2553 2554
  int32_t code = 0;
  SCatalog *clusterCtg = NULL;
D
dapan 已提交
2555

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

D
dapan1121 已提交
2559 2560 2561 2562 2563
    if (ctg && (*ctg)) {
      *catalogHandle = *ctg;
      qDebug("got catalog handle from cache, clusterId:%"PRIx64", CTG:%p", clusterId, *ctg);
      return TSDB_CODE_SUCCESS;
    }
D
dapan 已提交
2564

wafwerar's avatar
wafwerar 已提交
2565
    clusterCtg = taosMemoryCalloc(1, sizeof(SCatalog));
D
dapan1121 已提交
2566 2567 2568 2569 2570
    if (NULL == clusterCtg) {
      qError("calloc %d failed", (int32_t)sizeof(SCatalog));
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }

D
dapan1121 已提交
2571 2572
    clusterCtg->clusterId = clusterId;

D
dapan 已提交
2573 2574
    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 已提交
2575

D
dapan 已提交
2576
    clusterCtg->dbCache = taosHashInit(gCtgMgmt.cfg.maxDBCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_ENTRY_LOCK);
D
dapan1121 已提交
2577 2578 2579 2580 2581
    if (NULL == clusterCtg->dbCache) {
      qError("taosHashInit %d dbCache failed", CTG_DEFAULT_CACHE_DB_NUMBER);
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
    }

D
dapan 已提交
2582
    SHashObj *metaCache = taosHashInit(gCtgMgmt.cfg.maxTblCacheNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_ENTRY_LOCK);
D
dapan1121 已提交
2583
    if (NULL == metaCache) {
D
dapan 已提交
2584
      qError("taosHashInit failed, num:%d", gCtgMgmt.cfg.maxTblCacheNum);
D
dapan1121 已提交
2585 2586 2587
      CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
    }
    
D
dapan 已提交
2588
    code = taosHashPut(gCtgMgmt.pCluster, &clusterId, sizeof(clusterId), &clusterCtg, POINTER_BYTES);
D
dapan1121 已提交
2589 2590 2591 2592 2593 2594 2595 2596 2597 2598 2599 2600 2601
    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 已提交
2602
  }
D
dapan1121 已提交
2603 2604

  *catalogHandle = clusterCtg;
D
dapan1121 已提交
2605 2606

  CTG_CACHE_STAT_ADD(clusterNum, 1);
D
dapan 已提交
2607
  
D
dapan1121 已提交
2608
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
2609 2610 2611 2612 2613 2614 2615 2616

_return:

  ctgFreeHandle(clusterCtg);
  
  CTG_RET(code);
}

D
dapan1121 已提交
2617 2618
void catalogFreeHandle(SCatalog* pCtg) {
  if (NULL == pCtg) {
D
dapan1121 已提交
2619 2620
    return;
  }
D
dapan1121 已提交
2621

D
dapan 已提交
2622
  if (taosHashRemove(gCtgMgmt.pCluster, &pCtg->clusterId, sizeof(pCtg->clusterId))) {
D
dapan1121 已提交
2623
    ctgWarn("taosHashRemove from cluster failed, may already be freed, clusterId:%"PRIx64, pCtg->clusterId);
D
dapan1121 已提交
2624 2625 2626
    return;
  }

D
dapan1121 已提交
2627 2628
  CTG_CACHE_STAT_SUB(clusterNum, 1);

D
dapan1121 已提交
2629
  uint64_t clusterId = pCtg->clusterId;
D
dapan1121 已提交
2630
  
D
dapan1121 已提交
2631
  ctgFreeHandle(pCtg);
D
dapan1121 已提交
2632
  
D
dapan1121 已提交
2633
  ctgInfo("handle freed, culsterId:%"PRIx64, clusterId);
D
dapan 已提交
2634 2635
}

D
dapan1121 已提交
2636
int32_t catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* version, int64_t* dbId, int32_t *tableNum) {
D
dapan1121 已提交
2637 2638
  CTG_API_ENTER();

D
dapan1121 已提交
2639
  if (NULL == pCtg || NULL == dbFName || NULL == version || NULL == dbId) {
D
dapan1121 已提交
2640 2641 2642 2643
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

  SCtgDBCache *dbCache = NULL;
D
dapan 已提交
2644
  bool inCache = false;
D
dapan1121 已提交
2645
  int32_t code = 0;
D
dapan1121 已提交
2646

D
dapan1121 已提交
2647 2648
  CTG_ERR_JRET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache, &inCache));
  if (!inCache) {
D
dapan1121 已提交
2649
    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
2650
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2651 2652
  }

D
dapan1121 已提交
2653
  *version = dbCache->vgInfo->vgVersion;
D
dapan1121 已提交
2654
  *dbId = dbCache->dbId;
D
dapan1121 已提交
2655
  *tableNum = dbCache->vgInfo->numOfTable;
D
dapan1121 已提交
2656 2657 2658

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

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

D
dapan1121 已提交
2662
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2663 2664 2665 2666

_return:

  CTG_API_LEAVE(code);
D
dapan1121 已提交
2667 2668
}

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

D
dapan1121 已提交
2672 2673 2674
  if (NULL == pCtg || NULL == dbFName || NULL == pRpc || NULL == pMgmtEps || NULL == vgroupList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
2675

D
dapan1121 已提交
2676
  SCtgDBCache* dbCache = NULL;
2677
  int32_t code = 0;
D
dapan1121 已提交
2678
  SArray *vgList = NULL;
D
dapan1121 已提交
2679 2680
  SHashObj *vgHash = NULL;
  SDBVgInfo *vgInfo = NULL;
D
dapan1121 已提交
2681
  CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pRpc, pMgmtEps, dbFName, &dbCache, &vgInfo));
D
dapan1121 已提交
2682 2683 2684 2685
  if (dbCache) {
    vgHash = dbCache->vgInfo->vgHash;
  } else {
    vgHash = vgInfo->vgHash;
D
dapan1121 已提交
2686 2687
  }

D
dapan1121 已提交
2688
  CTG_ERR_JRET(ctgGenerateVgList(pCtg, vgHash, &vgList));
D
dapan1121 已提交
2689 2690 2691 2692 2693

  *vgroupList = vgList;
  vgList = NULL;

_return:
D
dapan1121 已提交
2694 2695

  if (dbCache) {
D
dapan1121 已提交
2696 2697
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
2698 2699
  }

D
dapan1121 已提交
2700 2701
  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
wafwerar's avatar
wafwerar 已提交
2702
    taosMemoryFreeClear(vgInfo);
D
dapan1121 已提交
2703 2704
  }

D
dapan1121 已提交
2705
  CTG_API_LEAVE(code);  
D
dapan1121 已提交
2706 2707 2708
}


D
dapan1121 已提交
2709
int32_t catalogUpdateDBVgInfo(SCatalog* pCtg, const char* dbFName, uint64_t dbId, SDBVgInfo* dbInfo) {
D
dapan1121 已提交
2710
  CTG_API_ENTER();
D
dapan1121 已提交
2711 2712

  int32_t code = 0;
D
dapan1121 已提交
2713
  
D
dapan1121 已提交
2714
  if (NULL == pCtg || NULL == dbFName || NULL == dbInfo) {
D
dapan1121 已提交
2715
    ctgFreeVgInfo(dbInfo);
D
dapan1121 已提交
2716 2717 2718
    CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
  }

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

D
dapan1121 已提交
2721 2722
_return:

D
dapan1121 已提交
2723
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2724 2725 2726
}


D
dapan1121 已提交
2727 2728 2729
int32_t catalogRemoveDB(SCatalog* pCtg, const char* dbFName, uint64_t dbId) {
  CTG_API_ENTER();

D
dapan1121 已提交
2730 2731
  int32_t code = 0;
  
D
dapan1121 已提交
2732 2733
  if (NULL == pCtg || NULL == dbFName) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2734 2735
  }

D
dapan1121 已提交
2736
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2737
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2738
  }
D
dapan1121 已提交
2739

D
dapan1121 已提交
2740
  CTG_ERR_JRET(ctgPushRmDBMsgInQueue(pCtg, dbFName, dbId));
D
dapan 已提交
2741

D
dapan1121 已提交
2742
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2743
  
D
dapan1121 已提交
2744 2745
_return:

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

D
dapan1121 已提交
2749
int32_t catalogUpdateVgEpSet(SCatalog* pCtg, const char* dbFName, int32_t vgId, SEpSet *epSet) {
2750
  return 0;
D
dapan1121 已提交
2751
}
D
dapan1121 已提交
2752

D
dapan1121 已提交
2753
int32_t catalogRemoveTableMeta(SCatalog* pCtg, const SName* pTableName) {
D
dapan 已提交
2754 2755 2756 2757 2758 2759 2760 2761 2762 2763 2764 2765 2766
  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;
D
dapan1121 已提交
2767
  bool inCache = false;
D
dapan 已提交
2768
  uint64_t dbId = 0;
D
dapan1121 已提交
2769
  CTG_ERR_JRET(ctgGetTableMetaFromCache(pCtg, pTableName, &tblMeta, &inCache, 0, &dbId));
D
dapan 已提交
2770

D
dapan1121 已提交
2771
  if (!inCache) {
D
dapan 已提交
2772 2773 2774 2775 2776 2777 2778 2779
    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) {
D
dapan1121 已提交
2780
    CTG_ERR_JRET(ctgPushRmStbMsgInQueue(pCtg, dbFName, dbId, pTableName->tname, tblMeta->suid, true));
D
dapan 已提交
2781
  } else {
D
dapan1121 已提交
2782
    CTG_ERR_JRET(ctgPushRmTblMsgInQueue(pCtg, dbFName, dbId, pTableName->tname, true));
D
dapan 已提交
2783 2784 2785 2786 2787
  }

 
_return:

wafwerar's avatar
wafwerar 已提交
2788
  taosMemoryFreeClear(tblMeta);
D
dapan 已提交
2789 2790 2791 2792 2793

  CTG_API_LEAVE(code);
}


D
dapan1121 已提交
2794 2795 2796
int32_t catalogRemoveStbMeta(SCatalog* pCtg, const char* dbFName, uint64_t dbId, const char* stbName, uint64_t suid) {
  CTG_API_ENTER();

D
dapan 已提交
2797 2798
  int32_t code = 0;
  
D
dapan1121 已提交
2799 2800
  if (NULL == pCtg || NULL == dbFName || NULL == stbName) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
2801 2802
  }

D
dapan1121 已提交
2803
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2804
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2805
  }
D
dapan1121 已提交
2806

D
dapan1121 已提交
2807
  CTG_ERR_JRET(ctgPushRmStbMsgInQueue(pCtg, dbFName, dbId, stbName, suid, true));
D
dapan 已提交
2808

D
dapan1121 已提交
2809
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2810
  
D
dapan1121 已提交
2811 2812
_return:

D
dapan1121 已提交
2813
  CTG_API_LEAVE(code);
D
dapan 已提交
2814 2815
}

D
dapan1121 已提交
2816
int32_t catalogGetIndexMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, const char *pIndexName, SIndexMeta** pIndexMeta) {
2817
  return 0;
D
dapan1121 已提交
2818
}
D
dapan1121 已提交
2819

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

D
dapan1121 已提交
2823
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_UNKNOWN_STB));
D
dapan1121 已提交
2824
}
D
dapan1121 已提交
2825

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

D
dapan1121 已提交
2829
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_STB));
D
dapan1121 已提交
2830 2831
}

D
dapan1121 已提交
2832 2833 2834 2835 2836 2837 2838
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 已提交
2839
  STableMetaOutput *output = taosMemoryCalloc(1, sizeof(STableMetaOutput));
D
dapan1121 已提交
2840 2841 2842 2843 2844
  if (NULL == output) {
    ctgError("malloc %d failed", (int32_t)sizeof(STableMetaOutput));
    CTG_API_LEAVE(TSDB_CODE_CTG_MEM_ERROR);
  }
  
D
dapan1121 已提交
2845 2846
  int32_t code = 0;

D
dapan1121 已提交
2847 2848
  strcpy(output->dbFName, rspMsg->dbFName);
  strcpy(output->tbName, rspMsg->tbName);
D
dapan1121 已提交
2849

D
dapan1121 已提交
2850
  output->dbId = rspMsg->dbId;
D
dapan1121 已提交
2851
  
D
dapan1121 已提交
2852
  SET_META_TYPE_TABLE(output->metaType);
D
dapan1121 已提交
2853
  
D
dapan1121 已提交
2854
  CTG_ERR_JRET(queryCreateTableMetaFromMsg(rspMsg, true, &output->tbMeta));
D
dapan1121 已提交
2855

D
dapan1121 已提交
2856
  CTG_ERR_JRET(ctgPushUpdateTblMsgInQueue(pCtg, output, false));
D
dapan 已提交
2857

D
dapan1121 已提交
2858 2859
  CTG_API_LEAVE(code);
  
D
dapan1121 已提交
2860 2861
_return:

wafwerar's avatar
wafwerar 已提交
2862 2863
  taosMemoryFreeClear(output->tbMeta);
  taosMemoryFreeClear(output);
D
dapan1121 已提交
2864
  
D
dapan1121 已提交
2865
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2866 2867
}

D
dapan1121 已提交
2868 2869 2870 2871 2872 2873 2874 2875 2876
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 已提交
2877

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

D
dapan1121 已提交
2881
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pTableName) {
D
dapan1121 已提交
2882 2883 2884
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2885
  CTG_API_LEAVE(ctgRefreshTblMeta(pCtg, pTrans, pMgmtEps, pTableName, CTG_FLAG_FORCE_UPDATE | CTG_FLAG_MAKE_STB(isSTable), NULL, true));
2886
}
2887

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

D
dapan1121 已提交
2891
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_FORCE_UPDATE | CTG_FLAG_MAKE_STB(isSTable)));
D
dapan1121 已提交
2892 2893
}

D
dapan1121 已提交
2894
int32_t catalogGetTableDistVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgList) {
D
dapan1121 已提交
2895
  CTG_API_ENTER();
D
dapan1121 已提交
2896 2897 2898 2899

  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == pTableName || NULL == pVgList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
2900

D
dapan1121 已提交
2901
  if (CTG_IS_SYS_DBNAME(pTableName->dbname)) {
D
dapan1121 已提交
2902 2903 2904
    ctgError("no valid vgInfo for db, dbname:%s", pTableName->dbname);
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
2905

D
dapan1121 已提交
2906
  int32_t code = 0;
D
dapan1121 已提交
2907

D
dapan1121 已提交
2908 2909 2910 2911 2912
  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 已提交
2913

D
dapan1121 已提交
2914 2915 2916 2917 2918 2919
        char dbFName[TSDB_DB_FNAME_LEN] = {0};
        tNameGetFullDbName(pTableName, dbFName);        
        CTG_ERR_JRET(ctgRefreshDBVgInfo(pCtg, pRpc, pMgmtEps, dbFName));
        
        continue;
      }
D
dapan1121 已提交
2920
    }
D
dapan 已提交
2921

D
dapan1121 已提交
2922
    break;
D
dapan1121 已提交
2923
  }
D
dapan 已提交
2924

D
dapan1121 已提交
2925
_return:
D
dapan 已提交
2926

D
dapan1121 已提交
2927
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2928 2929 2930
}


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

D
dapan1121 已提交
2934
  if (CTG_IS_SYS_DBNAME(pTableName->dbname)) {
D
dapan1121 已提交
2935 2936 2937 2938
    ctgError("no valid vgInfo for db, dbname:%s", pTableName->dbname);
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2939 2940
  SCtgDBCache* dbCache = NULL;
  int32_t code = 0;
H
Haojun Liao 已提交
2941 2942
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);
D
dapan1121 已提交
2943

D
dapan1121 已提交
2944
  SDBVgInfo *vgInfo = NULL;
D
dapan1121 已提交
2945
  CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pTrans, pMgmtEps, db, &dbCache, &vgInfo));
D
dapan1121 已提交
2946

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

D
dapan1121 已提交
2949
_return:
D
dapan1121 已提交
2950

D
dapan1121 已提交
2951
  if (dbCache) {
D
dapan1121 已提交
2952 2953 2954 2955 2956 2957
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
  }

  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
wafwerar's avatar
wafwerar 已提交
2958
    taosMemoryFreeClear(vgInfo);
D
dapan1121 已提交
2959
  }
D
dapan1121 已提交
2960

D
dapan1121 已提交
2961
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2962 2963 2964
}


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

D
dapan1121 已提交
2968
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pReq || NULL == pRsp) {
D
dapan1121 已提交
2969 2970 2971
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2972
  int32_t code = 0;
D
dapan1121 已提交
2973
  pRsp->pTableMeta = NULL;
D
dapan1121 已提交
2974 2975 2976

  if (pReq->pTableName) {
    int32_t tbNum = (int32_t)taosArrayGetSize(pReq->pTableName);
D
dapan1121 已提交
2977
    if (tbNum <= 0) {
D
dapan1121 已提交
2978
      ctgError("empty table name list, tbNum:%d", tbNum);
D
dapan1121 已提交
2979
      CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2980
    }
H
Haojun Liao 已提交
2981

D
dapan1121 已提交
2982 2983
    pRsp->pTableMeta = taosArrayInit(tbNum, POINTER_BYTES);
    if (NULL == pRsp->pTableMeta) {
D
dapan1121 已提交
2984
      ctgError("taosArrayInit %d failed", tbNum);
D
dapan1121 已提交
2985
      CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
D
dapan1121 已提交
2986 2987 2988 2989 2990 2991
    }
    
    for (int32_t i = 0; i < tbNum; ++i) {
      SName *name = taosArrayGet(pReq->pTableName, i);
      STableMeta *pTableMeta = NULL;
      
D
dapan1121 已提交
2992
      CTG_ERR_JRET(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, name, &pTableMeta, CTG_FLAG_UNKNOWN_STB));
D
dapan1121 已提交
2993 2994 2995

      if (NULL == taosArrayPush(pRsp->pTableMeta, &pTableMeta)) {
        ctgError("taosArrayPush failed, idx:%d", i);
wafwerar's avatar
wafwerar 已提交
2996
        taosMemoryFreeClear(pTableMeta);
D
dapan1121 已提交
2997 2998 2999 3000 3001
        CTG_ERR_JRET(TSDB_CODE_CTG_MEM_ERROR);
      }
    }
  }

D
dapan1121 已提交
3002
  if (pReq->qNodeRequired) {
D
dapan 已提交
3003 3004
    pRsp->pQnodeList = taosArrayInit(10, sizeof(SQueryNodeAddr));
    CTG_ERR_JRET(ctgGetQnodeListFromMnode(pCtg, pTrans, pMgmtEps, pRsp->pQnodeList));
D
dapan1121 已提交
3005 3006
  }

D
dapan1121 已提交
3007
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
3008 3009

_return:  
D
dapan1121 已提交
3010

D
dapan1121 已提交
3011 3012 3013 3014
  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 已提交
3015
      taosMemoryFreeClear(pMeta);
D
dapan1121 已提交
3016 3017 3018
    }
    
    taosArrayDestroy(pRsp->pTableMeta);
D
dapan1121 已提交
3019
    pRsp->pTableMeta = NULL;
D
dapan1121 已提交
3020
  }
D
dapan 已提交
3021
  
D
dapan1121 已提交
3022
  CTG_API_LEAVE(code);
3023
}
D
dapan 已提交
3024

D
dapan1121 已提交
3025
int32_t catalogGetQnodeList(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SArray* pQnodeList) {
D
dapan1121 已提交
3026
  CTG_API_ENTER();
D
dapan1121 已提交
3027 3028
  
  int32_t code = 0;
D
dapan1121 已提交
3029 3030 3031 3032
  if (NULL == pCtg || NULL == pRpc  || NULL == pMgmtEps || NULL == pQnodeList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan 已提交
3033
  CTG_ERR_JRET(ctgGetQnodeListFromMnode(pCtg, pRpc, pMgmtEps, pQnodeList));
D
dapan1121 已提交
3034 3035

_return:
D
dapan 已提交
3036

D
dapan1121 已提交
3037
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
3038 3039
}

D
dapan1121 已提交
3040
int32_t catalogGetExpiredSTables(SCatalog* pCtg, SSTableMetaVersion **stables, uint32_t *num) {
D
dapan1121 已提交
3041 3042
  CTG_API_ENTER();

D
dapan1121 已提交
3043 3044
  if (NULL == pCtg || NULL == stables || NULL == num) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
3045 3046
  }

D
dapan1121 已提交
3047 3048 3049 3050
  CTG_API_LEAVE(ctgMetaRentGet(&pCtg->stbRent, (void **)stables, num, sizeof(SSTableMetaVersion)));
}

int32_t catalogGetExpiredDBs(SCatalog* pCtg, SDbVgVersion **dbs, uint32_t *num) {
D
dapan1121 已提交
3051
  CTG_API_ENTER();
D
dapan1121 已提交
3052 3053 3054 3055
  
  if (NULL == pCtg || NULL == dbs || NULL == num) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
3056

D
dapan1121 已提交
3057
  CTG_API_LEAVE(ctgMetaRentGet(&pCtg->dbRent, (void **)dbs, num, sizeof(SDbVgVersion)));
D
dapan1121 已提交
3058 3059
}

D
dapan 已提交
3060 3061 3062 3063 3064 3065 3066 3067 3068 3069 3070 3071 3072 3073 3074 3075 3076 3077 3078 3079 3080 3081 3082 3083 3084 3085 3086 3087 3088
int32_t catalogGetExpiredUsers(SCatalog* pCtg, SUserAuthVersion **users, uint32_t *num) {
  CTG_API_ENTER();
  
  if (NULL == pCtg || NULL == users || NULL == num) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

  *num = taosHashGetSize(pCtg->userCache);
  if (*num > 0) {
    *users = taosMemoryCalloc(*num, sizeof(SUserAuthVersion));
    if (NULL == *users) {
      ctgError("calloc %d userAuthVersion failed", *num);
      CTG_API_LEAVE(TSDB_CODE_OUT_OF_MEMORY);
    }
  }

  uint32_t i = 0;
  SCtgUserAuth *pAuth = taosHashIterate(pCtg->userCache, NULL);
  while (pAuth != NULL) {
    void *key = taosHashGetKey(pAuth, NULL);
    strncpy((*users)[i].user, key, sizeof((*users)[i].user));
    (*users)[i].version = pAuth->version;
    pAuth = taosHashIterate(pCtg->userCache, pAuth);
  }

  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
}


D
dapan1121 已提交
3089 3090 3091 3092 3093 3094 3095 3096 3097
int32_t catalogGetDBCfg(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* dbFName, SDbCfgInfo* pDbCfg) {
  CTG_API_ENTER();
  
  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == dbFName || NULL == pDbCfg) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

  CTG_API_LEAVE(ctgGetDBCfgFromMnode(pCtg, pRpc, pMgmtEps, dbFName, pDbCfg));
}
D
dapan 已提交
3098

D
dapan1121 已提交
3099 3100 3101 3102 3103 3104 3105 3106 3107 3108
int32_t catalogGetIndexInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* indexName, SIndexInfo* pInfo) {
  CTG_API_ENTER();
  
  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == indexName || NULL == pInfo) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

  CTG_API_LEAVE(ctgGetIndexInfoFromMnode(pCtg, pRpc, pMgmtEps, indexName, pInfo));
}

D
dapan1121 已提交
3109
int32_t catalogGetUdfInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* funcName, SFuncInfo** pInfo) {
D
dapan1121 已提交
3110 3111 3112 3113 3114 3115
  CTG_API_ENTER();
  
  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == funcName || NULL == pInfo) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
3116 3117 3118 3119 3120 3121 3122 3123 3124 3125 3126 3127 3128 3129 3130
  int32_t code = 0;
  *pInfo = taosMemoryMalloc(sizeof(SFuncInfo));
  if (NULL == *pInfo) {
    CTG_API_LEAVE(TSDB_CODE_OUT_OF_MEMORY);
  }

  CTG_ERR_JRET(ctgGetUdfInfoFromMnode(pCtg, pRpc, pMgmtEps, funcName, pInfo));
  
_return:

  if (code) {
    taosMemoryFreeClear(*pInfo);    
  }
  
  CTG_API_LEAVE(code);
D
dapan1121 已提交
3131 3132
}

D
dapan 已提交
3133
int32_t catalogChkAuth(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* user, const char* dbFName, AUTH_TYPE type, bool *pass) {
D
dapan 已提交
3134 3135
  CTG_API_ENTER();
  
D
dapan 已提交
3136
  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == user || NULL == dbFName || NULL == pass) {
D
dapan 已提交
3137 3138 3139 3140
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

  int32_t code = 0;
D
dapan 已提交
3141
  CTG_ERR_JRET(ctgChkAuth(pCtg, pRpc, pMgmtEps, user, dbFName, type, pass));
D
dapan 已提交
3142 3143 3144 3145 3146 3147
  
_return:

  CTG_API_LEAVE(code);
}

D
dapan 已提交
3148 3149 3150 3151 3152 3153 3154 3155 3156 3157
int32_t catalogUpdateUserAuthInfo(SCatalog* pCtg, SGetUserAuthRsp* pAuth) {
  CTG_API_ENTER();

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

  CTG_API_LEAVE(ctgPushUpdateUserMsgInQueue(pCtg, pAuth, false));
}

D
dapan1121 已提交
3158

D
dapan 已提交
3159
void catalogDestroy(void) {
D
dapan1121 已提交
3160 3161
  qInfo("start to destroy catalog");
  
wafwerar's avatar
wafwerar 已提交
3162
  if (NULL == gCtgMgmt.pCluster || atomic_load_8((int8_t*)&gCtgMgmt.exit)) {
D
dapan1121 已提交
3163 3164 3165
    return;
  }

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

D
dapan1121 已提交
3168 3169 3170 3171 3172 3173 3174
  if (tsem_post(&gCtgMgmt.queue.reqSem)) {
    qError("tsem_post failed, error:%s", tstrerror(TAOS_SYSTEM_ERROR(errno)));
  }
  
  if (tsem_post(&gCtgMgmt.queue.rspSem)) {
    qError("tsem_post failed, error:%s", tstrerror(TAOS_SYSTEM_ERROR(errno)));
  }
D
dapan1121 已提交
3175

D
dapan1121 已提交
3176
  while (CTG_IS_LOCKED(&gCtgMgmt.lock)) {
wafwerar's avatar
wafwerar 已提交
3177
    taosUsleep(1);
D
dapan1121 已提交
3178 3179
  }
  
D
dapan 已提交
3180
  CTG_LOCK(CTG_WRITE, &gCtgMgmt.lock);
D
dapan1121 已提交
3181

D
dapan1121 已提交
3182
  SCatalog *pCtg = NULL;
D
dapan 已提交
3183
  void *pIter = taosHashIterate(gCtgMgmt.pCluster, NULL);
D
dapan1121 已提交
3184
  while (pIter) {
D
dapan1121 已提交
3185
    pCtg = *(SCatalog **)pIter;
D
dapan1121 已提交
3186

D
dapan1121 已提交
3187 3188
    if (pCtg) {
      catalogFreeHandle(pCtg);
D
dapan1121 已提交
3189 3190
    }
    
D
dapan 已提交
3191
    pIter = taosHashIterate(gCtgMgmt.pCluster, pIter);
D
dapan 已提交
3192
  }
D
dapan1121 已提交
3193
  
D
dapan 已提交
3194 3195
  taosHashCleanup(gCtgMgmt.pCluster);
  gCtgMgmt.pCluster = NULL;
D
dapan1121 已提交
3196

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

D
dapan1121 已提交
3199
  qInfo("catalog destroyed");
D
dapan 已提交
3200 3201 3202 3203
}