catalog.c 90.6 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


D
dapan1121 已提交
567
int32_t ctgGetDBVgInfoFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, SBuildUseDBInput *input, SUseDbOutput *out) {
D
dapan1121 已提交
568 569 570
  char *msg = NULL;
  int32_t msgLen = 0;

D
dapan 已提交
571
  ctgDebug("try to get db vgInfo from mnode, dbFName:%s", input->db);
D
dapan1121 已提交
572 573 574 575 576 577

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

  SRpcMsg rpcRsp = {0};

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

D
dapan1121 已提交
593 594 595 596 597
  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 已提交
598

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

D
dapan1121 已提交
601 602
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
603

D
dapan1121 已提交
604 605 606 607 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
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 已提交
641 642 643 644 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
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 已提交
678
int32_t ctgGetUdfInfoFromMnode(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char *funcName, SFuncInfo **out) {
D
dapan1121 已提交
679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699
  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 已提交
700 701 702 703 704 705
    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 已提交
706 707 708 709
    ctgError("error rsp for get udf, error:%s, funcName:%s", tstrerror(rpcRsp.code), funcName);
    CTG_ERR_RET(rpcRsp.code);
  }

D
dapan1121 已提交
710
  code = queryProcessMsgRsp[TMSG_INDEX(TDMT_MND_RETRIEVE_FUNC)](*out, rpcRsp.pCont, rpcRsp.contLen);
D
dapan1121 已提交
711 712 713 714 715 716 717 718 719 720
  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 已提交
721 722 723 724 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
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 已提交
758

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

D
dapan1121 已提交
766 767
  SCtgDBCache *dbCache = NULL;
  ctgAcquireDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
768 769
  if (NULL == dbCache) {
    *exist = 0;
D
dapan1121 已提交
770 771 772
    return TSDB_CODE_SUCCESS;
  }

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

  *exist = 1;
D
dapan1121 已提交
787

D
dapan 已提交
788
  ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
789
  
D
dapan1121 已提交
790
  ctgDebug("tbmeta is in cache, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
791 792 793 794
  
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
795

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

D
dapan1121 已提交
802
  char dbFName[TSDB_DB_FNAME_LEN] = {0};
D
dapan1121 已提交
803
  if (CTG_FLAG_IS_SYS_DB(flag)) {
D
dapan1121 已提交
804 805 806 807
    strcpy(dbFName, pTableName->dbname);
  } else {
    tNameGetFullDbName(pTableName, dbFName);
  }
D
dapan1121 已提交
808

D
dapan1121 已提交
809 810
  *pTableMeta = NULL;

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

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

D
dapan1121 已提交
829 830 831
  if (dbId) {
    *dbId = dbCache->dbId;
  }
H
Haojun Liao 已提交
832

H
Haojun Liao 已提交
833
  STableMeta* tbMeta = *pTableMeta;
D
dapan1121 已提交
834

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

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

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

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

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

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

D
dapan1121 已提交
877
  ctgReleaseDBCache(pCtg, dbCache);
D
dapan 已提交
878

D
dapan1121 已提交
879 880 881
  *inCache = true;
  CTG_CACHE_STAT_ADD(tblHitNum, 1);

D
dapan1121 已提交
882
  ctgDebug("Got tbmeta from cache, dbFName:%s, tbName:%s", dbFName, pTableName->tname);
D
dapan1121 已提交
883
  
D
dapan1121 已提交
884 885 886 887 888 889 890
  return TSDB_CODE_SUCCESS;

_return:

  *inCache = false;
  CTG_CACHE_STAT_ADD(tblMissNum, 1);
  
D
dapan1121 已提交
891 892 893
  return TSDB_CODE_SUCCESS;
}

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

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

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

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

D
dapan1121 已提交
919 920 921 922
  taosHashRelease(dbCache->tbCache.metaCache, pTableMeta);

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

D
dapan 已提交
923
  ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
924

D
dapan 已提交
925
  ctgDebug("Got tbtype from cache, dbFName:%s, tbName:%s, type:%d", dbFName, tableName, *tbType);  
D
dapan1121 已提交
926 927 928 929
  
  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
930
int32_t ctgChkAuthFromCache(SCatalog* pCtg, const char* user, const char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass) {
D
dapan 已提交
931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947
  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 已提交
948
    *pass = true;
D
dapan 已提交
949 950 951 952 953
    return TSDB_CODE_SUCCESS;
  }

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

  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 已提交
979
int32_t ctgGetTableMetaFromMnodeImpl(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, char *dbFName, char* tbName, STableMetaOutput* output) {
D
dapan1121 已提交
980
  SBuildTableMetaInput bInput = {.vgId = 0, .dbFName = dbFName, .tbName = tbName};
D
dapan1121 已提交
981 982 983 984
  char *msg = NULL;
  SEpSet *pVnodeEpSet = NULL;
  int32_t msgLen = 0;

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

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

  SRpcMsg rpcMsg = {
D
dapan1121 已提交
994
      .msgType = TDMT_MND_TABLE_META,
D
dapan1121 已提交
995 996 997
      .pCont   = msg,
      .contLen = msgLen,
  };
D
dapan1121 已提交
998

D
dapan1121 已提交
999 1000
  SRpcMsg rpcRsp = {0};

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

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

D
dapan1121 已提交
1020
  ctgDebug("Got table meta from mnode, dbFName:%s, tbName:%s", dbFName, tbName);
D
dapan1121 已提交
1021 1022 1023 1024

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
1025
int32_t ctgGetTableMetaFromMnode(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, STableMetaOutput* output) {
D
dapan1121 已提交
1026 1027
  char dbFName[TSDB_DB_FNAME_LEN];
  tNameGetFullDbName(pTableName, dbFName);
D
dapan1121 已提交
1028

D
dapan1121 已提交
1029
  return ctgGetTableMetaFromMnodeImpl(pCtg, pTrans, pMgmtEps, dbFName, (char *)pTableName->tname, output);
D
dapan1121 已提交
1030
}
D
dapan1121 已提交
1031

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

D
dapan1121 已提交
1037 1038
  char dbFName[TSDB_DB_FNAME_LEN];
  tNameGetFullDbName(pTableName, dbFName);
D
dapan1121 已提交
1039

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

D
dapan1121 已提交
1042
  SBuildTableMetaInput bInput = {.vgId = vgroupInfo->vgId, .dbFName = dbFName, .tbName = (char *)tNameGetTableName(pTableName)};
D
dapan1121 已提交
1043 1044 1045
  char *msg = NULL;
  int32_t msgLen = 0;

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

  SRpcMsg rpcMsg = {
H
Hongze Cheng 已提交
1053
      .msgType = TDMT_VND_TABLE_META,
D
dapan1121 已提交
1054 1055 1056 1057 1058
      .pCont   = msg,
      .contLen = msgLen,
  };

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

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

D
dapan1121 已提交
1078
  ctgDebug("Got table meta from vnode, dbFName:%s, tbName:%s", dbFName, tNameGetTableName(pTableName));
D
dapan1121 已提交
1079 1080 1081
  return TSDB_CODE_SUCCESS;
}

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

1109 1110
int32_t ctgGetHashFunction(int8_t hashMethod, tableNameHashFp *fp) {
  switch (hashMethod) {
D
dapan1121 已提交
1111 1112 1113 1114 1115 1116 1117 1118
    default:
      *fp = MurmurHash3_32;
      break;
  }

  return TSDB_CODE_SUCCESS;
}

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

D
dapan1121 已提交
1126
  vgList = taosArrayInit(vgNum, sizeof(SVgroupInfo));
D
dapan1121 已提交
1127
  if (NULL == vgList) {
D
dapan1121 已提交
1128
    ctgError("taosArrayInit failed, num:%d", vgNum);
D
dapan 已提交
1129 1130 1131
    CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);    
  }

D
dapan1121 已提交
1132
  void *pIter = taosHashIterate(vgHash, NULL);
1133 1134
  while (pIter) {
    vgInfo = pIter;
D
dapan1121 已提交
1135

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

D
dapan1121 已提交
1146
  *pList = vgList;
D
dapan1121 已提交
1147

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

D
dapan1121 已提交
1150
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1151 1152 1153 1154 1155 1156 1157 1158

_return:

  if (vgList) {
    taosArrayDestroy(vgList);
  }

  CTG_RET(code);
D
dapan1121 已提交
1159 1160
}

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

1168
  if (vgNum <= 0) {
D
dapan1121 已提交
1169
    ctgError("db vgroup cache invalid, db:%s, vgroup number:%d", db, vgNum);
D
dapan1121 已提交
1170
    CTG_ERR_RET(TSDB_CODE_TSC_DB_NOT_SELECTED);
D
dapan1121 已提交
1171 1172
  }

1173 1174
  tableNameHashFp fp = NULL;
  SVgroupInfo *vgInfo = NULL;
D
dapan1121 已提交
1175

D
dapan1121 已提交
1176
  CTG_ERR_RET(ctgGetHashFunction(dbInfo->hashMethod, &fp));
1177 1178

  char tbFullName[TSDB_TABLE_FNAME_LEN];
H
Haojun Liao 已提交
1179
  tNameExtractFullName(pTableName, tbFullName);
1180 1181 1182

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

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

1195
  if (NULL == vgInfo) {
D
dapan1121 已提交
1196 1197
    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);
1198 1199 1200 1201
  }

  *pVgroup = *vgInfo;

1202
  CTG_RET(code);
D
dapan1121 已提交
1203 1204
}

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

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

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

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

D
dapan1121 已提交
1259 1260 1261 1262
  qDebug("meta rent initialized, type:%d, slotNum:%d", type, mgmt->slotNum);
  
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
1263

D
dapan1121 已提交
1264

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

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

D
dapan1121 已提交
1280 1281 1282
  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 已提交
1283 1284
  }

D
dapan1121 已提交
1285
  slot->needSort = true;
D
dapan1121 已提交
1286

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

D
dapan1121 已提交
1289 1290 1291 1292 1293 1294
_return:

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

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

D
dapan1121 已提交
1298
  SCtgRentSlot *slot = &mgmt->slots[widx];
D
dapan1121 已提交
1299
  int32_t code = 0;
1300

D
dapan1121 已提交
1301 1302
  CTG_LOCK(CTG_WRITE, &slot->lock);
  if (NULL == slot->meta) {
D
dapan1121 已提交
1303 1304
    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 已提交
1305 1306 1307
  }

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

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

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

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

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

D
dapan1121 已提交
1441
int32_t ctgAddNewDBCache(SCatalog *pCtg, const char *dbFName, uint64_t dbId) {
D
dapan1121 已提交
1442
  int32_t code = 0;
D
dapan1121 已提交
1443

D
dapan1121 已提交
1444 1445 1446
  SCtgDBCache newDBCache = {0};
  newDBCache.dbId = dbId;

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

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

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

  CTG_CACHE_STAT_ADD(dbNum, 1);
D
dapan1121 已提交
1471
 
D
dapan1121 已提交
1472
  SDbVgVersion vgVersion = {.dbId = newDBCache.dbId, .vgVersion = -1};
D
dapan1121 已提交
1473 1474
  strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));

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

D
dapan1121 已提交
1477 1478 1479
  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 已提交
1480

D
dapan1121 已提交
1481
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1482

D
dapan1121 已提交
1483
_return:
D
dapan1121 已提交
1484

D
dapan1121 已提交
1485
  ctgFreeDbCache(&newDBCache);
D
dapan1121 已提交
1486

D
dapan1121 已提交
1487 1488
  CTG_RET(code);
}
D
dapan1121 已提交
1489

D
dapan1121 已提交
1490

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

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


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

D
dapan1121 已提交
1515 1516 1517 1518 1519 1520
  atomic_store_8(&dbCache->deleted, 1);

  ctgRemoveStbRent(pCtg, &dbCache->tbCache);

  ctgFreeDbCache(dbCache);

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

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

D
dapan1121 已提交
1530 1531
  CTG_CACHE_STAT_SUB(dbNum, 1);

D
dapan 已提交
1532
  ctgInfo("db removed from cache, dbFName:%s, dbId:%"PRIx64, dbFName, dbId);
D
dapan1121 已提交
1533 1534 1535
  
  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
1536 1537


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

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

D
dapan1121 已提交
1572
  ctgGetDBCache(pCtg, dbFName, &dbCache);
D
dapan1121 已提交
1573

D
dapan1121 已提交
1574
  *pCache = dbCache;
D
dapan1121 已提交
1575

D
dapan1121 已提交
1576
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1577 1578 1579
}


D
dapan1121 已提交
1580 1581 1582
int32_t ctgUpdateDBVgInfo(SCatalog* pCtg, const char* dbFName, uint64_t dbId, SDBVgInfo** pDbInfo) {
  int32_t code = 0;
  SDBVgInfo* dbInfo = *pDbInfo;
D
dapan1121 已提交
1583 1584 1585 1586

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

D
dapan1121 已提交
1594
  bool newAdded = false;
D
dapan 已提交
1595
  SDbVgVersion vgVersion = {.dbId = dbId, .vgVersion = dbInfo->vgVersion, .numOfTable = dbInfo->numOfTable};
D
dapan1121 已提交
1596 1597 1598 1599 1600 1601 1602 1603 1604 1605

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

    ctgFreeVgInfo(dbCache->vgInfo);
D
dapan1121 已提交
1623 1624
  }

D
dapan1121 已提交
1625
  dbCache->vgInfo = dbInfo;
D
dapan1121 已提交
1626

D
dapan1121 已提交
1627
  *pDbInfo = NULL;
D
dapan1121 已提交
1628

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

D
dapan1121 已提交
1631
  ctgWReleaseVgInfo(dbCache);
D
dapan1121 已提交
1632

D
dapan1121 已提交
1633
  dbCache = NULL;
D
dapan1121 已提交
1634

D
dapan1121 已提交
1635
  strncpy(vgVersion.dbFName, dbFName, sizeof(vgVersion.dbFName));
D
dapan1121 已提交
1636
  CTG_ERR_RET(ctgMetaRentUpdate(&pCtg->dbRent, &vgVersion, vgVersion.dbId, sizeof(SDbVgVersion), ctgDbVgVersionSortCompare, ctgDbVgVersionSearchCompare));
D
dapan1121 已提交
1637
  
D
dapan1121 已提交
1638 1639 1640 1641
  CTG_RET(code);
}


D
dapan1121 已提交
1642 1643
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 已提交
1644

D
dapan1121 已提交
1645 1646 1647 1648 1649
  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 已提交
1650 1651
  }

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

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

D
dapan1121 已提交
1674
      origSuid = orig->suid;
D
dapan1121 已提交
1675
    }
D
dapan1121 已提交
1676
  }
D
dapan1121 已提交
1677

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

D
dapan1121 已提交
1692 1693 1694 1695
  if (NULL == orig) {
    CTG_CACHE_STAT_ADD(tblNum, 1);
  }

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

D
dapan1121 已提交
1699 1700 1701
  if (!isStb) {
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);  
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1702
  }
D
dapan1121 已提交
1703

D
dapan1121 已提交
1704
  if (origType == TSDB_SUPER_TABLE && origSuid == meta->suid) {
D
dapan1121 已提交
1705 1706 1707 1708
    CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);
    CTG_UNLOCK(CTG_READ, &tbCache->metaLock);  
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
1709

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

  CTG_CACHE_STAT_ADD(stblNum, 1);
D
dapan1121 已提交
1719 1720
  
  CTG_UNLOCK(CTG_WRITE, &tbCache->stbLock);
D
dapan1121 已提交
1721

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

D
dapan 已提交
1724
  ctgDebug("stb updated to stbCache, dbFName:%s, tbName:%s, tbType:%d", dbFName, tbName, meta->tableType);
D
dapan1121 已提交
1725 1726 1727 1728 1729 1730 1731

  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 已提交
1732 1733
}

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

  int32_t *vgId = NULL;
  void *pIter = taosHashIterate(src->vgHash, NULL);
  while (pIter) {
H
Haojun Liao 已提交
1754
    vgId = taosHashGetKey(pIter, NULL);
D
dapan 已提交
1755 1756 1757 1758 1759

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


  return TSDB_CODE_SUCCESS;
}



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

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

D
dapan1121 已提交
1779
  if (inCache) {
D
dapan1121 已提交
1780
    return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1781 1782 1783 1784 1785
  }

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

D
dapan1121 已提交
1786
  tstrncpy(input.db, dbFName, tListLen(input.db));
D
dapan1121 已提交
1787
  input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
H
Haojun Liao 已提交
1788

D
dapan1121 已提交
1789 1790 1791 1792 1793 1794 1795 1796 1797
  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 已提交
1798

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

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

D
dapan1121 已提交
1803
  return TSDB_CODE_SUCCESS;
D
dapan 已提交
1804 1805 1806

_return:

wafwerar's avatar
wafwerar 已提交
1807
  taosMemoryFreeClear(*pInfo);
D
dapan 已提交
1808 1809 1810
  *pInfo = DbOut.dbVgroup;
  
  CTG_RET(code);
D
dapan1121 已提交
1811 1812
}

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

D
dapan1121 已提交
1818
  CTG_ERR_RET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache, &inCache));
D
dapan1121 已提交
1819 1820 1821 1822 1823 1824

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

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

D
dapan1121 已提交
1827 1828
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
1829
  }
D
dapan1121 已提交
1830 1831 1832
  
  input.vgVersion = CTG_DEFAULT_INVALID_VERSION;
  input.numOfTable = 0;
D
dapan1121 已提交
1833 1834 1835

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

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

D
dapan1121 已提交
1858 1859 1860 1861
  memcpy(*pOutput, output, sizeof(STableMetaOutput));

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

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

D
dapan1121 已提交
1872 1873 1874
  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
1875 1876


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

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

D
dapan1121 已提交
1885
  if (!CTG_FLAG_IS_SYS_DB(flag)) {
D
dapan1121 已提交
1886 1887
    CTG_ERR_RET(catalogGetTableHashVgroup(pCtg, pTrans, pMgmtEps, pTableName, &vgroupInfo));
  }
D
dapan1121 已提交
1888

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

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

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

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

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

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

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

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

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

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

D
dapan 已提交
1951 1952 1953 1954 1955 1956
  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 已提交
1957 1958 1959 1960
  if (pOutput) {
    CTG_ERR_JRET(ctgCloneMetaOutput(output, pOutput));
  }

D
dapan1121 已提交
1961
  CTG_ERR_JRET(ctgPushUpdateTblMsgInQueue(pCtg, output, syncReq));
D
dapan 已提交
1962

D
dapan1121 已提交
1963
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1964 1965 1966

_return:

wafwerar's avatar
wafwerar 已提交
1967 1968
  taosMemoryFreeClear(output->tbMeta);
  taosMemoryFreeClear(output);
D
dapan1121 已提交
1969 1970 1971 1972
  
  CTG_RET(code);
}

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

D
dapan1121 已提交
1978
  bool inCache = false;
D
dapan1121 已提交
1979
  int32_t code = 0;
D
dapan1121 已提交
1980 1981 1982
  uint64_t dbId = 0;
  uint64_t suid = 0;
  STableMetaOutput *output = NULL;
D
dapan1121 已提交
1983

D
dapan1121 已提交
1984 1985
  if (CTG_IS_SYS_DBNAME(pTableName->dbname)) {
    CTG_FLAG_SET_SYS_DB(flag);
D
dapan1121 已提交
1986 1987
  }

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

D
dapan1121 已提交
1990 1991
  int32_t tbType = 0;

D
dapan1121 已提交
1992
  if (inCache) {
D
dapan1121 已提交
1993
    if (CTG_FLAG_MATCH_STB(flag, (*pTableMeta)->tableType) && ((!CTG_FLAG_IS_FORCE_UPDATE(flag)) || (CTG_FLAG_IS_SYS_DB(flag)))) {
D
dapan1121 已提交
1994
      goto _return;
D
dapan1121 已提交
1995
    }
D
dapan1121 已提交
1996

D
dapan1121 已提交
1997 1998
    tbType = (*pTableMeta)->tableType;
    suid = (*pTableMeta)->suid;
D
dapan1121 已提交
1999

wafwerar's avatar
wafwerar 已提交
2000
    taosMemoryFreeClear(*pTableMeta);
D
dapan1121 已提交
2001
  }
D
dapan1121 已提交
2002

D
dapan1121 已提交
2003 2004
  if (CTG_FLAG_IS_UNKNOWN_STB(flag)) {
    CTG_FLAG_SET_STB(flag, tbType);
D
dapan1121 已提交
2005 2006
  }

D
dapan1121 已提交
2007

D
dapan 已提交
2008
  while (true) {
D
dapan1121 已提交
2009
    CTG_ERR_JRET(ctgRefreshTblMeta(pCtg, pRpc, pMgmtEps, pTableName, flag, &output, false));
D
dapan1121 已提交
2010

D
dapan 已提交
2011 2012 2013 2014
    if (CTG_IS_META_TABLE(output->metaType)) {
      *pTableMeta = output->tbMeta;
      goto _return;
    }
D
dapan1121 已提交
2015

D
dapan 已提交
2016 2017 2018 2019 2020 2021
    if (CTG_IS_META_BOTH(output->metaType)) {
      memcpy(output->tbMeta, &output->ctbMeta, sizeof(output->ctbMeta));
      
      *pTableMeta = output->tbMeta;
      goto _return;
    }
D
dapan1121 已提交
2022

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

D
dapan 已提交
2029
    // HANDLE ONLY CHILD TABLE META
D
dapan1121 已提交
2030

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

D
dapan 已提交
2040 2041 2042 2043
    memcpy(*pTableMeta, &output->ctbMeta, sizeof(output->ctbMeta));

    break;
  }
D
dapan1121 已提交
2044 2045 2046

_return:

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

    if (TSDB_SUPER_TABLE == tbType) {
D
dapan1121 已提交
2056
      ctgPushRmStbMsgInQueue(pCtg, dbFName, dbId, pTableName->tname, suid, false);
D
dapan1121 已提交
2057
    } else {
D
dapan1121 已提交
2058
      ctgPushRmTblMsgInQueue(pCtg, dbFName, dbId, pTableName->tname, false);
D
dapan1121 已提交
2059 2060 2061
    }
  }

wafwerar's avatar
wafwerar 已提交
2062
  taosMemoryFreeClear(output);
D
dapan1121 已提交
2063

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

D
dapan1121 已提交
2069 2070 2071
  CTG_RET(code);
}

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

  if (inCache) {
    return TSDB_CODE_SUCCESS;
  }

  SGetUserAuthRsp authRsp = {0};
  CTG_ERR_RET(ctgGetUserDbAuthFromMnode(pCtg, pRpc, pMgmtEps, user, &authRsp));
  
  if (authRsp.superAuth) {
D
dapan 已提交
2088
    *pass = true;
D
dapan 已提交
2089 2090 2091 2092
    goto _return;
  }

  if (authRsp.createdDbs && taosHashGet(authRsp.createdDbs, dbFName, strlen(dbFName))) {
D
dapan 已提交
2093
    *pass = true;
D
dapan 已提交
2094 2095 2096
    goto _return;
  }

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

D
dapan 已提交
2101 2102
  if (authRsp.writeDbs && taosHashGet(authRsp.writeDbs, dbFName, strlen(dbFName)) && type == AUTH_TYPE_WRITE) {
    *pass = true;
D
dapan 已提交
2103 2104 2105 2106 2107 2108 2109 2110
  }

_return:

  ctgPushUpdateUserMsgInQueue(pCtg, &authRsp, false);

  return TSDB_CODE_SUCCESS;
}
D
dapan1121 已提交
2111 2112 2113 2114 2115 2116 2117 2118 2119 2120


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

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

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

wafwerar's avatar
wafwerar 已提交
2240
  taosMemoryFreeClear(msg);
D
dapan1121 已提交
2241 2242 2243 2244 2245
  
  CTG_RET(code);
}

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

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

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

D
dapan1121 已提交
2273
_return:
D
dapan1121 已提交
2274

wafwerar's avatar
wafwerar 已提交
2275
  taosMemoryFreeClear(msg);
H
Haojun Liao 已提交
2276

D
dapan1121 已提交
2277
  CTG_RET(code);
D
dapan1121 已提交
2278 2279
}

D
dapan 已提交
2280 2281 2282 2283 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
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 已提交
2341 2342 2343 2344 2345 2346

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

  qInfo("catalog update thread started");

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

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

D
dapan1121 已提交
2363
    ctgDebug("process [%s] action", gCtgAction[action->act].name);
D
dapan 已提交
2364 2365 2366
    
    (*gCtgAction[action->act].func)(action);

D
dapan1121 已提交
2367 2368 2369 2370 2371 2372
    gCtgMgmt.queue.seqDone = action->seqId;

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

D
dapan1121 已提交
2373
    CTG_RUNTIME_STAT_ADD(qDoneNum, 1); 
D
dapan1121 已提交
2374

D
dapan1121 已提交
2375
    ctgdShowClusterCache(pCtg);
D
dapan1121 已提交
2376 2377
  }

D
dapan 已提交
2378
  CTG_UNLOCK(CTG_READ, &gCtgMgmt.lock);
D
dapan1121 已提交
2379 2380 2381 2382 2383 2384 2385 2386

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


int32_t ctgStartUpdateThread() {
wafwerar's avatar
wafwerar 已提交
2387 2388 2389
  TdThreadAttr thAttr;
  taosThreadAttrInit(&thAttr);
  taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
D
dapan1121 已提交
2390

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

D
dapan1121 已提交
2400 2401 2402 2403 2404 2405 2406 2407 2408 2409 2410 2411 2412 2413 2414 2415
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 已提交
2416
  CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pRpc, pMgmtEps, db, &dbCache, &vgInfo));
D
dapan1121 已提交
2417 2418 2419 2420 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

  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 已提交
2461
  taosMemoryFreeClear(tbMeta);
D
dapan1121 已提交
2462 2463 2464

  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
wafwerar's avatar
wafwerar 已提交
2465
    taosMemoryFreeClear(vgInfo);
D
dapan1121 已提交
2466 2467 2468 2469 2470 2471 2472 2473 2474 2475
  }

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

  CTG_RET(code);
}

D
dapan1121 已提交
2476
int32_t catalogInit(SCatalogCfg *cfg) {
D
dapan 已提交
2477
  if (gCtgMgmt.pCluster) {
D
dapan 已提交
2478
    qError("catalog already initialized");
D
dapan1121 已提交
2479
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2480 2481
  }

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

D
dapan1121 已提交
2484
  if (cfg) {
D
dapan 已提交
2485
    memcpy(&gCtgMgmt.cfg, cfg, sizeof(*cfg));
H
Haojun Liao 已提交
2486

D
dapan 已提交
2487 2488
    if (gCtgMgmt.cfg.maxDBCacheNum == 0) {
      gCtgMgmt.cfg.maxDBCacheNum = CTG_DEFAULT_CACHE_DB_NUMBER;
D
dapan1121 已提交
2489 2490
    }

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

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

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

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

D
dapan1121 已提交
2515 2516 2517 2518 2519 2520 2521 2522 2523
  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 已提交
2524

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

D
dapan1121 已提交
2532 2533
  CTG_ERR_RET(ctgStartUpdateThread());

D
dapan 已提交
2534
  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 已提交
2535

D
dapan 已提交
2536
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
2537 2538
}

D
dapan1121 已提交
2539
int32_t catalogGetHandle(uint64_t clusterId, SCatalog** catalogHandle) {
2540
  if (NULL == catalogHandle) {
D
dapan1121 已提交
2541
    CTG_ERR_RET(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
2542 2543
  }

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

D
dapan1121 已提交
2549 2550
  int32_t code = 0;
  SCatalog *clusterCtg = NULL;
D
dapan 已提交
2551

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

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

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

D
dapan1121 已提交
2567 2568
    clusterCtg->clusterId = clusterId;

D
dapan 已提交
2569 2570
    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 已提交
2571

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

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

  *catalogHandle = clusterCtg;
D
dapan1121 已提交
2601 2602

  CTG_CACHE_STAT_ADD(clusterNum, 1);
D
dapan 已提交
2603
  
D
dapan1121 已提交
2604
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
2605 2606 2607 2608 2609 2610 2611 2612

_return:

  ctgFreeHandle(clusterCtg);
  
  CTG_RET(code);
}

D
dapan1121 已提交
2613 2614
void catalogFreeHandle(SCatalog* pCtg) {
  if (NULL == pCtg) {
D
dapan1121 已提交
2615 2616
    return;
  }
D
dapan1121 已提交
2617

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

D
dapan1121 已提交
2623 2624
  CTG_CACHE_STAT_SUB(clusterNum, 1);

D
dapan1121 已提交
2625
  uint64_t clusterId = pCtg->clusterId;
D
dapan1121 已提交
2626
  
D
dapan1121 已提交
2627
  ctgFreeHandle(pCtg);
D
dapan1121 已提交
2628
  
D
dapan1121 已提交
2629
  ctgInfo("handle freed, culsterId:%"PRIx64, clusterId);
D
dapan 已提交
2630 2631
}

D
dapan1121 已提交
2632
int32_t catalogGetDBVgVersion(SCatalog* pCtg, const char* dbFName, int32_t* version, int64_t* dbId, int32_t *tableNum) {
D
dapan1121 已提交
2633 2634
  CTG_API_ENTER();

D
dapan1121 已提交
2635
  if (NULL == pCtg || NULL == dbFName || NULL == version || NULL == dbId) {
D
dapan1121 已提交
2636 2637 2638 2639
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

  SCtgDBCache *dbCache = NULL;
D
dapan 已提交
2640
  bool inCache = false;
D
dapan1121 已提交
2641
  int32_t code = 0;
D
dapan1121 已提交
2642

D
dapan1121 已提交
2643 2644
  CTG_ERR_JRET(ctgAcquireVgInfoFromCache(pCtg, dbFName, &dbCache, &inCache));
  if (!inCache) {
D
dapan1121 已提交
2645
    *version = CTG_DEFAULT_INVALID_VERSION;
D
dapan1121 已提交
2646
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2647 2648
  }

D
dapan1121 已提交
2649
  *version = dbCache->vgInfo->vgVersion;
D
dapan1121 已提交
2650
  *dbId = dbCache->dbId;
D
dapan1121 已提交
2651
  *tableNum = dbCache->vgInfo->numOfTable;
D
dapan1121 已提交
2652 2653 2654

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

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

D
dapan1121 已提交
2658
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2659 2660 2661 2662

_return:

  CTG_API_LEAVE(code);
D
dapan1121 已提交
2663 2664
}

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

D
dapan1121 已提交
2668 2669 2670
  if (NULL == pCtg || NULL == dbFName || NULL == pRpc || NULL == pMgmtEps || NULL == vgroupList) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
2671

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

D
dapan1121 已提交
2684
  CTG_ERR_JRET(ctgGenerateVgList(pCtg, vgHash, &vgList));
D
dapan1121 已提交
2685 2686 2687 2688 2689

  *vgroupList = vgList;
  vgList = NULL;

_return:
D
dapan1121 已提交
2690 2691

  if (dbCache) {
D
dapan1121 已提交
2692 2693
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
D
dapan1121 已提交
2694 2695
  }

D
dapan1121 已提交
2696 2697
  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
wafwerar's avatar
wafwerar 已提交
2698
    taosMemoryFreeClear(vgInfo);
D
dapan1121 已提交
2699 2700
  }

D
dapan1121 已提交
2701
  CTG_API_LEAVE(code);  
D
dapan1121 已提交
2702 2703 2704
}


D
dapan1121 已提交
2705
int32_t catalogUpdateDBVgInfo(SCatalog* pCtg, const char* dbFName, uint64_t dbId, SDBVgInfo* dbInfo) {
D
dapan1121 已提交
2706
  CTG_API_ENTER();
D
dapan1121 已提交
2707 2708

  int32_t code = 0;
D
dapan1121 已提交
2709
  
D
dapan1121 已提交
2710
  if (NULL == pCtg || NULL == dbFName || NULL == dbInfo) {
D
dapan1121 已提交
2711
    ctgFreeVgInfo(dbInfo);
D
dapan1121 已提交
2712 2713 2714
    CTG_ERR_JRET(TSDB_CODE_CTG_INVALID_INPUT);
  }

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

D
dapan1121 已提交
2717 2718
_return:

D
dapan1121 已提交
2719
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2720 2721 2722
}


D
dapan1121 已提交
2723 2724 2725
int32_t catalogRemoveDB(SCatalog* pCtg, const char* dbFName, uint64_t dbId) {
  CTG_API_ENTER();

D
dapan1121 已提交
2726 2727
  int32_t code = 0;
  
D
dapan1121 已提交
2728 2729
  if (NULL == pCtg || NULL == dbFName) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
2730 2731
  }

D
dapan1121 已提交
2732
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2733
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2734
  }
D
dapan1121 已提交
2735

D
dapan1121 已提交
2736
  CTG_ERR_JRET(ctgPushRmDBMsgInQueue(pCtg, dbFName, dbId));
D
dapan 已提交
2737

D
dapan1121 已提交
2738
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
2739
  
D
dapan1121 已提交
2740 2741
_return:

D
dapan1121 已提交
2742
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2743 2744
}

D
dapan1121 已提交
2745
int32_t catalogUpdateVgEpSet(SCatalog* pCtg, const char* dbFName, int32_t vgId, SEpSet *epSet) {
2746
  return 0;
D
dapan1121 已提交
2747
}
D
dapan1121 已提交
2748

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

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

 
_return:

wafwerar's avatar
wafwerar 已提交
2784
  taosMemoryFreeClear(tblMeta);
D
dapan 已提交
2785 2786 2787 2788 2789

  CTG_API_LEAVE(code);
}


D
dapan1121 已提交
2790 2791 2792
int32_t catalogRemoveStbMeta(SCatalog* pCtg, const char* dbFName, uint64_t dbId, const char* stbName, uint64_t suid) {
  CTG_API_ENTER();

D
dapan 已提交
2793 2794
  int32_t code = 0;
  
D
dapan1121 已提交
2795 2796
  if (NULL == pCtg || NULL == dbFName || NULL == stbName) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan 已提交
2797 2798
  }

D
dapan1121 已提交
2799
  if (NULL == pCtg->dbCache) {
D
dapan1121 已提交
2800
    CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2801
  }
D
dapan1121 已提交
2802

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

D
dapan1121 已提交
2805
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
2806
  
D
dapan1121 已提交
2807 2808
_return:

D
dapan1121 已提交
2809
  CTG_API_LEAVE(code);
D
dapan 已提交
2810 2811
}

D
dapan1121 已提交
2812
int32_t catalogGetIndexMeta(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, const SName* pTableName, const char *pIndexName, SIndexMeta** pIndexMeta) {
2813
  return 0;
D
dapan1121 已提交
2814
}
D
dapan1121 已提交
2815

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

D
dapan1121 已提交
2819
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_UNKNOWN_STB));
D
dapan1121 已提交
2820
}
D
dapan1121 已提交
2821

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

D
dapan1121 已提交
2825
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_STB));
D
dapan1121 已提交
2826 2827
}

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

D
dapan1121 已提交
2843 2844
  strcpy(output->dbFName, rspMsg->dbFName);
  strcpy(output->tbName, rspMsg->tbName);
D
dapan1121 已提交
2845

D
dapan1121 已提交
2846
  output->dbId = rspMsg->dbId;
D
dapan1121 已提交
2847
  
D
dapan1121 已提交
2848
  SET_META_TYPE_TABLE(output->metaType);
D
dapan1121 已提交
2849
  
D
dapan1121 已提交
2850
  CTG_ERR_JRET(queryCreateTableMetaFromMsg(rspMsg, true, &output->tbMeta));
D
dapan1121 已提交
2851

D
dapan1121 已提交
2852
  CTG_ERR_JRET(ctgPushUpdateTblMsgInQueue(pCtg, output, false));
D
dapan 已提交
2853

D
dapan1121 已提交
2854 2855
  CTG_API_LEAVE(code);
  
D
dapan1121 已提交
2856 2857
_return:

wafwerar's avatar
wafwerar 已提交
2858 2859
  taosMemoryFreeClear(output->tbMeta);
  taosMemoryFreeClear(output);
D
dapan1121 已提交
2860
  
D
dapan1121 已提交
2861
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2862 2863
}

D
dapan1121 已提交
2864 2865 2866 2867 2868 2869 2870 2871 2872
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 已提交
2873

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

D
dapan1121 已提交
2877
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pTableName) {
D
dapan1121 已提交
2878 2879 2880
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2881
  CTG_API_LEAVE(ctgRefreshTblMeta(pCtg, pTrans, pMgmtEps, pTableName, CTG_FLAG_FORCE_UPDATE | CTG_FLAG_MAKE_STB(isSTable), NULL, true));
2882
}
2883

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

D
dapan1121 已提交
2887
  CTG_API_LEAVE(ctgGetTableMeta(pCtg, pTrans, pMgmtEps, pTableName, pTableMeta, CTG_FLAG_FORCE_UPDATE | CTG_FLAG_MAKE_STB(isSTable)));
D
dapan1121 已提交
2888 2889
}

D
dapan1121 已提交
2890
int32_t catalogGetTableDistVgInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const SName* pTableName, SArray** pVgList) {
D
dapan1121 已提交
2891
  CTG_API_ENTER();
D
dapan1121 已提交
2892 2893 2894 2895

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

D
dapan1121 已提交
2897
  if (CTG_IS_SYS_DBNAME(pTableName->dbname)) {
D
dapan1121 已提交
2898 2899 2900
    ctgError("no valid vgInfo for db, dbname:%s", pTableName->dbname);
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
2901

D
dapan1121 已提交
2902
  int32_t code = 0;
D
dapan1121 已提交
2903

D
dapan1121 已提交
2904 2905 2906 2907 2908
  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 已提交
2909

D
dapan1121 已提交
2910 2911 2912 2913 2914 2915
        char dbFName[TSDB_DB_FNAME_LEN] = {0};
        tNameGetFullDbName(pTableName, dbFName);        
        CTG_ERR_JRET(ctgRefreshDBVgInfo(pCtg, pRpc, pMgmtEps, dbFName));
        
        continue;
      }
D
dapan1121 已提交
2916
    }
D
dapan 已提交
2917

D
dapan1121 已提交
2918
    break;
D
dapan1121 已提交
2919
  }
D
dapan 已提交
2920

D
dapan1121 已提交
2921
_return:
D
dapan 已提交
2922

D
dapan1121 已提交
2923
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2924 2925 2926
}


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

D
dapan1121 已提交
2930
  if (CTG_IS_SYS_DBNAME(pTableName->dbname)) {
D
dapan1121 已提交
2931 2932 2933 2934
    ctgError("no valid vgInfo for db, dbname:%s", pTableName->dbname);
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2935 2936
  SCtgDBCache* dbCache = NULL;
  int32_t code = 0;
H
Haojun Liao 已提交
2937 2938
  char db[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pTableName, db);
D
dapan1121 已提交
2939

D
dapan1121 已提交
2940
  SDBVgInfo *vgInfo = NULL;
D
dapan1121 已提交
2941
  CTG_ERR_JRET(ctgGetDBVgInfo(pCtg, pTrans, pMgmtEps, db, &dbCache, &vgInfo));
D
dapan1121 已提交
2942

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

D
dapan1121 已提交
2945
_return:
D
dapan1121 已提交
2946

D
dapan1121 已提交
2947
  if (dbCache) {
D
dapan1121 已提交
2948 2949 2950 2951 2952 2953
    ctgReleaseVgInfo(dbCache);
    ctgReleaseDBCache(pCtg, dbCache);
  }

  if (vgInfo) {
    taosHashCleanup(vgInfo->vgHash);
wafwerar's avatar
wafwerar 已提交
2954
    taosMemoryFreeClear(vgInfo);
D
dapan1121 已提交
2955
  }
D
dapan1121 已提交
2956

D
dapan1121 已提交
2957
  CTG_API_LEAVE(code);
D
dapan1121 已提交
2958 2959 2960
}


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

D
dapan1121 已提交
2964
  if (NULL == pCtg || NULL == pTrans || NULL == pMgmtEps || NULL == pReq || NULL == pRsp) {
D
dapan1121 已提交
2965 2966 2967
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
2968
  int32_t code = 0;
D
dapan1121 已提交
2969
  pRsp->pTableMeta = NULL;
D
dapan1121 已提交
2970 2971 2972

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

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

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

D
dapan1121 已提交
2998
  if (pReq->qNodeRequired) {
D
dapan 已提交
2999 3000
    pRsp->pQnodeList = taosArrayInit(10, sizeof(SQueryNodeAddr));
    CTG_ERR_JRET(ctgGetQnodeListFromMnode(pCtg, pTrans, pMgmtEps, pRsp->pQnodeList));
D
dapan1121 已提交
3001 3002
  }

D
dapan1121 已提交
3003
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan1121 已提交
3004 3005

_return:  
D
dapan1121 已提交
3006

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

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

D
dapan 已提交
3029
  CTG_ERR_JRET(ctgGetQnodeListFromMnode(pCtg, pRpc, pMgmtEps, pQnodeList));
D
dapan1121 已提交
3030 3031

_return:
D
dapan 已提交
3032

D
dapan1121 已提交
3033
  CTG_API_LEAVE(TSDB_CODE_SUCCESS);
D
dapan 已提交
3034 3035
}

D
dapan1121 已提交
3036
int32_t catalogGetExpiredSTables(SCatalog* pCtg, SSTableMetaVersion **stables, uint32_t *num) {
D
dapan1121 已提交
3037 3038
  CTG_API_ENTER();

D
dapan1121 已提交
3039 3040
  if (NULL == pCtg || NULL == stables || NULL == num) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
D
dapan1121 已提交
3041 3042
  }

D
dapan1121 已提交
3043 3044 3045 3046
  CTG_API_LEAVE(ctgMetaRentGet(&pCtg->stbRent, (void **)stables, num, sizeof(SSTableMetaVersion)));
}

int32_t catalogGetExpiredDBs(SCatalog* pCtg, SDbVgVersion **dbs, uint32_t *num) {
D
dapan1121 已提交
3047
  CTG_API_ENTER();
D
dapan1121 已提交
3048 3049 3050 3051
  
  if (NULL == pCtg || NULL == dbs || NULL == num) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }
D
dapan1121 已提交
3052

D
dapan1121 已提交
3053
  CTG_API_LEAVE(ctgMetaRentGet(&pCtg->dbRent, (void **)dbs, num, sizeof(SDbVgVersion)));
D
dapan1121 已提交
3054 3055
}

D
dapan 已提交
3056 3057 3058 3059 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
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 已提交
3085 3086 3087 3088 3089 3090 3091 3092 3093
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 已提交
3094

D
dapan1121 已提交
3095 3096 3097 3098 3099 3100 3101 3102 3103 3104
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 已提交
3105
int32_t catalogGetUdfInfo(SCatalog* pCtg, void *pRpc, const SEpSet* pMgmtEps, const char* funcName, SFuncInfo** pInfo) {
D
dapan1121 已提交
3106 3107 3108 3109 3110 3111
  CTG_API_ENTER();
  
  if (NULL == pCtg || NULL == pRpc || NULL == pMgmtEps || NULL == funcName || NULL == pInfo) {
    CTG_API_LEAVE(TSDB_CODE_CTG_INVALID_INPUT);
  }

D
dapan1121 已提交
3112 3113 3114 3115 3116 3117 3118 3119 3120 3121 3122 3123 3124 3125 3126
  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 已提交
3127 3128
}

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

  int32_t code = 0;
D
dapan 已提交
3137
  CTG_ERR_JRET(ctgChkAuth(pCtg, pRpc, pMgmtEps, user, dbFName, type, pass));
D
dapan 已提交
3138 3139 3140 3141 3142 3143
  
_return:

  CTG_API_LEAVE(code);
}

D
dapan 已提交
3144 3145 3146 3147 3148 3149 3150 3151 3152 3153
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 已提交
3154

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

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

D
dapan1121 已提交
3164 3165 3166 3167 3168 3169 3170
  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 已提交
3171

D
dapan1121 已提交
3172
  while (CTG_IS_LOCKED(&gCtgMgmt.lock)) {
wafwerar's avatar
wafwerar 已提交
3173
    taosUsleep(1);
D
dapan1121 已提交
3174 3175
  }
  
D
dapan 已提交
3176
  CTG_LOCK(CTG_WRITE, &gCtgMgmt.lock);
D
dapan1121 已提交
3177

D
dapan1121 已提交
3178
  SCatalog *pCtg = NULL;
D
dapan 已提交
3179
  void *pIter = taosHashIterate(gCtgMgmt.pCluster, NULL);
D
dapan1121 已提交
3180
  while (pIter) {
D
dapan1121 已提交
3181
    pCtg = *(SCatalog **)pIter;
D
dapan1121 已提交
3182

D
dapan1121 已提交
3183 3184
    if (pCtg) {
      catalogFreeHandle(pCtg);
D
dapan1121 已提交
3185 3186
    }
    
D
dapan 已提交
3187
    pIter = taosHashIterate(gCtgMgmt.pCluster, pIter);
D
dapan 已提交
3188
  }
D
dapan1121 已提交
3189
  
D
dapan 已提交
3190 3191
  taosHashCleanup(gCtgMgmt.pCluster);
  gCtgMgmt.pCluster = NULL;
D
dapan1121 已提交
3192

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

D
dapan1121 已提交
3195
  qInfo("catalog destroyed");
D
dapan 已提交
3196 3197 3198 3199
}